freenet 0.2.47

use std::collections::VecDeque;

use chacha20poly1305::{XChaCha20Poly1305, XNonce};
use freenet_stdlib::prelude::{
    ApplicationMessage, DelegateContainer, DelegateContext, DelegateError, DelegateInterfaceResult,
    DelegateKey, DelegateMessage, GetContractRequest, InboundDelegateMsg, MessageOrigin,
    OutboundDelegateMsg, Parameters, PutContractRequest, SecretsId, SubscribeContractRequest,
    UpdateContractRequest,
};

use super::engine::{InstanceHandle, WasmEngine};
use super::native_api::{CURRENT_DELEGATE_INSTANCE, DELEGATE_ENV, DelegateCallEnv, InstanceId};
use super::{Runtime, RuntimeResult};
use crate::wasm_runtime::delegate_api::DelegateApiVersion;

/// RAII guard that ensures cleanup of delegate environment state.
/// When dropped, it clears the thread-local instance ID and removes the
/// entry from the global DELEGATE_ENV map.
struct DelegateEnvGuard {
    instance_id: InstanceId,
}

impl DelegateEnvGuard {
    fn new(instance_id: InstanceId) -> Self {
        Self { instance_id }
    }
}

impl Drop for DelegateEnvGuard {
    fn drop(&mut self) {
        // Clear thread-local first, then remove from global map
        CURRENT_DELEGATE_INSTANCE.with(|c| c.set(-1));
        DELEGATE_ENV.remove(&self.instance_id);
    }
}

#[derive(thiserror::Error, Debug)]
#[non_exhaustive]
pub enum DelegateExecError {
    #[error(transparent)]
    DelegateError(#[from] DelegateError),

    #[error("Permission denied: secret {secret} cannot be accesed by {delegate} at this time")]
    UnauthorizedSecretAccess {
        secret: SecretsId,
        delegate: DelegateKey,
    },

    #[error("Received an unexpected message from the client apps: {0}")]
    UnexpectedMessage(&'static str),
}

pub(crate) trait DelegateRuntimeInterface {
    fn inbound_app_message(
        &mut self,
        key: &DelegateKey,
        params: &Parameters,
        origin: Option<&MessageOrigin>,
        inbound: Vec<InboundDelegateMsg>,
    ) -> RuntimeResult<Vec<OutboundDelegateMsg>>;

    fn register_delegate(
        &mut self,
        delegate: DelegateContainer,
        cipher: XChaCha20Poly1305,
        nonce: XNonce,
    ) -> RuntimeResult<()>;

    fn unregister_delegate(&mut self, key: &DelegateKey) -> RuntimeResult<()>;
}

impl Runtime {
    /// Execute the delegate's `process` function with the DelegateCallEnv set up
    /// so that host functions for context and secret access are available.
    ///
    /// Uses RAII guard pattern to ensure cleanup happens even if WASM execution panics.
    #[allow(clippy::too_many_arguments)]
    fn exec_inbound_with_env(
        &mut self,
        delegate_key: &DelegateKey,
        params: &Parameters<'_>,
        origin: Option<&MessageOrigin>,
        msg: &InboundDelegateMsg,
        context: Vec<u8>,
        handle: &InstanceHandle,
        instance_id: i64,
        api_version: DelegateApiVersion,
    ) -> RuntimeResult<(Vec<OutboundDelegateMsg>, Vec<u8>)> {
        // Set up the delegate call environment with context, secret store, and
        // contract store access.
        // SAFETY: `self.secret_store` and `self.contract_store` are valid for the
        // duration of the WASM `process()` call below, and the `DelegateEnvGuard`
        // ensures the env is removed from `DELEGATE_ENV` before this function returns.
        // Build the origin_contracts list from the MessageOrigin. Only WebApp
        // attestations grant the receiving delegate access to a contract on
        // behalf of the caller; an inter-delegate caller (Delegate variant)
        // does not propagate contract access — its identity is conveyed only
        // via the `origin` argument forwarded into the WASM `process()` call.
        let origin_contracts = match origin {
            Some(MessageOrigin::WebApp(contract_id)) => vec![*contract_id],
            Some(MessageOrigin::Delegate(_)) | None => Vec::new(),
            // MessageOrigin is `#[non_exhaustive]`; future variants reach
            // this arm because the compiler requires it. Default to "no
            // contract access" (fail closed) AND log a warning so the gap
            // is visible during the PR that adds the new variant — the
            // catch-all should not silently default in production.
            Some(other) => {
                tracing::warn!(
                    delegate_key = %delegate_key,
                    origin = ?other,
                    "Unknown MessageOrigin variant reached fail-closed default; \
                     wasm_runtime::delegate::Runtime::inbound_app_message must \
                     decide explicitly whether this variant grants contract access"
                );
                Vec::new()
            }
        };

        // SAFETY: The `DelegateCallEnv` does not outlive `self`. The raw pointers to
        // `secret_store`, `contract_store`, and `delegate_store` remain valid for the
        // duration of the WASM `process()` call, and are cleaned up via DELEGATE_ENV
        // removal below.
        let env = unsafe {
            DelegateCallEnv::new(
                context,
                &mut self.secret_store,
                &self.contract_store,
                self.state_store_db.clone(),
                delegate_key.clone(),
                &mut self.delegate_store,
                0, // creation_depth: always 0 for top-level calls
                origin_contracts,
            )
        };

        debug_assert!(
            !DELEGATE_ENV.contains_key(&instance_id),
            "Instance ID {instance_id} already exists in DELEGATE_ENV - this indicates a bug"
        );

        DELEGATE_ENV.insert(instance_id, env);
        CURRENT_DELEGATE_INSTANCE.with(|c| c.set(instance_id));

        // Create RAII guard to ensure cleanup on all exit paths (including panic)
        let _guard = DelegateEnvGuard::new(instance_id);

        // Execute the WASM process function.
        // V2 delegates use call_async (async host functions for contract access).
        // V1 delegates use synchronous call.
        let result = self.exec_inbound(params, origin, msg, handle, api_version);

        // Read back the (possibly mutated) context before guard drops
        let updated_context = DELEGATE_ENV
            .get(&instance_id)
            .map(|env| env.context.clone())
            .unwrap_or_default();

        let outbound = result?;
        Ok((outbound, updated_context))
    }

    fn exec_inbound(
        &mut self,
        params: &Parameters<'_>,
        origin: Option<&MessageOrigin>,
        msg: &InboundDelegateMsg,
        handle: &InstanceHandle,
        api_version: DelegateApiVersion,
    ) -> RuntimeResult<Vec<OutboundDelegateMsg>> {
        let param_buf_ptr = {
            let mut param_buf = self.init_buf(handle, params)?;
            param_buf.write(params)?;
            param_buf.ptr()
        };
        let origin_buf_ptr = {
            let bytes = match origin {
                Some(o) => bincode::serialize(o)?,
                None => Vec::new(),
            };
            let mut origin_buf = self.init_buf(handle, &bytes)?;
            origin_buf.write(bytes)?;
            origin_buf.ptr()
        };
        let msg_ptr = {
            let msg = bincode::serialize(msg)?;
            let mut msg_buf = self.init_buf(handle, &msg)?;
            msg_buf.write(msg)?;
            msg_buf.ptr()
        };
        let inbound_msg_name = match msg {
            InboundDelegateMsg::ApplicationMessage(_) => "ApplicationMessage",
            InboundDelegateMsg::UserResponse(_) => "UserResponse",
            InboundDelegateMsg::GetContractResponse(_) => "GetContractResponse",
            InboundDelegateMsg::PutContractResponse(_) => "PutContractResponse",
            InboundDelegateMsg::UpdateContractResponse(_) => "UpdateContractResponse",
            InboundDelegateMsg::SubscribeContractResponse(_) => "SubscribeContractResponse",
            InboundDelegateMsg::ContractNotification(_) => "ContractNotification",
            InboundDelegateMsg::DelegateMessage(_) => "DelegateMessage",
            // `InboundDelegateMsg` is `#[non_exhaustive]` (stdlib 0.6.0+).
            // Future variants land here for tracing only — they still flow
            // through the wasm boundary as raw bincode below; classifying
            // them as "Unknown" affects logs only, not delivery.
            _ => "Unknown",
        };
        tracing::debug!(
            inbound_msg_name,
            api_version = %api_version,
            "Calling delegate with inbound message"
        );

        let res = match api_version {
            DelegateApiVersion::V1 => {
                // V1: synchronous call — no async host functions involved.
                // Must stay on calling thread for thread-local env.
                self.engine.call_3i64(
                    handle,
                    "process",
                    param_buf_ptr as i64,
                    origin_buf_ptr as i64,
                    msg_ptr as i64,
                )?
            }
            DelegateApiVersion::V2 => {
                // V2: async call — contract host functions are async.
                // Uses Store::into_async() + call_async() under the hood.
                self.engine.call_3i64_async_imports(
                    handle,
                    "process",
                    param_buf_ptr as i64,
                    origin_buf_ptr as i64,
                    msg_ptr as i64,
                )?
            }
        };

        let linear_mem = self.linear_mem(handle)?;
        // SAFETY: `res` is the return value from the WASM `process` call and
        // `linear_mem` points to the instance's live linear memory, so `from_raw`
        // reads a valid, in-bounds result descriptor.
        let outbound = unsafe {
            DelegateInterfaceResult::from_raw(res, &linear_mem)
                .unwrap(linear_mem)
                .map_err(Into::<DelegateExecError>::into)?
        };
        self.log_delegate_exec_result(inbound_msg_name, &outbound);
        Ok(outbound)
    }

    fn log_delegate_exec_result(&self, inbound_msg_name: &str, outbound: &[OutboundDelegateMsg]) {
        if tracing::enabled!(tracing::Level::DEBUG) {
            let outbound_message_names = outbound
                .iter()
                .map(|m| match m {
                    OutboundDelegateMsg::ApplicationMessage(am) => format!(
                        "ApplicationMessage(payload_len={}, processed={}, context_len={})",
                        am.payload.len(),
                        am.processed,
                        am.context.as_ref().len()
                    ),
                    OutboundDelegateMsg::RequestUserInput(_) => "RequestUserInput".to_string(),
                    OutboundDelegateMsg::ContextUpdated(_) => "ContextUpdated".to_string(),
                    OutboundDelegateMsg::GetContractRequest(req) => {
                        format!("GetContractRequest(contract={})", req.contract_id)
                    }
                    OutboundDelegateMsg::PutContractRequest(req) => {
                        format!("PutContractRequest(contract={})", req.contract.key())
                    }
                    OutboundDelegateMsg::UpdateContractRequest(req) => {
                        format!("UpdateContractRequest(contract={})", req.contract_id)
                    }
                    OutboundDelegateMsg::SubscribeContractRequest(req) => {
                        format!("SubscribeContractRequest(contract={})", req.contract_id)
                    }
                    OutboundDelegateMsg::SendDelegateMessage(msg) => {
                        format!(
                            "SendDelegateMessage(target={}, payload_len={})",
                            msg.target,
                            msg.payload.len()
                        )
                    }
                })
                .collect::<Vec<String>>()
                .join(", ");
            tracing::debug!(
                inbound_msg_name,
                outbound_message_names,
                "Delegate returned outbound messages"
            );
        } else {
            tracing::debug!(
                inbound_msg_name,
                outbound_len = outbound.len(),
                "Delegate returned outbound messages"
            );
        }
    }

    fn log_process_outbound_entry(
        &self,
        delegate_key: &DelegateKey,
        origin: Option<&MessageOrigin>,
        outbound_msgs: &VecDeque<OutboundDelegateMsg>,
    ) {
        tracing::debug!(
            delegate_key = ?delegate_key,
            ?origin,
            outbound_msgs_len = outbound_msgs.len(),
            outbound_msg_details = debug(if tracing::enabled!(tracing::Level::DEBUG) {
                outbound_msgs.iter().map(|msg| {
                    match msg {
                        OutboundDelegateMsg::ApplicationMessage(m) => format!("AppMsg(payload_len={})", m.payload.len()),
                        OutboundDelegateMsg::RequestUserInput(_) => "UserInputReq".to_string(),
                        OutboundDelegateMsg::ContextUpdated(_) => "ContextUpdate".to_string(),
                        OutboundDelegateMsg::GetContractRequest(r) => format!("GetContractReq({})", r.contract_id),
                        OutboundDelegateMsg::PutContractRequest(r) => format!("PutContractReq({})", r.contract.key()),
                        OutboundDelegateMsg::UpdateContractRequest(r) => format!("UpdateContractReq({})", r.contract_id),
                        OutboundDelegateMsg::SubscribeContractRequest(r) => format!("SubscribeContractReq({})", r.contract_id),
                        OutboundDelegateMsg::SendDelegateMessage(m) => format!("SendDelegateMsg(target={})", m.target),
                    }
                }).collect::<Vec<_>>()
            } else {
                Vec::new()
            }),
            "process_outbound called"
        );
    }

    /// Process outbound messages from a delegate.
    #[allow(clippy::too_many_arguments)]
    fn process_outbound(
        &mut self,
        delegate_key: &DelegateKey,
        _handle: &InstanceHandle,
        _instance_id: i64,
        _params: &Parameters<'_>,
        origin: Option<&MessageOrigin>,
        outbound_msgs: &mut VecDeque<OutboundDelegateMsg>,
        context: &mut Vec<u8>,
        results: &mut Vec<OutboundDelegateMsg>,
    ) -> RuntimeResult<()> {
        self.log_process_outbound_entry(delegate_key, origin, outbound_msgs);

        while let Some(outbound) = outbound_msgs.pop_front() {
            match outbound {
                OutboundDelegateMsg::ApplicationMessage(mut msg) => {
                    tracing::debug!(
                        payload_len = msg.payload.len(),
                        processed = msg.processed,
                        "Adding ApplicationMessage to results"
                    );
                    msg.context = DelegateContext::default();
                    results.push(OutboundDelegateMsg::ApplicationMessage(msg));
                    for remaining in outbound_msgs.drain(..) {
                        results.push(remaining);
                    }
                    break;
                }

                OutboundDelegateMsg::RequestUserInput(req) => {
                    tracing::debug!(
                        request_id = req.request_id,
                        "Passing RequestUserInput to executor for user prompting"
                    );
                    results.push(OutboundDelegateMsg::RequestUserInput(req));
                    for remaining in outbound_msgs.drain(..) {
                        results.push(remaining);
                    }
                    break;
                }

                OutboundDelegateMsg::ContextUpdated(new_context) => {
                    *context = new_context.as_ref().to_vec();
                }
                OutboundDelegateMsg::GetContractRequest(req) if !req.processed => {
                    tracing::debug!(
                        contract_id = %req.contract_id,
                        "Passing GetContractRequest to executor for async handling"
                    );
                    results.push(OutboundDelegateMsg::GetContractRequest(req));
                    for remaining in outbound_msgs.drain(..) {
                        results.push(remaining);
                    }
                    break;
                }
                OutboundDelegateMsg::GetContractRequest(GetContractRequest {
                    context: ctx,
                    ..
                }) => {
                    tracing::debug!("GetContractRequest processed");
                    *context = ctx.as_ref().to_vec();
                }
                OutboundDelegateMsg::PutContractRequest(req) if !req.processed => {
                    tracing::debug!(
                        contract = %req.contract.key(),
                        "Passing PutContractRequest to executor for async handling"
                    );
                    results.push(OutboundDelegateMsg::PutContractRequest(req));
                    for remaining in outbound_msgs.drain(..) {
                        results.push(remaining);
                    }
                    break;
                }
                OutboundDelegateMsg::PutContractRequest(PutContractRequest {
                    context: ctx,
                    ..
                }) => {
                    tracing::debug!("PutContractRequest processed");
                    *context = ctx.as_ref().to_vec();
                }
                OutboundDelegateMsg::UpdateContractRequest(req) if !req.processed => {
                    tracing::debug!(
                        contract_id = %req.contract_id,
                        "Passing UpdateContractRequest to executor for async handling"
                    );
                    results.push(OutboundDelegateMsg::UpdateContractRequest(req));
                    for remaining in outbound_msgs.drain(..) {
                        results.push(remaining);
                    }
                    break;
                }
                OutboundDelegateMsg::UpdateContractRequest(UpdateContractRequest {
                    context: ctx,
                    ..
                }) => {
                    tracing::debug!("UpdateContractRequest processed");
                    *context = ctx.as_ref().to_vec();
                }
                OutboundDelegateMsg::SubscribeContractRequest(req) if !req.processed => {
                    tracing::debug!(
                        contract_id = %req.contract_id,
                        "Passing SubscribeContractRequest to executor for async handling"
                    );
                    results.push(OutboundDelegateMsg::SubscribeContractRequest(req));
                    for remaining in outbound_msgs.drain(..) {
                        results.push(remaining);
                    }
                    break;
                }
                OutboundDelegateMsg::SubscribeContractRequest(SubscribeContractRequest {
                    context: ctx,
                    ..
                }) => {
                    tracing::debug!("SubscribeContractRequest processed");
                    *context = ctx.as_ref().to_vec();
                }
                OutboundDelegateMsg::SendDelegateMessage(mut msg) if !msg.processed => {
                    tracing::debug!(
                        target_delegate = %msg.target,
                        "Passing SendDelegateMessage to executor for delivery"
                    );
                    // Sender attestation: overwrite sender with the actual delegate key
                    msg.sender = delegate_key.clone();
                    results.push(OutboundDelegateMsg::SendDelegateMessage(msg));
                    // Attest any remaining SendDelegateMessage variants to prevent
                    // spoofing via drain bypass (see PR #3282 review).
                    for remaining in outbound_msgs.drain(..) {
                        match remaining {
                            OutboundDelegateMsg::SendDelegateMessage(mut m) if !m.processed => {
                                m.sender = delegate_key.clone();
                                results.push(OutboundDelegateMsg::SendDelegateMessage(m));
                            }
                            msg @ (OutboundDelegateMsg::ApplicationMessage(_)
                            | OutboundDelegateMsg::RequestUserInput(_)
                            | OutboundDelegateMsg::ContextUpdated(_)
                            | OutboundDelegateMsg::GetContractRequest(_)
                            | OutboundDelegateMsg::PutContractRequest(_)
                            | OutboundDelegateMsg::UpdateContractRequest(_)
                            | OutboundDelegateMsg::SubscribeContractRequest(_)
                            | OutboundDelegateMsg::SendDelegateMessage(_)) => results.push(msg),
                        }
                    }
                    break;
                }
                OutboundDelegateMsg::SendDelegateMessage(DelegateMessage {
                    context: ctx, ..
                }) => {
                    tracing::debug!("SendDelegateMessage processed");
                    *context = ctx.as_ref().to_vec();
                }
            }
        }
        Ok(())
    }
}

impl DelegateRuntimeInterface for Runtime {
    fn inbound_app_message(
        &mut self,
        delegate_key: &DelegateKey,
        params: &Parameters,
        origin: Option<&MessageOrigin>,
        inbound: Vec<InboundDelegateMsg>,
    ) -> RuntimeResult<Vec<OutboundDelegateMsg>> {
        let mut results = Vec::with_capacity(inbound.len());
        if inbound.is_empty() {
            return Ok(results);
        }
        let (mut running, api_version) = self.prepare_delegate_call(params, delegate_key, 4096)?;
        let instance_id = running.id;

        tracing::debug!(
            delegate_key = %delegate_key,
            api_version = %api_version,
            "Starting delegate execution"
        );

        // State maintained across process() calls within this conversation
        let mut context: Vec<u8> = Vec::new();

        // Process all messages, collecting the result.
        // Cleanup happens after the loop regardless of success/failure.
        let process_result: RuntimeResult<()> = (|| {
            for msg in inbound {
                // The wildcard arm at the bottom of this match exists
                // solely because `InboundDelegateMsg` is `#[non_exhaustive]`
                // (stdlib 0.6.0+); every currently-known variant is
                // enumerated above. Re-listing them in a `pat | _` shape
                // (as `wildcard_enum_match_arm` would prefer) is needless
                // duplication that defeats the safety net the wildcard
                // provides for future variants.
                #[allow(clippy::wildcard_enum_match_arm)]
                match msg {
                    InboundDelegateMsg::ApplicationMessage(ApplicationMessage {
                        payload,
                        processed,
                        ..
                    }) => {
                        let app_msg = InboundDelegateMsg::ApplicationMessage(
                            ApplicationMessage::new(payload)
                                .processed(processed)
                                .with_context(DelegateContext::new(context.clone())),
                        );

                        let (outbound, updated_context) = self.exec_inbound_with_env(
                            delegate_key,
                            params,
                            origin,
                            &app_msg,
                            context.clone(),
                            &running.handle,
                            instance_id,
                            api_version,
                        )?;
                        context = updated_context;

                        let mut outbound_queue = VecDeque::from(outbound);
                        self.process_outbound(
                            delegate_key,
                            &running.handle,
                            instance_id,
                            params,
                            origin,
                            &mut outbound_queue,
                            &mut context,
                            &mut results,
                        )?;
                    }
                    InboundDelegateMsg::UserResponse(response) => {
                        let (outbound, updated_context) = self.exec_inbound_with_env(
                            delegate_key,
                            params,
                            origin,
                            &InboundDelegateMsg::UserResponse(response),
                            context.clone(),
                            &running.handle,
                            instance_id,
                            api_version,
                        )?;
                        context = updated_context;

                        let mut outbound_queue = VecDeque::from(outbound);
                        self.process_outbound(
                            delegate_key,
                            &running.handle,
                            instance_id,
                            params,
                            origin,
                            &mut outbound_queue,
                            &mut context,
                            &mut results,
                        )?;
                    }
                    InboundDelegateMsg::GetContractResponse(response) => {
                        let (outbound, updated_context) = self.exec_inbound_with_env(
                            delegate_key,
                            params,
                            origin,
                            &InboundDelegateMsg::GetContractResponse(response),
                            context.clone(),
                            &running.handle,
                            instance_id,
                            api_version,
                        )?;
                        context = updated_context;

                        let mut outbound_queue = VecDeque::from(outbound);
                        self.process_outbound(
                            delegate_key,
                            &running.handle,
                            instance_id,
                            params,
                            origin,
                            &mut outbound_queue,
                            &mut context,
                            &mut results,
                        )?;
                    }
                    msg @ (InboundDelegateMsg::PutContractResponse(_)
                    | InboundDelegateMsg::UpdateContractResponse(_)
                    | InboundDelegateMsg::SubscribeContractResponse(_)
                    | InboundDelegateMsg::ContractNotification(_)
                    | InboundDelegateMsg::DelegateMessage(_)) => {
                        let (outbound, updated_context) = self.exec_inbound_with_env(
                            delegate_key,
                            params,
                            origin,
                            &msg,
                            context.clone(),
                            &running.handle,
                            instance_id,
                            api_version,
                        )?;
                        context = updated_context;

                        let mut outbound_queue = VecDeque::from(outbound);
                        self.process_outbound(
                            delegate_key,
                            &running.handle,
                            instance_id,
                            params,
                            origin,
                            &mut outbound_queue,
                            &mut context,
                            &mut results,
                        )?;
                    }
                    // `InboundDelegateMsg` is `#[non_exhaustive]` (stdlib
                    // 0.6.0+). Future variants are forwarded to the WASM
                    // through the same generic exec path so a delegate
                    // built against a newer stdlib can handle them; the
                    // host neither inspects nor classifies their payload.
                    other => {
                        let (outbound, updated_context) = self.exec_inbound_with_env(
                            delegate_key,
                            params,
                            origin,
                            &other,
                            context.clone(),
                            &running.handle,
                            instance_id,
                            api_version,
                        )?;
                        context = updated_context;

                        let mut outbound_queue = VecDeque::from(outbound);
                        self.process_outbound(
                            delegate_key,
                            &running.handle,
                            instance_id,
                            params,
                            origin,
                            &mut outbound_queue,
                            &mut context,
                            &mut results,
                        )?;
                    }
                }
            }
            Ok(())
        })();

        // Always clean up the WASM Instance, even on error.
        self.drop_running_instance(&mut running);

        process_result?;

        tracing::debug!(
            count = results.len(),
            "Final results returned by inbound_app_message"
        );
        Ok(results)
    }

    #[inline]
    fn register_delegate(
        &mut self,
        delegate: DelegateContainer,
        cipher: XChaCha20Poly1305,
        nonce: XNonce,
    ) -> RuntimeResult<()> {
        self.secret_store
            .register_delegate(delegate.key().clone(), cipher, nonce)?;
        self.delegate_store.store_delegate(delegate)
    }

    #[inline]
    fn unregister_delegate(&mut self, key: &DelegateKey) -> RuntimeResult<()> {
        self.delegate_modules.lock().unwrap().pop(key);
        self.delegate_store.remove_delegate(key)
    }
}

#[cfg(all(test, feature = "wasmtime-backend"))]
mod test {
    use chacha20poly1305::aead::{AeadCore, KeyInit, OsRng};
    use freenet_stdlib::prelude::*;
    use serde::{Deserialize, Serialize};
    use std::sync::Arc;

    use std::os::unix::fs::PermissionsExt;

    use crate::util::tests::get_temp_dir;

    use super::super::{ContractStore, SecretsStore, delegate_store::DelegateStore};
    use super::*;

    const TEST_DELEGATE_2: &str = "test_delegate_2";

    /// Message types for test-delegate-2 (host function API)
    mod delegate2_messages {
        use super::*;

        #[derive(Debug, Serialize, Deserialize)]
        pub enum InboundAppMessage {
            CreateInboxRequest,
            PleaseSignMessage(Vec<u8>),
            WriteContext(Vec<u8>),
            ReadContext,
            ClearContext,
            IncrementCounter,
            HasSecret(Vec<u8>),
            GetNonExistentSecret(Vec<u8>),
            StoreSecret { key: Vec<u8>, value: Vec<u8> },
            RemoveSecret(Vec<u8>),
            WriteLargeContext(usize),
            StoreLargeSecret { key: Vec<u8>, size: usize },
        }

        #[derive(Debug, Serialize, Deserialize)]
        pub enum OutboundAppMessage {
            CreateInboxResponse(Vec<u8>),
            MessageSigned(Vec<u8>),
            ContextData(Vec<u8>),
            CounterValue(u32),
            SecretExists(bool),
            SecretResult(Option<Vec<u8>>),
            ContextWritten,
            ContextCleared,
            SecretStored,
            SecretRemoved,
            LargeContextWritten(usize),
            LargeSecretStored(usize),
            SecretStoreFailed,
        }
    }

    async fn setup_runtime(
        name: &str,
    ) -> Result<(DelegateContainer, Runtime, tempfile::TempDir), Box<dyn std::error::Error>> {
        use crate::contract::storages::Storage;
        let temp_dir = get_temp_dir();
        let contracts_dir = temp_dir.path().join("contracts");
        let delegates_dir = temp_dir.path().join("delegates");
        let secrets_dir = temp_dir.path().join("secrets");

        let db = Storage::new(temp_dir.path()).await?;
        let contract_store = ContractStore::new(contracts_dir, 10_000, db.clone())?;
        let delegate_store = DelegateStore::new(delegates_dir, 10_000, db.clone())?;
        let secret_store = SecretsStore::new(secrets_dir, Default::default(), db)?;

        let mut runtime =
            Runtime::build(contract_store, delegate_store, secret_store, false).unwrap();

        let delegate = {
            let bytes = super::super::tests::get_test_module(name)?;
            DelegateContainer::Wasm(DelegateWasmAPIVersion::V1(Delegate::from((
                &bytes.into(),
                &vec![].into(),
            ))))
        };
        let _stored = runtime.delegate_store.store_delegate(delegate.clone());

        let key = XChaCha20Poly1305::generate_key(&mut OsRng);
        let cipher = XChaCha20Poly1305::new(&key);
        let nonce = XChaCha20Poly1305::generate_nonce(&mut OsRng);
        let _registered =
            runtime
                .secret_store
                .register_delegate(delegate.key().clone(), cipher, nonce);

        Ok((delegate, runtime, temp_dir))
    }

    const TEST_DELEGATE_CAPABILITIES: &str = "test_delegate_capabilities";

    /// Message types for test-delegate-capabilities (must match the delegate's types)
    mod capabilities_messages {
        use super::*;

        #[derive(Debug, Serialize, Deserialize)]
        #[allow(clippy::enum_variant_names)]
        pub enum DelegateCommand {
            GetContractState {
                contract_id: ContractInstanceId,
            },
            GetMultipleContractStates {
                contract_ids: Vec<ContractInstanceId>,
            },
            GetContractWithEcho {
                contract_id: ContractInstanceId,
                echo_message: String,
            },
            PutContractState {
                contract: ContractContainer,
                state: Vec<u8>,
            },
            UpdateContractState {
                contract_id: ContractInstanceId,
                state: Vec<u8>,
            },
            SubscribeContract {
                contract_id: ContractInstanceId,
            },
        }

        #[derive(Debug, Serialize, Deserialize)]
        pub enum DelegateResponse {
            ContractState {
                contract_id: ContractInstanceId,
                state: Option<Vec<u8>>,
            },
            MultipleContractStates {
                results: Vec<(ContractInstanceId, Option<Vec<u8>>)>,
            },
            Echo {
                message: String,
            },
            ContractPutResult {
                contract_id: ContractInstanceId,
                success: bool,
                error: Option<String>,
            },
            ContractUpdateResult {
                contract_id: ContractInstanceId,
                success: bool,
                error: Option<String>,
            },
            ContractSubscribeResult {
                contract_id: ContractInstanceId,
                success: bool,
                error: Option<String>,
            },
            ContractNotificationReceived {
                contract_id: ContractInstanceId,
                new_state: Vec<u8>,
            },
            Error {
                message: String,
            },
        }
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_get_contract_request_response() -> Result<(), Box<dyn std::error::Error>> {
        use capabilities_messages::*;

        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_CAPABILITIES).await?;
        let target_contract_id = ContractInstanceId::new([42u8; 32]);
        let _app_id = ContractInstanceId::new([1u8; 32]);

        let command = DelegateCommand::GetContractState {
            contract_id: target_contract_id,
        };
        let payload = bincode::serialize(&command)?;
        let app_msg = ApplicationMessage::new(payload);

        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(app_msg)],
        )?;

        assert_eq!(outbound.len(), 1, "Expected exactly one outbound message");
        let contract_request = match &outbound[0] {
            OutboundDelegateMsg::GetContractRequest(req) => req.clone(),
            other @ OutboundDelegateMsg::ApplicationMessage(_)
            | other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected GetContractRequest, got {:?}", other)
            }
        };
        assert_eq!(contract_request.contract_id, target_contract_id);
        assert!(!contract_request.processed);

        let contract_state = vec![1, 2, 3, 4, 5];
        let response = InboundDelegateMsg::GetContractResponse(GetContractResponse {
            contract_id: target_contract_id,
            state: Some(WrappedState::new(contract_state.clone())),
            context: contract_request.context.clone(),
        });

        let final_outbound =
            runtime.inbound_app_message(delegate.key(), &vec![].into(), None, vec![response])?;

        assert_eq!(final_outbound.len(), 1);
        let final_msg = match &final_outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => msg,
            other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage, got {:?}", other)
            }
        };
        assert!(final_msg.processed);

        let response: DelegateResponse = bincode::deserialize(&final_msg.payload)?;
        match response {
            DelegateResponse::ContractState { contract_id, state } => {
                assert_eq!(contract_id, target_contract_id);
                assert_eq!(state, Some(contract_state));
            }
            other @ DelegateResponse::MultipleContractStates { .. }
            | other @ DelegateResponse::Echo { .. }
            | other @ DelegateResponse::ContractPutResult { .. }
            | other @ DelegateResponse::ContractUpdateResult { .. }
            | other @ DelegateResponse::ContractSubscribeResult { .. }
            | other @ DelegateResponse::ContractNotificationReceived { .. }
            | other @ DelegateResponse::Error { .. } => {
                panic!("Expected ContractState response, got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_get_contract_not_found() -> Result<(), Box<dyn std::error::Error>> {
        use capabilities_messages::*;

        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_CAPABILITIES).await?;
        let target_contract_id = ContractInstanceId::new([99u8; 32]);
        let _app_id = ContractInstanceId::new([1u8; 32]);

        let command = DelegateCommand::GetContractState {
            contract_id: target_contract_id,
        };
        let payload = bincode::serialize(&command)?;
        let app_msg = ApplicationMessage::new(payload);

        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(app_msg)],
        )?;

        let contract_request = match &outbound[0] {
            OutboundDelegateMsg::GetContractRequest(req) => req.clone(),
            other @ OutboundDelegateMsg::ApplicationMessage(_)
            | other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected GetContractRequest, got {:?}", other)
            }
        };

        let response = InboundDelegateMsg::GetContractResponse(GetContractResponse {
            contract_id: target_contract_id,
            state: None,
            context: contract_request.context.clone(),
        });

        let final_outbound =
            runtime.inbound_app_message(delegate.key(), &vec![].into(), None, vec![response])?;

        let final_msg = match &final_outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => msg,
            other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage, got {:?}", other)
            }
        };

        let response: DelegateResponse = bincode::deserialize(&final_msg.payload)?;
        match response {
            DelegateResponse::ContractState { state, .. } => {
                assert!(state.is_none());
            }
            other @ DelegateResponse::MultipleContractStates { .. }
            | other @ DelegateResponse::Echo { .. }
            | other @ DelegateResponse::ContractPutResult { .. }
            | other @ DelegateResponse::ContractUpdateResult { .. }
            | other @ DelegateResponse::ContractSubscribeResult { .. }
            | other @ DelegateResponse::ContractNotificationReceived { .. }
            | other @ DelegateResponse::Error { .. } => {
                panic!("Expected ContractState response, got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_multiple_contract_requests() -> Result<(), Box<dyn std::error::Error>> {
        use capabilities_messages::*;

        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_CAPABILITIES).await?;

        let contract1 = ContractInstanceId::new([1u8; 32]);
        let contract2 = ContractInstanceId::new([2u8; 32]);
        let contract3 = ContractInstanceId::new([3u8; 32]);
        let _app_id = ContractInstanceId::new([10u8; 32]);

        let command = DelegateCommand::GetMultipleContractStates {
            contract_ids: vec![contract1, contract2, contract3],
        };
        let payload = bincode::serialize(&command)?;
        let app_msg = ApplicationMessage::new(payload);

        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(app_msg)],
        )?;

        assert_eq!(outbound.len(), 1);
        let req1 = match &outbound[0] {
            OutboundDelegateMsg::GetContractRequest(req) => req.clone(),
            other @ OutboundDelegateMsg::ApplicationMessage(_)
            | other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected GetContractRequest, got {:?}", other)
            }
        };
        assert_eq!(req1.contract_id, contract1);

        let response1 = InboundDelegateMsg::GetContractResponse(GetContractResponse {
            contract_id: contract1,
            state: Some(WrappedState::new(vec![1, 1, 1])),
            context: req1.context,
        });

        let outbound2 =
            runtime.inbound_app_message(delegate.key(), &vec![].into(), None, vec![response1])?;

        assert_eq!(outbound2.len(), 1);
        let req2 = match &outbound2[0] {
            OutboundDelegateMsg::GetContractRequest(req) => req.clone(),
            other @ OutboundDelegateMsg::ApplicationMessage(_)
            | other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected GetContractRequest for contract2, got {:?}", other)
            }
        };
        assert_eq!(req2.contract_id, contract2);

        let response2 = InboundDelegateMsg::GetContractResponse(GetContractResponse {
            contract_id: contract2,
            state: Some(WrappedState::new(vec![2, 2, 2])),
            context: req2.context,
        });

        let outbound3 =
            runtime.inbound_app_message(delegate.key(), &vec![].into(), None, vec![response2])?;

        assert_eq!(outbound3.len(), 1);
        let req3 = match &outbound3[0] {
            OutboundDelegateMsg::GetContractRequest(req) => req.clone(),
            other @ OutboundDelegateMsg::ApplicationMessage(_)
            | other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected GetContractRequest for contract3, got {:?}", other)
            }
        };
        assert_eq!(req3.contract_id, contract3);

        let response3 = InboundDelegateMsg::GetContractResponse(GetContractResponse {
            contract_id: contract3,
            state: Some(WrappedState::new(vec![3, 3, 3])),
            context: req3.context,
        });

        let final_outbound =
            runtime.inbound_app_message(delegate.key(), &vec![].into(), None, vec![response3])?;

        assert_eq!(final_outbound.len(), 1);
        let final_msg = match &final_outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => msg,
            other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage, got {:?}", other)
            }
        };
        assert!(final_msg.processed);

        let response: DelegateResponse = bincode::deserialize(&final_msg.payload)?;
        match response {
            DelegateResponse::MultipleContractStates { results } => {
                assert_eq!(results.len(), 3);
                assert_eq!(results[0].0, contract1);
                assert_eq!(results[0].1, Some(vec![1, 1, 1]));
                assert_eq!(results[1].0, contract2);
                assert_eq!(results[1].1, Some(vec![2, 2, 2]));
                assert_eq!(results[2].0, contract3);
                assert_eq!(results[2].1, Some(vec![3, 3, 3]));
            }
            other @ DelegateResponse::ContractState { .. }
            | other @ DelegateResponse::Echo { .. }
            | other @ DelegateResponse::ContractPutResult { .. }
            | other @ DelegateResponse::ContractUpdateResult { .. }
            | other @ DelegateResponse::ContractSubscribeResult { .. }
            | other @ DelegateResponse::ContractNotificationReceived { .. }
            | other @ DelegateResponse::Error { .. } => {
                panic!("Expected MultipleContractStates, got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_message_accumulation() -> Result<(), Box<dyn std::error::Error>> {
        use capabilities_messages::*;

        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_CAPABILITIES).await?;

        let contract_id = ContractInstanceId::new([42u8; 32]);
        let _app_id = ContractInstanceId::new([1u8; 32]);
        let echo_message = "Hello from test!".to_string();

        let command = DelegateCommand::GetContractWithEcho {
            contract_id,
            echo_message: echo_message.clone(),
        };
        let payload = bincode::serialize(&command)?;
        let app_msg = ApplicationMessage::new(payload);

        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(app_msg)],
        )?;

        assert_eq!(outbound.len(), 2);

        let contract_request = outbound
            .iter()
            .find_map(|msg| match msg {
                OutboundDelegateMsg::GetContractRequest(req) => Some(req.clone()),
                OutboundDelegateMsg::ApplicationMessage(_)
                | OutboundDelegateMsg::RequestUserInput(_)
                | OutboundDelegateMsg::ContextUpdated(_)
                | OutboundDelegateMsg::PutContractRequest(_)
                | OutboundDelegateMsg::UpdateContractRequest(_)
                | OutboundDelegateMsg::SubscribeContractRequest(_)
                | OutboundDelegateMsg::SendDelegateMessage(_) => None,
            })
            .expect("Expected a GetContractRequest");
        assert_eq!(contract_request.contract_id, contract_id);

        let echo_msg = outbound
            .iter()
            .find_map(|msg| match msg {
                OutboundDelegateMsg::ApplicationMessage(m) => Some(m.clone()),
                OutboundDelegateMsg::RequestUserInput(_)
                | OutboundDelegateMsg::ContextUpdated(_)
                | OutboundDelegateMsg::GetContractRequest(_)
                | OutboundDelegateMsg::PutContractRequest(_)
                | OutboundDelegateMsg::UpdateContractRequest(_)
                | OutboundDelegateMsg::SubscribeContractRequest(_)
                | OutboundDelegateMsg::SendDelegateMessage(_) => None,
            })
            .expect("Expected an ApplicationMessage (Echo)");
        assert!(echo_msg.processed);

        let echo_response: DelegateResponse = bincode::deserialize(&echo_msg.payload)?;
        match echo_response {
            DelegateResponse::Echo { message } => {
                assert_eq!(message, echo_message);
            }
            other @ DelegateResponse::ContractState { .. }
            | other @ DelegateResponse::MultipleContractStates { .. }
            | other @ DelegateResponse::ContractPutResult { .. }
            | other @ DelegateResponse::ContractUpdateResult { .. }
            | other @ DelegateResponse::ContractSubscribeResult { .. }
            | other @ DelegateResponse::ContractNotificationReceived { .. }
            | other @ DelegateResponse::Error { .. } => {
                panic!("Expected Echo response, got {:?}", other)
            }
        }

        let contract_response = InboundDelegateMsg::GetContractResponse(GetContractResponse {
            contract_id,
            state: Some(WrappedState::new(vec![1, 2, 3, 4])),
            context: contract_request.context,
        });

        let final_outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![contract_response],
        )?;

        assert_eq!(final_outbound.len(), 1);
        let final_msg = match &final_outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => msg,
            other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage, got {:?}", other)
            }
        };

        let response: DelegateResponse = bincode::deserialize(&final_msg.payload)?;
        match response {
            DelegateResponse::ContractState {
                contract_id: id,
                state,
            } => {
                assert_eq!(id, contract_id);
                assert_eq!(state, Some(vec![1, 2, 3, 4]));
            }
            other @ DelegateResponse::MultipleContractStates { .. }
            | other @ DelegateResponse::Echo { .. }
            | other @ DelegateResponse::ContractPutResult { .. }
            | other @ DelegateResponse::ContractUpdateResult { .. }
            | other @ DelegateResponse::ContractSubscribeResult { .. }
            | other @ DelegateResponse::ContractNotificationReceived { .. }
            | other @ DelegateResponse::Error { .. } => {
                panic!("Expected ContractState response, got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn validate_host_function_delegate() -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_2).await?;
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let payload: Vec<u8> = bincode::serialize(&InboundAppMessage::CreateInboxRequest).unwrap();
        let create_msg = ApplicationMessage::new(payload);
        let inbound = InboundDelegateMsg::ApplicationMessage(create_msg);
        let outbound =
            runtime.inbound_app_message(delegate.key(), &vec![].into(), None, vec![inbound])?;

        let expected_payload =
            bincode::serialize(&OutboundAppMessage::CreateInboxResponse(vec![1])).unwrap();
        assert_eq!(outbound.len(), 1);
        assert!(matches!(
            outbound.first(),
            Some(OutboundDelegateMsg::ApplicationMessage(msg)) if *msg.payload == expected_payload
        ));

        let payload: Vec<u8> =
            bincode::serialize(&InboundAppMessage::PleaseSignMessage(vec![1, 2, 3])).unwrap();
        let sign_msg = ApplicationMessage::new(payload);
        let inbound = InboundDelegateMsg::ApplicationMessage(sign_msg);
        let outbound =
            runtime.inbound_app_message(delegate.key(), &vec![].into(), None, vec![inbound])?;

        let expected_payload =
            bincode::serialize(&OutboundAppMessage::MessageSigned(vec![4, 5, 2])).unwrap();
        assert_eq!(outbound.len(), 1);
        assert!(matches!(
            outbound.first(),
            Some(OutboundDelegateMsg::ApplicationMessage(msg)) if *msg.payload == expected_payload
        ));

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_context_persistence_within_call() -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_2).await?;
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let test_data = vec![1, 2, 3, 4, 5];

        let write_payload =
            bincode::serialize(&InboundAppMessage::WriteContext(test_data.clone()))?;
        let read_payload = bincode::serialize(&InboundAppMessage::ReadContext)?;

        let messages = vec![
            InboundDelegateMsg::ApplicationMessage(ApplicationMessage::new(write_payload)),
            InboundDelegateMsg::ApplicationMessage(ApplicationMessage::new(read_payload)),
        ];

        let outbound =
            runtime.inbound_app_message(delegate.key(), &vec![].into(), None, messages)?;

        assert_eq!(outbound.len(), 2);

        let response1: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        assert!(matches!(response1, OutboundAppMessage::ContextWritten));

        let response2: OutboundAppMessage = match &outbound[1] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        match response2 {
            OutboundAppMessage::ContextData(data) => {
                assert_eq!(data, test_data);
            }
            other @ OutboundAppMessage::CreateInboxResponse(_)
            | other @ OutboundAppMessage::MessageSigned(_)
            | other @ OutboundAppMessage::CounterValue(_)
            | other @ OutboundAppMessage::SecretExists(_)
            | other @ OutboundAppMessage::SecretResult(_)
            | other @ OutboundAppMessage::ContextWritten
            | other @ OutboundAppMessage::ContextCleared
            | other @ OutboundAppMessage::SecretStored
            | other @ OutboundAppMessage::SecretRemoved
            | other @ OutboundAppMessage::LargeContextWritten(_)
            | other @ OutboundAppMessage::LargeSecretStored(_)
            | other @ OutboundAppMessage::SecretStoreFailed => {
                panic!("Expected ContextData, got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_context_reset_between_calls() -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_2).await?;
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let payload = bincode::serialize(&InboundAppMessage::IncrementCounter)?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;

        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        assert!(matches!(response, OutboundAppMessage::CounterValue(1)));

        let payload = bincode::serialize(&InboundAppMessage::IncrementCounter)?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;

        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        assert!(matches!(response, OutboundAppMessage::CounterValue(1)));

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_has_secret_host_function() -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_2).await?;
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let secret_key = vec![10, 20, 30];
        let secret_value = vec![100, 200];

        let payload = bincode::serialize(&InboundAppMessage::HasSecret(secret_key.clone()))?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;

        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        assert!(matches!(response, OutboundAppMessage::SecretExists(false)));

        let payload = bincode::serialize(&InboundAppMessage::StoreSecret {
            key: secret_key.clone(),
            value: secret_value.clone(),
        })?;
        let msg = ApplicationMessage::new(payload);
        let _ = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;

        let payload = bincode::serialize(&InboundAppMessage::HasSecret(secret_key.clone()))?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;

        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        assert!(matches!(response, OutboundAppMessage::SecretExists(true)));

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_get_nonexistent_secret() -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_2).await?;
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let nonexistent_key = vec![99, 98, 97];
        let payload =
            bincode::serialize(&InboundAppMessage::GetNonExistentSecret(nonexistent_key))?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;

        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        assert!(matches!(response, OutboundAppMessage::SecretResult(None)));

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_store_and_retrieve_secret() -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_2).await?;
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let secret_key = vec![42, 43, 44];
        let secret_value = vec![1, 2, 3, 4, 5, 6, 7, 8];

        let payload = bincode::serialize(&InboundAppMessage::StoreSecret {
            key: secret_key.clone(),
            value: secret_value.clone(),
        })?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;

        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        assert!(matches!(response, OutboundAppMessage::SecretStored));

        let payload =
            bincode::serialize(&InboundAppMessage::GetNonExistentSecret(secret_key.clone()))?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;

        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        match response {
            OutboundAppMessage::SecretResult(Some(value)) => {
                assert_eq!(value, secret_value);
            }
            other @ OutboundAppMessage::CreateInboxResponse(_)
            | other @ OutboundAppMessage::MessageSigned(_)
            | other @ OutboundAppMessage::ContextData(_)
            | other @ OutboundAppMessage::CounterValue(_)
            | other @ OutboundAppMessage::SecretExists(_)
            | other @ OutboundAppMessage::SecretResult(_)
            | other @ OutboundAppMessage::ContextWritten
            | other @ OutboundAppMessage::ContextCleared
            | other @ OutboundAppMessage::SecretStored
            | other @ OutboundAppMessage::SecretRemoved
            | other @ OutboundAppMessage::LargeContextWritten(_)
            | other @ OutboundAppMessage::LargeSecretStored(_)
            | other @ OutboundAppMessage::SecretStoreFailed => {
                panic!("Expected SecretResult(Some(...)), got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_set_secret_failure_returns_secret_store_failed()
    -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_2).await?;
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        // Make secrets directory read-only so store_secret fails with an I/O error
        let secrets_dir = temp_dir.path().join("secrets");
        std::fs::set_permissions(&secrets_dir, std::fs::Permissions::from_mode(0o444))?;

        let payload = bincode::serialize(&InboundAppMessage::StoreSecret {
            key: vec![1, 2, 3],
            value: vec![4, 5, 6],
        })?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;

        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        assert!(
            matches!(response, OutboundAppMessage::SecretStoreFailed),
            "Expected SecretStoreFailed when secrets dir is read-only, got {:?}",
            response
        );

        // Restore permissions so temp_dir cleanup works
        std::fs::set_permissions(&secrets_dir, std::fs::Permissions::from_mode(0o755))?;
        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_read_empty_context() -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_2).await?;
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let payload = bincode::serialize(&InboundAppMessage::ReadContext)?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;

        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        match response {
            OutboundAppMessage::ContextData(data) => {
                assert!(data.is_empty());
            }
            other @ OutboundAppMessage::CreateInboxResponse(_)
            | other @ OutboundAppMessage::MessageSigned(_)
            | other @ OutboundAppMessage::CounterValue(_)
            | other @ OutboundAppMessage::SecretExists(_)
            | other @ OutboundAppMessage::SecretResult(_)
            | other @ OutboundAppMessage::ContextWritten
            | other @ OutboundAppMessage::ContextCleared
            | other @ OutboundAppMessage::SecretStored
            | other @ OutboundAppMessage::SecretRemoved
            | other @ OutboundAppMessage::LargeContextWritten(_)
            | other @ OutboundAppMessage::LargeSecretStored(_)
            | other @ OutboundAppMessage::SecretStoreFailed => {
                panic!("Expected ContextData, got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_context_clear() -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_2).await?;
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let test_data = vec![1, 2, 3, 4, 5];
        let payload = bincode::serialize(&InboundAppMessage::WriteContext(test_data.clone()))?;
        let msg = ApplicationMessage::new(payload);
        let _ = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;

        let payload = bincode::serialize(&InboundAppMessage::ClearContext)?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;

        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        assert!(matches!(response, OutboundAppMessage::ContextCleared));

        let payload = bincode::serialize(&InboundAppMessage::ReadContext)?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;

        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        match response {
            OutboundAppMessage::ContextData(data) => {
                assert!(data.is_empty());
            }
            other @ OutboundAppMessage::CreateInboxResponse(_)
            | other @ OutboundAppMessage::MessageSigned(_)
            | other @ OutboundAppMessage::CounterValue(_)
            | other @ OutboundAppMessage::SecretExists(_)
            | other @ OutboundAppMessage::SecretResult(_)
            | other @ OutboundAppMessage::ContextWritten
            | other @ OutboundAppMessage::ContextCleared
            | other @ OutboundAppMessage::SecretStored
            | other @ OutboundAppMessage::SecretRemoved
            | other @ OutboundAppMessage::LargeContextWritten(_)
            | other @ OutboundAppMessage::LargeSecretStored(_)
            | other @ OutboundAppMessage::SecretStoreFailed => {
                panic!("Expected ContextData, got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_context_shared_across_batch() -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_2).await?;
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let messages: Vec<InboundDelegateMsg> = (0..3)
            .map(|_| {
                let payload = bincode::serialize(&InboundAppMessage::IncrementCounter).unwrap();
                InboundDelegateMsg::ApplicationMessage(ApplicationMessage::new(payload))
            })
            .collect();

        let outbound =
            runtime.inbound_app_message(delegate.key(), &vec![].into(), None, messages)?;

        assert_eq!(outbound.len(), 3);

        for (i, msg) in outbound.iter().enumerate() {
            let response: OutboundAppMessage = match msg {
                OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
                OutboundDelegateMsg::RequestUserInput(_)
                | OutboundDelegateMsg::ContextUpdated(_)
                | OutboundDelegateMsg::GetContractRequest(_)
                | OutboundDelegateMsg::PutContractRequest(_)
                | OutboundDelegateMsg::UpdateContractRequest(_)
                | OutboundDelegateMsg::SubscribeContractRequest(_)
                | OutboundDelegateMsg::SendDelegateMessage(_) => {
                    panic!("Expected ApplicationMessage")
                }
            };
            match response {
                OutboundAppMessage::CounterValue(value) => {
                    assert_eq!(value, (i + 1) as u32);
                }
                other @ OutboundAppMessage::CreateInboxResponse(_)
                | other @ OutboundAppMessage::MessageSigned(_)
                | other @ OutboundAppMessage::ContextData(_)
                | other @ OutboundAppMessage::SecretExists(_)
                | other @ OutboundAppMessage::SecretResult(_)
                | other @ OutboundAppMessage::ContextWritten
                | other @ OutboundAppMessage::ContextCleared
                | other @ OutboundAppMessage::SecretStored
                | other @ OutboundAppMessage::SecretRemoved
                | other @ OutboundAppMessage::LargeContextWritten(_)
                | other @ OutboundAppMessage::LargeSecretStored(_)
                | other @ OutboundAppMessage::SecretStoreFailed => {
                    panic!("Expected CounterValue, got {:?}", other)
                }
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_remove_secret_host_function() -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_2).await?;
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let secret_key = vec![50, 51, 52];
        let secret_value = vec![200, 201, 202];

        let payload = bincode::serialize(&InboundAppMessage::StoreSecret {
            key: secret_key.clone(),
            value: secret_value.clone(),
        })?;
        let msg = ApplicationMessage::new(payload);
        let _ = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;

        let payload = bincode::serialize(&InboundAppMessage::HasSecret(secret_key.clone()))?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;
        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        assert!(matches!(response, OutboundAppMessage::SecretExists(true)));

        let payload = bincode::serialize(&InboundAppMessage::RemoveSecret(secret_key.clone()))?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;
        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        assert!(matches!(response, OutboundAppMessage::SecretRemoved));

        let payload = bincode::serialize(&InboundAppMessage::HasSecret(secret_key.clone()))?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;
        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        assert!(matches!(response, OutboundAppMessage::SecretExists(false)));

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_large_context_data() -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_2).await?;
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let large_size = 1024 * 1024;

        let payload = bincode::serialize(&InboundAppMessage::WriteLargeContext(large_size))?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;
        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        match response {
            OutboundAppMessage::LargeContextWritten(size) => {
                assert_eq!(size, large_size);
            }
            other @ OutboundAppMessage::CreateInboxResponse(_)
            | other @ OutboundAppMessage::MessageSigned(_)
            | other @ OutboundAppMessage::ContextData(_)
            | other @ OutboundAppMessage::CounterValue(_)
            | other @ OutboundAppMessage::SecretExists(_)
            | other @ OutboundAppMessage::SecretResult(_)
            | other @ OutboundAppMessage::ContextWritten
            | other @ OutboundAppMessage::ContextCleared
            | other @ OutboundAppMessage::SecretStored
            | other @ OutboundAppMessage::SecretRemoved
            | other @ OutboundAppMessage::LargeSecretStored(_)
            | other @ OutboundAppMessage::SecretStoreFailed => {
                panic!("Expected LargeContextWritten, got {:?}", other)
            }
        }

        let payload = bincode::serialize(&InboundAppMessage::ReadContext)?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;
        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        match response {
            OutboundAppMessage::ContextData(data) => {
                assert!(data.is_empty());
            }
            other @ OutboundAppMessage::CreateInboxResponse(_)
            | other @ OutboundAppMessage::MessageSigned(_)
            | other @ OutboundAppMessage::CounterValue(_)
            | other @ OutboundAppMessage::SecretExists(_)
            | other @ OutboundAppMessage::SecretResult(_)
            | other @ OutboundAppMessage::ContextWritten
            | other @ OutboundAppMessage::ContextCleared
            | other @ OutboundAppMessage::SecretStored
            | other @ OutboundAppMessage::SecretRemoved
            | other @ OutboundAppMessage::LargeContextWritten(_)
            | other @ OutboundAppMessage::LargeSecretStored(_)
            | other @ OutboundAppMessage::SecretStoreFailed => {
                panic!("Expected ContextData, got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_large_context_within_batch() -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_2).await?;
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let large_size = 256 * 1024;

        let write_payload = bincode::serialize(&InboundAppMessage::WriteLargeContext(large_size))?;
        let read_payload = bincode::serialize(&InboundAppMessage::ReadContext)?;

        let messages = vec![
            InboundDelegateMsg::ApplicationMessage(ApplicationMessage::new(write_payload)),
            InboundDelegateMsg::ApplicationMessage(ApplicationMessage::new(read_payload)),
        ];

        let outbound =
            runtime.inbound_app_message(delegate.key(), &vec![].into(), None, messages)?;

        assert_eq!(outbound.len(), 2);

        let response: OutboundAppMessage = match &outbound[1] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        match response {
            OutboundAppMessage::ContextData(data) => {
                assert_eq!(data.len(), large_size);
                for (i, byte) in data.iter().enumerate() {
                    assert_eq!(*byte, (i % 256) as u8, "Data pattern mismatch at index {i}");
                }
            }
            other @ OutboundAppMessage::CreateInboxResponse(_)
            | other @ OutboundAppMessage::MessageSigned(_)
            | other @ OutboundAppMessage::CounterValue(_)
            | other @ OutboundAppMessage::SecretExists(_)
            | other @ OutboundAppMessage::SecretResult(_)
            | other @ OutboundAppMessage::ContextWritten
            | other @ OutboundAppMessage::ContextCleared
            | other @ OutboundAppMessage::SecretStored
            | other @ OutboundAppMessage::SecretRemoved
            | other @ OutboundAppMessage::LargeContextWritten(_)
            | other @ OutboundAppMessage::LargeSecretStored(_)
            | other @ OutboundAppMessage::SecretStoreFailed => {
                panic!("Expected ContextData, got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    /// Indirect regression test for #3248 (stale WASM memory base pointer).
    ///
    /// Storing a 1 MB secret forces `memory.grow` which can relocate WASM linear
    /// memory. If the cached `MEM_ADDR.start_ptr` is not refreshed via
    /// `refresh_mem_addr_from_caller`, the subsequent read uses a stale pointer
    /// and returns garbage data. Under full parallel test suite runs (~1600 tests)
    /// the relocation is more likely due to memory pressure.
    #[tokio::test(flavor = "multi_thread")]
    async fn test_large_secret_data() -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_2).await?;
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let secret_key = vec![77, 88, 99];
        let large_size = 1024 * 1024;

        let payload = bincode::serialize(&InboundAppMessage::StoreLargeSecret {
            key: secret_key.clone(),
            size: large_size,
        })?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;
        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        match response {
            OutboundAppMessage::LargeSecretStored(size) => {
                assert_eq!(size, large_size);
            }
            other @ OutboundAppMessage::CreateInboxResponse(_)
            | other @ OutboundAppMessage::MessageSigned(_)
            | other @ OutboundAppMessage::ContextData(_)
            | other @ OutboundAppMessage::CounterValue(_)
            | other @ OutboundAppMessage::SecretExists(_)
            | other @ OutboundAppMessage::SecretResult(_)
            | other @ OutboundAppMessage::ContextWritten
            | other @ OutboundAppMessage::ContextCleared
            | other @ OutboundAppMessage::SecretStored
            | other @ OutboundAppMessage::SecretRemoved
            | other @ OutboundAppMessage::LargeContextWritten(_)
            | other @ OutboundAppMessage::SecretStoreFailed => {
                panic!("Expected LargeSecretStored, got {:?}", other)
            }
        }

        let payload =
            bincode::serialize(&InboundAppMessage::GetNonExistentSecret(secret_key.clone()))?;
        let msg = ApplicationMessage::new(payload);
        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(msg)],
        )?;
        let response: OutboundAppMessage = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => bincode::deserialize(&m.payload)?,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage")
            }
        };
        match response {
            OutboundAppMessage::SecretResult(Some(data)) => {
                assert_eq!(data.len(), large_size);
                for (i, byte) in data.iter().enumerate() {
                    assert_eq!(*byte, (i % 256) as u8, "Data pattern mismatch at index {i}");
                }
            }
            other @ OutboundAppMessage::CreateInboxResponse(_)
            | other @ OutboundAppMessage::MessageSigned(_)
            | other @ OutboundAppMessage::ContextData(_)
            | other @ OutboundAppMessage::CounterValue(_)
            | other @ OutboundAppMessage::SecretExists(_)
            | other @ OutboundAppMessage::SecretResult(_)
            | other @ OutboundAppMessage::ContextWritten
            | other @ OutboundAppMessage::ContextCleared
            | other @ OutboundAppMessage::SecretStored
            | other @ OutboundAppMessage::SecretRemoved
            | other @ OutboundAppMessage::LargeContextWritten(_)
            | other @ OutboundAppMessage::LargeSecretStored(_)
            | other @ OutboundAppMessage::SecretStoreFailed => {
                panic!("Expected SecretResult(Some(...)), got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn test_concurrent_delegate_execution() -> Result<(), Box<dyn std::error::Error>> {
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};
        use std::sync::Arc;
        use tokio::sync::Barrier;

        let (delegate1, runtime1, temp_dir1) = setup_runtime(TEST_DELEGATE_2).await?;
        let (delegate2, runtime2, temp_dir2) = setup_runtime(TEST_DELEGATE_2).await?;

        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let runtime1 = Arc::new(std::sync::Mutex::new(runtime1));
        let runtime2 = Arc::new(std::sync::Mutex::new(runtime2));
        let delegate1 = Arc::new(delegate1);
        let delegate2 = Arc::new(delegate2);

        let barrier = Arc::new(Barrier::new(2));

        let barrier1 = barrier.clone();
        let runtime1_clone = runtime1.clone();
        let delegate1_clone = delegate1.clone();
        let handle1 = tokio::spawn(async move {
            barrier1.wait().await;

            let secret_key = b"thread1_key".to_vec();
            let secret_value = b"thread1_value".to_vec();

            let payload = bincode::serialize(&InboundAppMessage::StoreSecret {
                key: secret_key.clone(),
                value: secret_value.clone(),
            })
            .unwrap();
            let msg = ApplicationMessage::new(payload);
            {
                let mut runtime = runtime1_clone.lock().unwrap();
                let _ = runtime
                    .inbound_app_message(
                        delegate1_clone.key(),
                        &vec![].into(),
                        None,
                        vec![InboundDelegateMsg::ApplicationMessage(msg)],
                    )
                    .unwrap();
            }

            let payload =
                bincode::serialize(&InboundAppMessage::GetNonExistentSecret(secret_key.clone()))
                    .unwrap();
            let msg = ApplicationMessage::new(payload);
            let outbound = {
                let mut runtime = runtime1_clone.lock().unwrap();
                runtime
                    .inbound_app_message(
                        delegate1_clone.key(),
                        &vec![].into(),
                        None,
                        vec![InboundDelegateMsg::ApplicationMessage(msg)],
                    )
                    .unwrap()
            };

            let response: OutboundAppMessage = match &outbound[0] {
                OutboundDelegateMsg::ApplicationMessage(m) => {
                    bincode::deserialize(&m.payload).unwrap()
                }
                OutboundDelegateMsg::RequestUserInput(_)
                | OutboundDelegateMsg::ContextUpdated(_)
                | OutboundDelegateMsg::GetContractRequest(_)
                | OutboundDelegateMsg::PutContractRequest(_)
                | OutboundDelegateMsg::UpdateContractRequest(_)
                | OutboundDelegateMsg::SubscribeContractRequest(_)
                | OutboundDelegateMsg::SendDelegateMessage(_) => {
                    panic!("Expected ApplicationMessage")
                }
            };
            match response {
                OutboundAppMessage::SecretResult(Some(value)) => {
                    assert_eq!(value, secret_value);
                }
                other @ OutboundAppMessage::CreateInboxResponse(_)
                | other @ OutboundAppMessage::MessageSigned(_)
                | other @ OutboundAppMessage::ContextData(_)
                | other @ OutboundAppMessage::CounterValue(_)
                | other @ OutboundAppMessage::SecretExists(_)
                | other @ OutboundAppMessage::SecretResult(_)
                | other @ OutboundAppMessage::ContextWritten
                | other @ OutboundAppMessage::ContextCleared
                | other @ OutboundAppMessage::SecretStored
                | other @ OutboundAppMessage::SecretRemoved
                | other @ OutboundAppMessage::LargeContextWritten(_)
                | other @ OutboundAppMessage::LargeSecretStored(_)
                | other @ OutboundAppMessage::SecretStoreFailed => panic!(
                    "Thread 1: Expected SecretResult(Some(...)), got {:?}",
                    other
                ),
            }
        });

        let barrier2 = barrier.clone();
        let runtime2_clone = runtime2.clone();
        let delegate2_clone = delegate2.clone();
        let handle2 = tokio::spawn(async move {
            barrier2.wait().await;

            let secret_key = b"thread2_key".to_vec();
            let secret_value = b"thread2_value".to_vec();

            let payload = bincode::serialize(&InboundAppMessage::StoreSecret {
                key: secret_key.clone(),
                value: secret_value.clone(),
            })
            .unwrap();
            let msg = ApplicationMessage::new(payload);
            {
                let mut runtime = runtime2_clone.lock().unwrap();
                let _ = runtime
                    .inbound_app_message(
                        delegate2_clone.key(),
                        &vec![].into(),
                        None,
                        vec![InboundDelegateMsg::ApplicationMessage(msg)],
                    )
                    .unwrap();
            }

            let payload =
                bincode::serialize(&InboundAppMessage::GetNonExistentSecret(secret_key.clone()))
                    .unwrap();
            let msg = ApplicationMessage::new(payload);
            let outbound = {
                let mut runtime = runtime2_clone.lock().unwrap();
                runtime
                    .inbound_app_message(
                        delegate2_clone.key(),
                        &vec![].into(),
                        None,
                        vec![InboundDelegateMsg::ApplicationMessage(msg)],
                    )
                    .unwrap()
            };

            let response: OutboundAppMessage = match &outbound[0] {
                OutboundDelegateMsg::ApplicationMessage(m) => {
                    bincode::deserialize(&m.payload).unwrap()
                }
                OutboundDelegateMsg::RequestUserInput(_)
                | OutboundDelegateMsg::ContextUpdated(_)
                | OutboundDelegateMsg::GetContractRequest(_)
                | OutboundDelegateMsg::PutContractRequest(_)
                | OutboundDelegateMsg::UpdateContractRequest(_)
                | OutboundDelegateMsg::SubscribeContractRequest(_)
                | OutboundDelegateMsg::SendDelegateMessage(_) => {
                    panic!("Expected ApplicationMessage")
                }
            };
            match response {
                OutboundAppMessage::SecretResult(Some(value)) => {
                    assert_eq!(value, secret_value);
                }
                other @ OutboundAppMessage::CreateInboxResponse(_)
                | other @ OutboundAppMessage::MessageSigned(_)
                | other @ OutboundAppMessage::ContextData(_)
                | other @ OutboundAppMessage::CounterValue(_)
                | other @ OutboundAppMessage::SecretExists(_)
                | other @ OutboundAppMessage::SecretResult(_)
                | other @ OutboundAppMessage::ContextWritten
                | other @ OutboundAppMessage::ContextCleared
                | other @ OutboundAppMessage::SecretStored
                | other @ OutboundAppMessage::SecretRemoved
                | other @ OutboundAppMessage::LargeContextWritten(_)
                | other @ OutboundAppMessage::LargeSecretStored(_)
                | other @ OutboundAppMessage::SecretStoreFailed => panic!(
                    "Thread 2: Expected SecretResult(Some(...)), got {:?}",
                    other
                ),
            }
        });

        handle1.await?;
        handle2.await?;

        std::mem::drop(temp_dir1);
        std::mem::drop(temp_dir2);
        Ok(())
    }

    /// Verify that V1 delegates are correctly detected as V1 even when
    /// state_store_db is configured. This ensures backward compatibility —
    /// V2 detection is based on module imports, not runtime configuration.
    #[tokio::test(flavor = "multi_thread")]
    async fn test_v1_delegate_detected_as_v1_with_state_store()
    -> Result<(), Box<dyn std::error::Error>> {
        use crate::contract::storages::Storage;
        use delegate2_messages::{InboundAppMessage, OutboundAppMessage};

        let temp_dir = get_temp_dir();
        let contracts_dir = temp_dir.path().join("contracts");
        let delegates_dir = temp_dir.path().join("delegates");
        let secrets_dir = temp_dir.path().join("secrets");

        let db = Storage::new(temp_dir.path()).await?;
        let contract_store = ContractStore::new(contracts_dir, 10_000, db.clone())?;
        let delegate_store = DelegateStore::new(delegates_dir, 10_000, db.clone())?;
        let secret_store = SecretsStore::new(secrets_dir, Default::default(), db.clone())?;

        let mut runtime =
            Runtime::build(contract_store, delegate_store, secret_store, false).unwrap();

        // Configure state_store_db — V1 delegates should STILL be detected as V1
        runtime.set_state_store_db(db);

        let delegate = {
            let bytes = super::super::tests::get_test_module(TEST_DELEGATE_2)?;
            DelegateContainer::Wasm(DelegateWasmAPIVersion::V1(Delegate::from((
                &bytes.into(),
                &vec![].into(),
            ))))
        };
        let _stored = runtime.delegate_store.store_delegate(delegate.clone());

        let key = XChaCha20Poly1305::generate_key(&mut OsRng);
        let cipher = XChaCha20Poly1305::new(&key);
        let nonce = XChaCha20Poly1305::generate_nonce(&mut OsRng);
        let _registered =
            runtime
                .secret_store
                .register_delegate(delegate.key().clone(), cipher, nonce);

        // Verify API version detection: V1 delegate should be V1
        let (mut running, api_version) =
            runtime.prepare_delegate_call(&vec![].into(), delegate.key(), 4096)?;
        assert_eq!(
            api_version,
            DelegateApiVersion::V1,
            "V1 delegate should be detected as V1 even with state_store_db configured"
        );
        runtime.drop_running_instance(&mut running);

        // Verify the delegate still works normally via the V1 path
        let contract = WrappedContract::new(
            Arc::new(ContractCode::from(vec![1])),
            Parameters::from(vec![]),
        );
        let _app = ContractInstanceId::try_from(contract.key.to_string()).unwrap();

        let payload: Vec<u8> = bincode::serialize(&InboundAppMessage::CreateInboxRequest).unwrap();
        let create_msg = ApplicationMessage::new(payload);
        let inbound = InboundDelegateMsg::ApplicationMessage(create_msg);
        let outbound =
            runtime.inbound_app_message(delegate.key(), &vec![].into(), None, vec![inbound])?;

        let expected_payload =
            bincode::serialize(&OutboundAppMessage::CreateInboxResponse(vec![1])).unwrap();
        assert_eq!(outbound.len(), 1);
        assert!(matches!(
            outbound.first(),
            Some(OutboundDelegateMsg::ApplicationMessage(msg)) if *msg.payload == expected_payload
        ));

        std::mem::drop(temp_dir);
        Ok(())
    }

    const TEST_DELEGATE_V2_CONTRACTS: &str = "test_delegate_v2_contracts";

    /// Message types for test-delegate-v2-contracts (must match the delegate's types)
    mod v2_contracts_messages {
        use super::*;

        #[derive(Debug, Serialize, Deserialize)]
        pub enum InboundAppMessage {
            GetContractState {
                contract_id: [u8; 32],
            },
            PutContractState {
                contract_id: [u8; 32],
                state: Vec<u8>,
            },
            UpdateContractState {
                contract_id: [u8; 32],
                state: Vec<u8>,
            },
            SubscribeContract {
                contract_id: [u8; 32],
            },
        }

        #[derive(Debug, Serialize, Deserialize)]
        pub enum OutboundAppMessage {
            ContractState {
                contract_id: [u8; 32],
                state: Vec<u8>,
            },
            ContractNotFound {
                contract_id: [u8; 32],
                error_code: i64,
            },
            Success {
                contract_id: [u8; 32],
            },
            Failed {
                contract_id: [u8; 32],
                error_code: i64,
            },
        }
    }

    /// V2 delegate end-to-end test: a real compiled WASM delegate that reads
    /// contract state via host functions from the `freenet_delegate_contracts`
    /// namespace. This exercises the full V2 async call path:
    ///
    /// 1. Module is detected as V2 (imports `freenet_delegate_contracts`)
    /// 2. `call_3i64_async_imports` is used instead of `call_3i64`
    /// 3. Host functions `get_contract_state_len` and `get_contract_state`
    ///    read from the ReDb state store
    #[tokio::test(flavor = "multi_thread")]
    async fn test_v2_delegate_reads_contract_state() -> Result<(), Box<dyn std::error::Error>> {
        use crate::contract::storages::Storage;
        use crate::wasm_runtime::StateStorage;
        use v2_contracts_messages::*;

        let temp_dir = get_temp_dir();
        let contracts_dir = temp_dir.path().join("contracts");
        let delegates_dir = temp_dir.path().join("delegates");
        let secrets_dir = temp_dir.path().join("secrets");

        let db = Storage::new(temp_dir.path()).await?;
        let contract_store = ContractStore::new(contracts_dir, 10_000, db.clone())?;
        let delegate_store = DelegateStore::new(delegates_dir, 10_000, db.clone())?;
        let secret_store = SecretsStore::new(secrets_dir, Default::default(), db.clone())?;

        let mut runtime =
            Runtime::build(contract_store, delegate_store, secret_store, false).unwrap();
        runtime.set_state_store_db(db.clone());

        // Store contract state in the DB so the V2 delegate can read it
        let contract_instance_id = ContractInstanceId::new([42u8; 32]);
        let contract_code = ContractCode::from(vec![1, 2, 3]);
        let contract_key =
            ContractKey::from_id_and_code(contract_instance_id, *contract_code.hash());
        let expected_state = vec![10, 20, 30, 40, 50, 60, 70, 80];
        db.store(contract_key, WrappedState::new(expected_state.clone()))
            .await?;
        // Index the contract so code_hash_from_id() works
        runtime.contract_store.ensure_key_indexed(&contract_key)?;

        // Load the V2 delegate
        let delegate = {
            let bytes = super::super::tests::get_test_module(TEST_DELEGATE_V2_CONTRACTS)?;
            DelegateContainer::Wasm(DelegateWasmAPIVersion::V1(Delegate::from((
                &bytes.into(),
                &vec![].into(),
            ))))
        };
        let _stored = runtime.delegate_store.store_delegate(delegate.clone());

        let key = XChaCha20Poly1305::generate_key(&mut OsRng);
        let cipher = XChaCha20Poly1305::new(&key);
        let nonce = XChaCha20Poly1305::generate_nonce(&mut OsRng);
        let _registered =
            runtime
                .secret_store
                .register_delegate(delegate.key().clone(), cipher, nonce);

        // Verify the module is detected as V2
        let (mut running, api_version) =
            runtime.prepare_delegate_call(&vec![].into(), delegate.key(), 4096)?;
        assert_eq!(
            api_version,
            DelegateApiVersion::V2,
            "V2 delegate should be detected as V2 (imports freenet_delegate_contracts)"
        );
        runtime.drop_running_instance(&mut running);

        // Send a message asking the delegate to read the contract state
        let _app_id = ContractInstanceId::new([1u8; 32]);
        let command = InboundAppMessage::GetContractState {
            contract_id: [42u8; 32],
        };
        let payload = bincode::serialize(&command)?;
        let app_msg = ApplicationMessage::new(payload);

        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(app_msg)],
        )?;

        assert_eq!(outbound.len(), 1, "Expected exactly one outbound message");
        let response_msg = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => msg,
            other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage, got {:?}", other)
            }
        };
        assert!(response_msg.processed);

        let response: OutboundAppMessage = bincode::deserialize(&response_msg.payload)?;
        match response {
            OutboundAppMessage::ContractState { contract_id, state } => {
                assert_eq!(contract_id, [42u8; 32]);
                assert_eq!(
                    state, expected_state,
                    "V2 delegate should read contract state via host functions"
                );
            }
            other @ OutboundAppMessage::ContractNotFound { .. }
            | other @ OutboundAppMessage::Success { .. }
            | other @ OutboundAppMessage::Failed { .. } => {
                panic!("V2 delegate returned {other:?} — expected ContractState");
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    /// V2 delegate: contract not found returns error code.
    #[tokio::test(flavor = "multi_thread")]
    async fn test_v2_delegate_contract_not_found() -> Result<(), Box<dyn std::error::Error>> {
        use crate::contract::storages::Storage;
        use v2_contracts_messages::*;

        let temp_dir = get_temp_dir();
        let contracts_dir = temp_dir.path().join("contracts");
        let delegates_dir = temp_dir.path().join("delegates");
        let secrets_dir = temp_dir.path().join("secrets");

        let db = Storage::new(temp_dir.path()).await?;
        let contract_store = ContractStore::new(contracts_dir, 10_000, db.clone())?;
        let delegate_store = DelegateStore::new(delegates_dir, 10_000, db.clone())?;
        let secret_store = SecretsStore::new(secrets_dir, Default::default(), db.clone())?;

        let mut runtime =
            Runtime::build(contract_store, delegate_store, secret_store, false).unwrap();
        runtime.set_state_store_db(db);

        // Load the V2 delegate (no contract state stored — should get not-found)
        let delegate = {
            let bytes = super::super::tests::get_test_module(TEST_DELEGATE_V2_CONTRACTS)?;
            DelegateContainer::Wasm(DelegateWasmAPIVersion::V1(Delegate::from((
                &bytes.into(),
                &vec![].into(),
            ))))
        };
        let _stored = runtime.delegate_store.store_delegate(delegate.clone());

        let key = XChaCha20Poly1305::generate_key(&mut OsRng);
        let cipher = XChaCha20Poly1305::new(&key);
        let nonce = XChaCha20Poly1305::generate_nonce(&mut OsRng);
        let _registered =
            runtime
                .secret_store
                .register_delegate(delegate.key().clone(), cipher, nonce);

        // Ask for a contract that doesn't exist
        let _app_id = ContractInstanceId::new([1u8; 32]);
        let command = InboundAppMessage::GetContractState {
            contract_id: [99u8; 32],
        };
        let payload = bincode::serialize(&command)?;
        let app_msg = ApplicationMessage::new(payload);

        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(app_msg)],
        )?;

        assert_eq!(outbound.len(), 1);
        let response_msg = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => msg,
            other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage, got {:?}", other)
            }
        };

        let response: OutboundAppMessage = bincode::deserialize(&response_msg.payload)?;
        match response {
            OutboundAppMessage::ContractNotFound { error_code, .. } => {
                assert!(
                    error_code < 0,
                    "Expected negative error code for not-found, got {error_code}"
                );
            }
            other @ OutboundAppMessage::ContractState { .. }
            | other @ OutboundAppMessage::Success { .. }
            | other @ OutboundAppMessage::Failed { .. } => {
                panic!("Expected ContractNotFound for non-existent contract, got {other:?}");
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    /// Helper: set up a V2 delegate runtime with a registered contract.
    async fn setup_v2_runtime_with_contract(
        contract_id_byte: u8,
        initial_state: Option<&[u8]>,
    ) -> Result<
        (
            DelegateContainer,
            Runtime,
            ContractInstanceId,
            tempfile::TempDir,
        ),
        Box<dyn std::error::Error>,
    > {
        use crate::contract::storages::Storage;
        use crate::wasm_runtime::StateStorage;

        let temp_dir = get_temp_dir();
        let contracts_dir = temp_dir.path().join("contracts");
        let delegates_dir = temp_dir.path().join("delegates");
        let secrets_dir = temp_dir.path().join("secrets");

        let db = Storage::new(temp_dir.path()).await?;
        let contract_store = ContractStore::new(contracts_dir, 10_000, db.clone())?;
        let delegate_store = DelegateStore::new(delegates_dir, 10_000, db.clone())?;
        let secret_store = SecretsStore::new(secrets_dir, Default::default(), db.clone())?;

        let mut runtime =
            Runtime::build(contract_store, delegate_store, secret_store, false).unwrap();
        runtime.set_state_store_db(db.clone());

        // Register the contract
        let contract_instance_id = ContractInstanceId::new([contract_id_byte; 32]);
        let contract_code = ContractCode::from(vec![contract_id_byte, 2, 3]);
        let contract_key =
            ContractKey::from_id_and_code(contract_instance_id, *contract_code.hash());
        runtime.contract_store.ensure_key_indexed(&contract_key)?;

        if let Some(state) = initial_state {
            db.store(contract_key, WrappedState::new(state.to_vec()))
                .await?;
        }

        // Load the V2 delegate
        let delegate = {
            let bytes = super::super::tests::get_test_module(TEST_DELEGATE_V2_CONTRACTS)?;
            DelegateContainer::Wasm(DelegateWasmAPIVersion::V1(Delegate::from((
                &bytes.into(),
                &vec![].into(),
            ))))
        };
        let _stored = runtime.delegate_store.store_delegate(delegate.clone());

        let key = XChaCha20Poly1305::generate_key(&mut OsRng);
        let cipher = XChaCha20Poly1305::new(&key);
        let nonce = XChaCha20Poly1305::generate_nonce(&mut OsRng);
        let _registered =
            runtime
                .secret_store
                .register_delegate(delegate.key().clone(), cipher, nonce);

        Ok((delegate, runtime, contract_instance_id, temp_dir))
    }

    /// Helper: send a message to the V2 delegate and deserialize the response.
    fn send_v2_message(
        runtime: &mut Runtime,
        delegate: &DelegateContainer,
        message: &v2_contracts_messages::InboundAppMessage,
    ) -> Result<v2_contracts_messages::OutboundAppMessage, Box<dyn std::error::Error>> {
        let _app_id = ContractInstanceId::new([1u8; 32]);
        let payload = bincode::serialize(message)?;
        let app_msg = ApplicationMessage::new(payload);

        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(app_msg)],
        )?;

        assert_eq!(outbound.len(), 1, "Expected exactly one outbound message");
        let response_msg = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => msg,
            other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage, got {:?}", other)
            }
        };
        assert!(response_msg.processed);

        Ok(bincode::deserialize(&response_msg.payload)?)
    }

    /// V2 E2E: PUT state via delegate, then GET it back.
    #[tokio::test(flavor = "multi_thread")]
    async fn test_v2_delegate_put_then_get() -> Result<(), Box<dyn std::error::Error>> {
        use v2_contracts_messages::*;

        let (delegate, mut runtime, contract_instance_id, _temp_dir) =
            setup_v2_runtime_with_contract(50, None).await?;
        let cid: [u8; 32] = contract_instance_id.as_bytes().try_into().unwrap();

        // PUT state
        let put_response = send_v2_message(
            &mut runtime,
            &delegate,
            &InboundAppMessage::PutContractState {
                contract_id: cid,
                state: vec![100, 200, 150],
            },
        )?;
        match put_response {
            OutboundAppMessage::Success { contract_id } => {
                assert_eq!(contract_id, cid);
            }
            other @ OutboundAppMessage::ContractState { .. }
            | other @ OutboundAppMessage::ContractNotFound { .. }
            | other @ OutboundAppMessage::Failed { .. } => {
                panic!("Expected Success from PUT, got {:?}", other)
            }
        }

        // GET it back
        let get_response = send_v2_message(
            &mut runtime,
            &delegate,
            &InboundAppMessage::GetContractState { contract_id: cid },
        )?;
        match get_response {
            OutboundAppMessage::ContractState { contract_id, state } => {
                assert_eq!(contract_id, cid);
                assert_eq!(state, vec![100, 200, 150]);
            }
            other @ OutboundAppMessage::ContractNotFound { .. }
            | other @ OutboundAppMessage::Success { .. }
            | other @ OutboundAppMessage::Failed { .. } => {
                panic!("Expected ContractState from GET, got {:?}", other)
            }
        }

        Ok(())
    }

    /// V2 E2E: UPDATE existing state via delegate.
    #[tokio::test(flavor = "multi_thread")]
    async fn test_v2_delegate_update_existing_state() -> Result<(), Box<dyn std::error::Error>> {
        use v2_contracts_messages::*;

        let (delegate, mut runtime, contract_instance_id, _temp_dir) =
            setup_v2_runtime_with_contract(51, Some(&[1, 2, 3])).await?;
        let cid: [u8; 32] = contract_instance_id.as_bytes().try_into().unwrap();

        // UPDATE the existing state
        let update_response = send_v2_message(
            &mut runtime,
            &delegate,
            &InboundAppMessage::UpdateContractState {
                contract_id: cid,
                state: vec![7, 8, 9],
            },
        )?;
        match update_response {
            OutboundAppMessage::Success { contract_id } => {
                assert_eq!(contract_id, cid);
            }
            other @ OutboundAppMessage::ContractState { .. }
            | other @ OutboundAppMessage::ContractNotFound { .. }
            | other @ OutboundAppMessage::Failed { .. } => {
                panic!("Expected Success from UPDATE, got {:?}", other)
            }
        }

        // Verify via GET
        let get_response = send_v2_message(
            &mut runtime,
            &delegate,
            &InboundAppMessage::GetContractState { contract_id: cid },
        )?;
        match get_response {
            OutboundAppMessage::ContractState { state, .. } => {
                assert_eq!(state, vec![7, 8, 9]);
            }
            other @ OutboundAppMessage::ContractNotFound { .. }
            | other @ OutboundAppMessage::Success { .. }
            | other @ OutboundAppMessage::Failed { .. } => {
                panic!("Expected ContractState, got {:?}", other)
            }
        }

        Ok(())
    }

    /// V2 E2E: UPDATE non-existent state returns error.
    #[tokio::test(flavor = "multi_thread")]
    async fn test_v2_delegate_update_nonexistent_fails() -> Result<(), Box<dyn std::error::Error>> {
        use v2_contracts_messages::*;

        let (delegate, mut runtime, contract_instance_id, _temp_dir) =
            setup_v2_runtime_with_contract(52, None).await?;
        let cid: [u8; 32] = contract_instance_id.as_bytes().try_into().unwrap();

        let response = send_v2_message(
            &mut runtime,
            &delegate,
            &InboundAppMessage::UpdateContractState {
                contract_id: cid,
                state: vec![1, 2, 3],
            },
        )?;
        match response {
            OutboundAppMessage::Failed { error_code, .. } => {
                assert!(
                    error_code < 0,
                    "Expected negative error code, got {error_code}"
                );
            }
            other @ OutboundAppMessage::ContractState { .. }
            | other @ OutboundAppMessage::ContractNotFound { .. }
            | other @ OutboundAppMessage::Success { .. } => panic!(
                "Expected Failed from UPDATE on non-existent, got {:?}",
                other
            ),
        }

        Ok(())
    }

    /// V2 E2E: SUBSCRIBE to a known contract succeeds.
    #[tokio::test(flavor = "multi_thread")]
    async fn test_v2_delegate_subscribe_known() -> Result<(), Box<dyn std::error::Error>> {
        use v2_contracts_messages::*;

        let (delegate, mut runtime, contract_instance_id, _temp_dir) =
            setup_v2_runtime_with_contract(53, Some(&[1])).await?;
        let cid: [u8; 32] = contract_instance_id.as_bytes().try_into().unwrap();

        let response = send_v2_message(
            &mut runtime,
            &delegate,
            &InboundAppMessage::SubscribeContract { contract_id: cid },
        )?;
        match response {
            OutboundAppMessage::Success { contract_id } => {
                assert_eq!(contract_id, cid);
            }
            other @ OutboundAppMessage::ContractState { .. }
            | other @ OutboundAppMessage::ContractNotFound { .. }
            | other @ OutboundAppMessage::Failed { .. } => {
                panic!("Expected Success from SUBSCRIBE, got {:?}", other)
            }
        }

        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_put_contract_request_response() -> Result<(), Box<dyn std::error::Error>> {
        use capabilities_messages::*;

        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_CAPABILITIES).await?;

        let _app_id = ContractInstanceId::new([1u8; 32]);

        let code = ContractCode::from(vec![0u8; 10]);
        let params = Parameters::from(vec![]);
        let wrapped = WrappedContract::new(Arc::new(code), params);
        let contract = ContractContainer::Wasm(ContractWasmAPIVersion::V1(wrapped));
        let contract_key = contract.key();

        let command = DelegateCommand::PutContractState {
            contract,
            state: vec![10, 20, 30],
        };
        let payload = bincode::serialize(&command)?;
        let app_msg = ApplicationMessage::new(payload);

        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(app_msg)],
        )?;

        assert_eq!(outbound.len(), 1, "Expected exactly one outbound message");
        let put_request = match &outbound[0] {
            OutboundDelegateMsg::PutContractRequest(req) => req.clone(),
            other @ OutboundDelegateMsg::ApplicationMessage(_)
            | other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected PutContractRequest, got {:?}", other)
            }
        };
        assert_eq!(put_request.contract.key(), contract_key);
        assert!(!put_request.processed);

        let response = InboundDelegateMsg::PutContractResponse(PutContractResponse {
            contract_id: *contract_key.id(),
            result: Ok(()),
            context: put_request.context.clone(),
        });

        let final_outbound =
            runtime.inbound_app_message(delegate.key(), &vec![].into(), None, vec![response])?;

        assert_eq!(
            final_outbound.len(),
            1,
            "Expected exactly one final message"
        );
        let final_msg = match &final_outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => msg,
            other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage, got {:?}", other)
            }
        };
        assert!(final_msg.processed);

        let response: DelegateResponse = bincode::deserialize(&final_msg.payload)?;
        match response {
            DelegateResponse::ContractPutResult { success, error, .. } => {
                assert!(success);
                assert!(error.is_none());
            }
            other @ DelegateResponse::ContractState { .. }
            | other @ DelegateResponse::MultipleContractStates { .. }
            | other @ DelegateResponse::Echo { .. }
            | other @ DelegateResponse::ContractUpdateResult { .. }
            | other @ DelegateResponse::ContractSubscribeResult { .. }
            | other @ DelegateResponse::ContractNotificationReceived { .. }
            | other @ DelegateResponse::Error { .. } => {
                panic!("Expected ContractPutResult, got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_update_contract_request_response() -> Result<(), Box<dyn std::error::Error>> {
        use capabilities_messages::*;

        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_CAPABILITIES).await?;

        let contract_id = ContractInstanceId::new([42u8; 32]);
        let _app_id = ContractInstanceId::new([1u8; 32]);

        let command = DelegateCommand::UpdateContractState {
            contract_id,
            state: vec![10, 20, 30],
        };
        let payload = bincode::serialize(&command)?;
        let app_msg = ApplicationMessage::new(payload);

        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(app_msg)],
        )?;

        assert_eq!(outbound.len(), 1, "Expected exactly one outbound message");
        let update_request = match &outbound[0] {
            OutboundDelegateMsg::UpdateContractRequest(req) => req.clone(),
            other @ OutboundDelegateMsg::ApplicationMessage(_)
            | other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected UpdateContractRequest, got {:?}", other)
            }
        };
        assert_eq!(update_request.contract_id, contract_id);
        assert!(!update_request.processed);

        let response = InboundDelegateMsg::UpdateContractResponse(UpdateContractResponse {
            contract_id,
            result: Ok(()),
            context: update_request.context.clone(),
        });

        let final_outbound =
            runtime.inbound_app_message(delegate.key(), &vec![].into(), None, vec![response])?;

        assert_eq!(
            final_outbound.len(),
            1,
            "Expected exactly one final message"
        );
        let final_msg = match &final_outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => msg,
            other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage, got {:?}", other)
            }
        };
        assert!(final_msg.processed);

        let response: DelegateResponse = bincode::deserialize(&final_msg.payload)?;
        match response {
            DelegateResponse::ContractUpdateResult {
                contract_id: id,
                success,
                error,
            } => {
                assert_eq!(id, contract_id);
                assert!(success);
                assert!(error.is_none());
            }
            other @ DelegateResponse::ContractState { .. }
            | other @ DelegateResponse::MultipleContractStates { .. }
            | other @ DelegateResponse::Echo { .. }
            | other @ DelegateResponse::ContractPutResult { .. }
            | other @ DelegateResponse::ContractSubscribeResult { .. }
            | other @ DelegateResponse::ContractNotificationReceived { .. }
            | other @ DelegateResponse::Error { .. } => {
                panic!("Expected ContractUpdateResult, got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_subscribe_contract_request_response() -> Result<(), Box<dyn std::error::Error>> {
        use capabilities_messages::*;

        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_CAPABILITIES).await?;

        let contract_id = ContractInstanceId::new([42u8; 32]);
        let _app_id = ContractInstanceId::new([1u8; 32]);

        let command = DelegateCommand::SubscribeContract { contract_id };
        let payload = bincode::serialize(&command)?;
        let app_msg = ApplicationMessage::new(payload);

        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(app_msg)],
        )?;

        assert_eq!(outbound.len(), 1, "Expected exactly one outbound message");
        let subscribe_request = match &outbound[0] {
            OutboundDelegateMsg::SubscribeContractRequest(req) => req.clone(),
            other @ OutboundDelegateMsg::ApplicationMessage(_)
            | other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected SubscribeContractRequest, got {:?}", other)
            }
        };
        assert_eq!(subscribe_request.contract_id, contract_id);
        assert!(!subscribe_request.processed);

        let response = InboundDelegateMsg::SubscribeContractResponse(SubscribeContractResponse {
            contract_id,
            result: Err("not yet implemented".to_string()),
            context: subscribe_request.context.clone(),
        });

        let final_outbound =
            runtime.inbound_app_message(delegate.key(), &vec![].into(), None, vec![response])?;

        assert_eq!(
            final_outbound.len(),
            1,
            "Expected exactly one final message"
        );
        let final_msg = match &final_outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => msg,
            other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage, got {:?}", other)
            }
        };
        assert!(final_msg.processed);

        let response: DelegateResponse = bincode::deserialize(&final_msg.payload)?;
        match response {
            DelegateResponse::ContractSubscribeResult {
                contract_id: id,
                success,
                error,
            } => {
                assert_eq!(id, contract_id);
                assert!(!success);
                assert!(error.is_some());
            }
            other @ DelegateResponse::ContractState { .. }
            | other @ DelegateResponse::MultipleContractStates { .. }
            | other @ DelegateResponse::Echo { .. }
            | other @ DelegateResponse::ContractPutResult { .. }
            | other @ DelegateResponse::ContractUpdateResult { .. }
            | other @ DelegateResponse::ContractNotificationReceived { .. }
            | other @ DelegateResponse::Error { .. } => {
                panic!("Expected ContractSubscribeResult, got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_contract_notification_delivered() -> Result<(), Box<dyn std::error::Error>> {
        use capabilities_messages::*;

        let (delegate, mut runtime, temp_dir) = setup_runtime(TEST_DELEGATE_CAPABILITIES).await?;

        let contract_id = ContractInstanceId::new([42u8; 32]);
        let new_state = vec![10, 20, 30, 40];

        let notification = InboundDelegateMsg::ContractNotification(ContractNotification {
            contract_id,
            new_state: WrappedState::new(new_state.clone()),
            context: DelegateContext::default(),
        });

        let outbound = runtime.inbound_app_message(
            delegate.key(),
            &vec![].into(),
            None,
            vec![notification],
        )?;

        assert_eq!(outbound.len(), 1, "Expected exactly one outbound message");
        let msg = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => msg,
            other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage, got {:?}", other)
            }
        };
        assert!(msg.processed);

        let response: DelegateResponse = bincode::deserialize(&msg.payload)?;
        match response {
            DelegateResponse::ContractNotificationReceived {
                contract_id: id,
                new_state: state,
            } => {
                assert_eq!(id, contract_id);
                assert_eq!(state, new_state);
            }
            other @ DelegateResponse::ContractState { .. }
            | other @ DelegateResponse::MultipleContractStates { .. }
            | other @ DelegateResponse::Echo { .. }
            | other @ DelegateResponse::ContractPutResult { .. }
            | other @ DelegateResponse::ContractUpdateResult { .. }
            | other @ DelegateResponse::ContractSubscribeResult { .. }
            | other @ DelegateResponse::Error { .. } => {
                panic!("Expected ContractNotificationReceived, got {:?}", other)
            }
        }

        std::mem::drop(temp_dir);
        Ok(())
    }

    /// End-to-end integration test: subscribe → registry populated → notification delivered.
    ///
    /// Verifies the full pipeline:
    /// 1. Delegate subscribes to a contract via SubscribeContractRequest
    /// 2. Subscription is registered in DELEGATE_SUBSCRIPTIONS
    /// 3. ContractNotification is delivered to the delegate
    /// 4. Delegate responds with ContractNotificationReceived
    /// 5. Cleanup: unregister delegate removes subscription entries
    #[tokio::test(flavor = "multi_thread")]
    async fn test_subscribe_then_notify_roundtrip() -> Result<(), Box<dyn std::error::Error>> {
        use crate::contract::storages::Storage;
        use crate::wasm_runtime::StateStorage;
        use capabilities_messages::*;

        let temp_dir = get_temp_dir();
        let contracts_dir = temp_dir.path().join("contracts");
        let delegates_dir = temp_dir.path().join("delegates");
        let secrets_dir = temp_dir.path().join("secrets");

        let db = Storage::new(temp_dir.path()).await?;
        let contract_store = ContractStore::new(contracts_dir, 10_000, db.clone())?;
        let delegate_store = DelegateStore::new(delegates_dir, 10_000, db.clone())?;
        let secret_store = SecretsStore::new(secrets_dir, Default::default(), db.clone())?;

        let mut runtime =
            Runtime::build(contract_store, delegate_store, secret_store, false).unwrap();
        runtime.set_state_store_db(db.clone());

        // Set up a contract so subscribe validation passes
        let contract_instance_id = ContractInstanceId::new([42u8; 32]);
        let contract_code = ContractCode::from(vec![42, 2, 3]);
        let contract_key =
            ContractKey::from_id_and_code(contract_instance_id, *contract_code.hash());
        runtime.contract_store.ensure_key_indexed(&contract_key)?;
        db.store(contract_key, WrappedState::new(vec![1, 2, 3]))
            .await?;

        // Load the delegate
        let delegate = {
            let bytes = super::super::tests::get_test_module(TEST_DELEGATE_CAPABILITIES)?;
            DelegateContainer::Wasm(DelegateWasmAPIVersion::V1(Delegate::from((
                &bytes.into(),
                &vec![].into(),
            ))))
        };
        let _stored = runtime.delegate_store.store_delegate(delegate.clone());

        let key = XChaCha20Poly1305::generate_key(&mut OsRng);
        let cipher = XChaCha20Poly1305::new(&key);
        let nonce = XChaCha20Poly1305::generate_nonce(&mut OsRng);
        let _registered =
            runtime
                .secret_store
                .register_delegate(delegate.key().clone(), cipher, nonce);

        let delegate_key = delegate.key().clone();
        let _app_id = ContractInstanceId::new([1u8; 32]);

        // --- Step 1: Delegate subscribes to the contract ---
        let subscribe_cmd = DelegateCommand::SubscribeContract {
            contract_id: contract_instance_id,
        };
        let payload = bincode::serialize(&subscribe_cmd)?;
        let app_msg = ApplicationMessage::new(payload);

        let outbound = runtime.inbound_app_message(
            &delegate_key,
            &vec![].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(app_msg)],
        )?;

        // Should emit SubscribeContractRequest
        assert_eq!(outbound.len(), 1);
        let subscribe_req = match &outbound[0] {
            OutboundDelegateMsg::SubscribeContractRequest(req) => req.clone(),
            other @ OutboundDelegateMsg::ApplicationMessage(_)
            | other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected SubscribeContractRequest, got {:?}", other)
            }
        };
        assert_eq!(subscribe_req.contract_id, contract_instance_id);

        // Simulate the V1 subscribe handler path (contract.rs:387-405):
        // validate contract existence via lookup, then register if found.
        let subscribe_result = if runtime
            .contract_store
            .code_hash_from_id(&subscribe_req.contract_id)
            .is_some()
        {
            crate::wasm_runtime::DELEGATE_SUBSCRIPTIONS
                .entry(subscribe_req.contract_id)
                .or_default()
                .insert(delegate_key.clone());
            Ok(())
        } else {
            Err("Contract not found".to_string())
        };
        assert!(
            subscribe_result.is_ok(),
            "Subscribe should succeed for known contract"
        );

        // Feed the SubscribeContractResponse back to the delegate
        let subscribe_response =
            InboundDelegateMsg::SubscribeContractResponse(SubscribeContractResponse {
                contract_id: subscribe_req.contract_id,
                result: subscribe_result,
                context: subscribe_req.context.clone(),
            });
        let outbound = runtime.inbound_app_message(
            &delegate_key,
            &vec![].into(),
            None,
            vec![subscribe_response],
        )?;
        // Delegate should emit a ContractSubscribeResult ApplicationMessage
        assert_eq!(outbound.len(), 1);
        match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => {
                let resp: DelegateResponse = bincode::deserialize(&msg.payload)?;
                match resp {
                    DelegateResponse::ContractSubscribeResult { success, .. } => {
                        assert!(success, "Subscribe response should indicate success");
                    }
                    other @ DelegateResponse::ContractState { .. }
                    | other @ DelegateResponse::MultipleContractStates { .. }
                    | other @ DelegateResponse::Echo { .. }
                    | other @ DelegateResponse::ContractPutResult { .. }
                    | other @ DelegateResponse::ContractUpdateResult { .. }
                    | other @ DelegateResponse::ContractNotificationReceived { .. }
                    | other @ DelegateResponse::Error { .. } => {
                        panic!("Expected ContractSubscribeResult, got {:?}", other)
                    }
                }
            }
            other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage, got {:?}", other)
            }
        }

        // --- Step 2: Verify registry is populated ---
        {
            let entry = crate::wasm_runtime::DELEGATE_SUBSCRIPTIONS.get(&contract_instance_id);
            let subscribers = entry.as_ref().unwrap();
            assert!(
                subscribers.contains(&delegate_key),
                "Delegate should be registered as subscriber"
            );
        }

        // Also verify that subscribing to an UNKNOWN contract fails validation
        let unknown_id = ContractInstanceId::new([99u8; 32]);
        let has_code = runtime
            .contract_store
            .code_hash_from_id(&unknown_id)
            .is_some();
        assert!(!has_code, "Unknown contract should not be in store");

        // --- Step 3: Deliver ContractNotification ---
        let updated_state = vec![10, 20, 30, 40, 50];
        let notification = InboundDelegateMsg::ContractNotification(ContractNotification {
            contract_id: contract_instance_id,
            new_state: WrappedState::new(updated_state.clone()),
            context: DelegateContext::default(),
        });

        let outbound =
            runtime.inbound_app_message(&delegate_key, &vec![].into(), None, vec![notification])?;

        // --- Step 4: Verify delegate responds correctly ---
        assert_eq!(outbound.len(), 1, "Expected one outbound from notification");
        let msg = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => msg,
            other @ OutboundDelegateMsg::RequestUserInput(_)
            | other @ OutboundDelegateMsg::ContextUpdated(_)
            | other @ OutboundDelegateMsg::GetContractRequest(_)
            | other @ OutboundDelegateMsg::PutContractRequest(_)
            | other @ OutboundDelegateMsg::UpdateContractRequest(_)
            | other @ OutboundDelegateMsg::SubscribeContractRequest(_)
            | other @ OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage, got {:?}", other)
            }
        };
        assert!(msg.processed);

        let response: DelegateResponse = bincode::deserialize(&msg.payload)?;
        match response {
            DelegateResponse::ContractNotificationReceived {
                contract_id: id,
                new_state: state,
            } => {
                assert_eq!(id, contract_instance_id);
                assert_eq!(state, updated_state);
            }
            other @ DelegateResponse::ContractState { .. }
            | other @ DelegateResponse::MultipleContractStates { .. }
            | other @ DelegateResponse::Echo { .. }
            | other @ DelegateResponse::ContractPutResult { .. }
            | other @ DelegateResponse::ContractUpdateResult { .. }
            | other @ DelegateResponse::ContractSubscribeResult { .. }
            | other @ DelegateResponse::Error { .. } => {
                panic!("Expected ContractNotificationReceived, got {:?}", other)
            }
        }

        // --- Step 5: Cleanup on delegate unregister ---
        crate::wasm_runtime::DELEGATE_SUBSCRIPTIONS.retain(|_, subscribers| {
            subscribers.remove(&delegate_key);
            !subscribers.is_empty()
        });

        // Verify cleanup
        let entry = crate::wasm_runtime::DELEGATE_SUBSCRIPTIONS.get(&contract_instance_id);
        assert!(
            entry.is_none() || entry.as_ref().unwrap().is_empty(),
            "Subscription should be cleaned up after delegate unregister"
        );

        std::mem::drop(temp_dir);
        Ok(())
    }

    /// Test: removing a contract cleans up DELEGATE_SUBSCRIPTIONS.
    #[tokio::test(flavor = "multi_thread")]
    async fn test_contract_removal_cleans_subscriptions() -> Result<(), Box<dyn std::error::Error>>
    {
        use crate::contract::storages::Storage;

        let temp_dir = get_temp_dir();
        let contracts_dir = temp_dir.path().join("contracts");
        let delegates_dir = temp_dir.path().join("delegates");
        let secrets_dir = temp_dir.path().join("secrets");

        let db = Storage::new(temp_dir.path()).await?;
        let contract_store = ContractStore::new(contracts_dir, 10_000, db.clone())?;
        let delegate_store = DelegateStore::new(delegates_dir, 10_000, db.clone())?;
        let secret_store = SecretsStore::new(secrets_dir, Default::default(), db.clone())?;

        let mut runtime =
            Runtime::build(contract_store, delegate_store, secret_store, false).unwrap();
        runtime.set_state_store_db(db.clone());

        // Create and store a contract
        let contract_instance_id = ContractInstanceId::new([99u8; 32]);
        let contract_code = ContractCode::from(vec![99, 2, 3]);
        let contract_key =
            ContractKey::from_id_and_code(contract_instance_id, *contract_code.hash());
        // Store the WASM file so remove_contract can delete it
        let wasm_path = runtime.contract_store.get_contract_path(&contract_key)?;
        std::fs::create_dir_all(wasm_path.parent().unwrap())?;
        std::fs::write(&wasm_path, [0u8; 10])?;
        runtime.contract_store.ensure_key_indexed(&contract_key)?;

        // Simulate delegate subscriptions
        let delegate_key_a = DelegateKey::new([1u8; 32], CodeHash::new([10u8; 32]));
        let delegate_key_b = DelegateKey::new([2u8; 32], CodeHash::new([20u8; 32]));
        {
            let mut entry = crate::wasm_runtime::DELEGATE_SUBSCRIPTIONS
                .entry(contract_instance_id)
                .or_default();
            entry.insert(delegate_key_a);
            entry.insert(delegate_key_b);
        }

        // Verify subscriptions exist
        assert!(
            crate::wasm_runtime::DELEGATE_SUBSCRIPTIONS
                .get(&contract_instance_id)
                .is_some()
        );

        // Remove the contract — should clean up subscriptions
        runtime.contract_store.remove_contract(&contract_key)?;

        // Verify subscriptions are cleaned up
        assert!(
            crate::wasm_runtime::DELEGATE_SUBSCRIPTIONS
                .get(&contract_instance_id)
                .is_none(),
            "DELEGATE_SUBSCRIPTIONS should be cleaned up when contract is removed"
        );

        std::mem::drop(temp_dir);
        Ok(())
    }

    // --- Delegate-to-delegate messaging tests ---

    const TEST_DELEGATE_MESSAGING: &str = "test_delegate_messaging";

    mod messaging_messages {
        use super::*;

        #[derive(Debug, Serialize, Deserialize)]
        pub enum InboundAppMessage {
            SendToDelegate {
                target_key_bytes: Vec<u8>,
                target_code_hash: Vec<u8>,
                payload: Vec<u8>,
            },
            Ping {
                data: Vec<u8>,
            },
        }

        #[derive(Debug, Serialize, Deserialize)]
        pub enum OutboundAppMessage {
            MessageSent,
            DelegateMessageReceived {
                sender_key_bytes: Vec<u8>,
                payload: Vec<u8>,
                /// Mirror of the same field in the WASM-side
                /// `OutboundAppMessage`; populated when the runtime delivered
                /// this message with `MessageOrigin::Delegate(k)` (#3860).
                origin_delegate_key_bytes: Option<Vec<u8>>,
            },
            PingResponse {
                data: Vec<u8>,
            },
        }
    }

    async fn setup_runtime_with_params(
        name: &str,
        params: Vec<u8>,
    ) -> Result<(DelegateContainer, Runtime, tempfile::TempDir), Box<dyn std::error::Error>> {
        use crate::contract::storages::Storage;
        let temp_dir = get_temp_dir();
        let contracts_dir = temp_dir.path().join("contracts");
        let delegates_dir = temp_dir.path().join("delegates");
        let secrets_dir = temp_dir.path().join("secrets");

        let db = Storage::new(temp_dir.path()).await?;
        let contract_store = ContractStore::new(contracts_dir, 10_000, db.clone())?;
        let delegate_store = DelegateStore::new(delegates_dir, 10_000, db.clone())?;
        let secret_store = SecretsStore::new(secrets_dir, Default::default(), db)?;

        let mut runtime =
            Runtime::build(contract_store, delegate_store, secret_store, false).unwrap();

        let delegate = {
            let bytes = super::super::tests::get_test_module(name)?;
            DelegateContainer::Wasm(DelegateWasmAPIVersion::V1(Delegate::from((
                &bytes.into(),
                &params.into(),
            ))))
        };
        let _stored = runtime.delegate_store.store_delegate(delegate.clone());

        let key = XChaCha20Poly1305::generate_key(&mut OsRng);
        let cipher = XChaCha20Poly1305::new(&key);
        let nonce = XChaCha20Poly1305::generate_nonce(&mut OsRng);
        let _registered =
            runtime
                .secret_store
                .register_delegate(delegate.key().clone(), cipher, nonce);

        Ok((delegate, runtime, temp_dir))
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_delegate_emits_send_delegate_message() -> Result<(), Box<dyn std::error::Error>> {
        use messaging_messages::*;

        let (delegate_a, mut runtime, _temp_dir) =
            setup_runtime_with_params(TEST_DELEGATE_MESSAGING, vec![1]).await?;
        let key_a = delegate_a.key().clone();

        // Create a fake target delegate key B
        let target_key_bytes = vec![42u8; 32];
        let target_code_hash = vec![99u8; 32];

        let _app_id = ContractInstanceId::new([1u8; 32]);
        let payload = bincode::serialize(&InboundAppMessage::SendToDelegate {
            target_key_bytes: target_key_bytes.clone(),
            target_code_hash: target_code_hash.clone(),
            payload: b"hello".to_vec(),
        })?;
        let app_msg = ApplicationMessage::new(payload);

        let outbound = runtime.inbound_app_message(
            &key_a,
            &vec![1u8].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(app_msg)],
        )?;

        // Should have SendDelegateMessage and ApplicationMessage(MessageSent)
        assert!(
            !outbound.is_empty(),
            "Expected at least 1 outbound message, got {}",
            outbound.len()
        );

        let send_msg = outbound
            .iter()
            .find_map(|m| match m {
                OutboundDelegateMsg::SendDelegateMessage(msg) => Some(msg),
                OutboundDelegateMsg::ApplicationMessage(_)
                | OutboundDelegateMsg::RequestUserInput(_)
                | OutboundDelegateMsg::ContextUpdated(_)
                | OutboundDelegateMsg::GetContractRequest(_)
                | OutboundDelegateMsg::PutContractRequest(_)
                | OutboundDelegateMsg::UpdateContractRequest(_)
                | OutboundDelegateMsg::SubscribeContractRequest(_) => None,
            })
            .expect("Expected SendDelegateMessage in outbound");

        // Verify target matches what we sent
        let mut expected_key = [0u8; 32];
        expected_key.copy_from_slice(&target_key_bytes);
        let mut expected_hash = [0u8; 32];
        expected_hash.copy_from_slice(&target_code_hash);
        assert_eq!(*send_msg.target, expected_key);
        assert_eq!(send_msg.target.code_hash(), &CodeHash::new(expected_hash));

        // Verify sender attestation: runtime overwrites sender with actual delegate key
        assert_eq!(
            send_msg.sender, key_a,
            "Sender should be attested as delegate A"
        );

        // Verify payload
        assert_eq!(send_msg.payload, b"hello");

        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_delegate_receives_delegate_message() -> Result<(), Box<dyn std::error::Error>> {
        use messaging_messages::*;

        let (delegate_b, mut runtime, _temp_dir) =
            setup_runtime_with_params(TEST_DELEGATE_MESSAGING, vec![2]).await?;
        let key_b = delegate_b.key().clone();

        // Create a fake sender key A
        let sender_key = DelegateKey::new([11u8; 32], CodeHash::new([22u8; 32]));

        // Deliver a DelegateMessage to B
        let delegate_msg =
            DelegateMessage::new(key_b.clone(), sender_key.clone(), b"hello".to_vec());

        let outbound = runtime.inbound_app_message(
            &key_b,
            &vec![2u8].into(),
            None,
            vec![InboundDelegateMsg::DelegateMessage(delegate_msg)],
        )?;

        assert_eq!(outbound.len(), 1, "Expected 1 outbound message");

        let app_msg = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => msg,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!("Expected ApplicationMessage, got {:?}", &outbound[0])
            }
        };

        let response: OutboundAppMessage = bincode::deserialize(&app_msg.payload)?;
        match response {
            OutboundAppMessage::DelegateMessageReceived {
                sender_key_bytes,
                payload,
                origin_delegate_key_bytes,
            } => {
                assert_eq!(sender_key_bytes, sender_key.bytes());
                assert_eq!(payload, b"hello");
                assert!(
                    origin_delegate_key_bytes.is_none(),
                    "origin was None, so receiver should see no Delegate origin"
                );
            }
            OutboundAppMessage::MessageSent | OutboundAppMessage::PingResponse { .. } => {
                panic!("Expected DelegateMessageReceived, got {:?}", response)
            }
        }

        Ok(())
    }

    /// Regression test for issue #3860: when the runtime delivers an inbound
    /// `DelegateMessage` with `Some(MessageOrigin::Delegate(caller_key))`, the
    /// receiving delegate's `process()` MUST see exactly that origin in its
    /// `origin` parameter. Previously the inter-delegate dispatch path passed
    /// `None`, leaving the receiver unable to authorize on caller identity.
    #[tokio::test(flavor = "multi_thread")]
    async fn test_inbound_app_message_propagates_delegate_origin()
    -> Result<(), Box<dyn std::error::Error>> {
        use messaging_messages::*;

        let (delegate_b, mut runtime, _temp_dir) =
            setup_runtime_with_params(TEST_DELEGATE_MESSAGING, vec![2]).await?;
        let key_b = delegate_b.key().clone();

        // Synthetic caller delegate A — its key is what we expect the
        // receiver to observe via MessageOrigin::Delegate.
        let caller_a = DelegateKey::new([0xA1u8; 32], CodeHash::new([0xA2u8; 32]));

        // Build an inbound DelegateMessage for B, with `sender` distinct from
        // `caller_a` so the test cannot pass by accident if the receiver
        // confuses `msg.sender` with the runtime-attested origin.
        let inband_sender = DelegateKey::new([0xBBu8; 32], CodeHash::new([0xCCu8; 32]));
        let delegate_msg =
            DelegateMessage::new(key_b.clone(), inband_sender, b"attest-me".to_vec());

        let origin = MessageOrigin::Delegate(caller_a.clone());
        let outbound = runtime.inbound_app_message(
            &key_b,
            &vec![2u8].into(),
            Some(&origin),
            vec![InboundDelegateMsg::DelegateMessage(delegate_msg)],
        )?;

        assert_eq!(outbound.len(), 1, "Expected exactly one outbound message");

        // Wildcard satisfies #[non_exhaustive] on OutboundDelegateMsg so
        // future stdlib variants don't break this test at compile time.
        #[allow(clippy::wildcard_enum_match_arm)]
        let app_msg = match &outbound[0] {
            OutboundDelegateMsg::ApplicationMessage(m) => m,
            other => panic!("Expected ApplicationMessage, got {other:?}"),
        };
        let response: OutboundAppMessage = bincode::deserialize(&app_msg.payload)?;
        // Wildcard satisfies #[non_exhaustive] on OutboundAppMessage so
        // future stdlib variants don't break this test at compile time.
        #[allow(clippy::wildcard_enum_match_arm)]
        match response {
            OutboundAppMessage::DelegateMessageReceived {
                origin_delegate_key_bytes,
                ..
            } => {
                let observed = origin_delegate_key_bytes
                    .expect("Receiver should see Some(MessageOrigin::Delegate(..))");
                assert_eq!(
                    observed,
                    caller_a.bytes(),
                    "Receiver must see the runtime-attested caller key, not msg.sender"
                );
            }
            other => panic!("Expected DelegateMessageReceived, got {other:?}"),
        }

        Ok(())
    }

    #[tokio::test(flavor = "multi_thread")]
    async fn test_delegate_to_delegate_roundtrip() -> Result<(), Box<dyn std::error::Error>> {
        use messaging_messages::*;

        // Load same code with different params → different keys
        let (delegate_a, mut runtime_a, _temp_dir_a) =
            setup_runtime_with_params(TEST_DELEGATE_MESSAGING, vec![1]).await?;
        let key_a = delegate_a.key().clone();

        let (delegate_b, mut runtime_b, _temp_dir_b) =
            setup_runtime_with_params(TEST_DELEGATE_MESSAGING, vec![2]).await?;
        let key_b = delegate_b.key().clone();

        // Step 1: Send command to A: "send a message to B"
        let _app_id = ContractInstanceId::new([1u8; 32]);
        let payload = bincode::serialize(&InboundAppMessage::SendToDelegate {
            target_key_bytes: key_b.bytes().to_vec(),
            target_code_hash: key_b.code_hash().as_ref().to_vec(),
            payload: b"inter-delegate".to_vec(),
        })?;
        let app_msg = ApplicationMessage::new(payload);

        let outbound_a = runtime_a.inbound_app_message(
            &key_a,
            &vec![1u8].into(),
            None,
            vec![InboundDelegateMsg::ApplicationMessage(app_msg)],
        )?;

        // Step 2: Extract the SendDelegateMessage from A's output
        let send_msg = outbound_a
            .iter()
            .find_map(|m| match m {
                OutboundDelegateMsg::SendDelegateMessage(msg) => Some(msg.clone()),
                OutboundDelegateMsg::ApplicationMessage(_)
                | OutboundDelegateMsg::RequestUserInput(_)
                | OutboundDelegateMsg::ContextUpdated(_)
                | OutboundDelegateMsg::GetContractRequest(_)
                | OutboundDelegateMsg::PutContractRequest(_)
                | OutboundDelegateMsg::UpdateContractRequest(_)
                | OutboundDelegateMsg::SubscribeContractRequest(_) => None,
            })
            .expect("Expected SendDelegateMessage from delegate A");

        assert_eq!(send_msg.sender, key_a, "Sender should be attested as A");
        assert_eq!(send_msg.payload, b"inter-delegate");

        // Step 3: Deliver to B as InboundDelegateMsg::DelegateMessage
        let outbound_b = runtime_b.inbound_app_message(
            &key_b,
            &vec![2u8].into(),
            None,
            vec![InboundDelegateMsg::DelegateMessage(send_msg)],
        )?;

        assert_eq!(outbound_b.len(), 1);

        let app_msg_b = match &outbound_b[0] {
            OutboundDelegateMsg::ApplicationMessage(msg) => msg,
            OutboundDelegateMsg::RequestUserInput(_)
            | OutboundDelegateMsg::ContextUpdated(_)
            | OutboundDelegateMsg::GetContractRequest(_)
            | OutboundDelegateMsg::PutContractRequest(_)
            | OutboundDelegateMsg::UpdateContractRequest(_)
            | OutboundDelegateMsg::SubscribeContractRequest(_)
            | OutboundDelegateMsg::SendDelegateMessage(_) => {
                panic!(
                    "Expected ApplicationMessage from B, got {:?}",
                    &outbound_b[0]
                )
            }
        };

        let response: OutboundAppMessage = bincode::deserialize(&app_msg_b.payload)?;
        match response {
            OutboundAppMessage::DelegateMessageReceived {
                sender_key_bytes,
                payload,
                origin_delegate_key_bytes,
            } => {
                assert_eq!(
                    sender_key_bytes,
                    key_a.bytes(),
                    "B should see A as the sender"
                );
                assert_eq!(payload, b"inter-delegate");
                // Origin not asserted here: this roundtrip test calls
                // inbound_app_message with origin=None (it bypasses the
                // executor that injects MessageOrigin::Delegate). End-to-end
                // propagation of MessageOrigin::Delegate through the WASM
                // boundary is covered by
                // `test_inbound_app_message_propagates_delegate_origin`.
                let _ = origin_delegate_key_bytes;
            }
            OutboundAppMessage::MessageSent | OutboundAppMessage::PingResponse { .. } => {
                panic!("Expected DelegateMessageReceived, got {:?}", response)
            }
        }

        Ok(())
    }

    /// Verify that when a delegate emits multiple SendDelegateMessage outbound,
    /// all of them get sender attestation (not just the first one).
    /// Regression test for PR #3282 review: drain(..) bypass.
    #[tokio::test(flavor = "multi_thread")]
    async fn test_multiple_send_delegate_messages_all_attested()
    -> Result<(), Box<dyn std::error::Error>> {
        let (delegate_a, mut runtime, _temp_dir) =
            setup_runtime_with_params(TEST_DELEGATE_MESSAGING, vec![1]).await?;
        let key_a = delegate_a.key().clone();

        // Create two different target keys
        let target_b = DelegateKey::new([42u8; 32], CodeHash::new([99u8; 32]));
        let target_c = DelegateKey::new([43u8; 32], CodeHash::new([98u8; 32]));

        // Send two messages via two separate inbound ApplicationMessages.
        // The first triggers SendDelegateMessage → break + drain, so
        // the second SendDelegateMessage goes through drain(..).
        let _app_id = ContractInstanceId::new([1u8; 32]);
        let payload1 =
            bincode::serialize(&messaging_messages::InboundAppMessage::SendToDelegate {
                target_key_bytes: target_b.bytes().to_vec(),
                target_code_hash: target_b.code_hash().as_ref().to_vec(),
                payload: b"msg1".to_vec(),
            })?;
        let payload2 =
            bincode::serialize(&messaging_messages::InboundAppMessage::SendToDelegate {
                target_key_bytes: target_c.bytes().to_vec(),
                target_code_hash: target_c.code_hash().as_ref().to_vec(),
                payload: b"msg2".to_vec(),
            })?;

        let outbound = runtime.inbound_app_message(
            &key_a,
            &vec![1u8].into(),
            None,
            vec![
                InboundDelegateMsg::ApplicationMessage(ApplicationMessage::new(payload1)),
                InboundDelegateMsg::ApplicationMessage(ApplicationMessage::new(payload2)),
            ],
        )?;

        // Collect all SendDelegateMessage from outbound
        let send_msgs: Vec<&DelegateMessage> = outbound
            .iter()
            .filter_map(|m| match m {
                OutboundDelegateMsg::SendDelegateMessage(msg) => Some(msg),
                OutboundDelegateMsg::ApplicationMessage(_)
                | OutboundDelegateMsg::RequestUserInput(_)
                | OutboundDelegateMsg::ContextUpdated(_)
                | OutboundDelegateMsg::GetContractRequest(_)
                | OutboundDelegateMsg::PutContractRequest(_)
                | OutboundDelegateMsg::UpdateContractRequest(_)
                | OutboundDelegateMsg::SubscribeContractRequest(_) => None,
            })
            .collect();

        // Should have at least 1 (the first triggers break+drain,
        // second may come through drain)
        assert!(
            !send_msgs.is_empty(),
            "Expected at least one SendDelegateMessage"
        );

        // ALL SendDelegateMessage must have sender attested as key_a
        for (i, msg) in send_msgs.iter().enumerate() {
            assert_eq!(
                msg.sender, key_a,
                "SendDelegateMessage[{i}] sender should be attested as delegate A, \
                 but got {:?}",
                msg.sender
            );
        }

        Ok(())
    }
}