use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use wasmtime::{AsContextMut, Linker, Memory, TypedFunc};

use crate::plugin::caller::CallerSession;
use crate::plugin::capabilities::{PluginCapability, is_free_import, requirement_for_import};

/// State stored in wasmtime's Store during plugin execution
pub struct PluginState {
    pub plugin_id: String,
    pub scope: String,
    pub secrets: HashMap<String, String>,
    pub config: serde_json::Value,
    pub db: Option<sqlx::AnyPool>,
    pub db_backend: crate::db::DatabaseBackend,
    pub http_client: reqwest::Client,
    pub lexicons: Arc<crate::catalog::LexiconCatalog>,
    pub usage: super::ResourceUsage,
    pub memory: Option<Memory>,
    pub alloc: Option<TypedFunc<u32, u32>>,
    pub dealloc: Option<TypedFunc<(u32, u32), ()>>,
    pub capabilities: std::collections::HashSet<crate::plugin::capabilities::PluginCapability>,
    pub allowed_hosts: Vec<String>,
    pub plugin_type: crate::plugin::PluginType,
    pub executor: Option<crate::plugin::PluginExecutor>,
    pub call_ctx: crate::plugin::library::LibraryCallContext,
    pub depth: u8,
    /// The credentials of the user whose script is running, when there are
    /// any. Every `host_caller_*` import refuses without it, except
    /// `host_caller_xrpc_query`, which a session-less query, record-event or
    /// label script can also reach.
    pub caller: Option<Arc<crate::plugin::caller::CallerSession>>,
    /// The script run this instance is executing, for the five imports that
    /// act on it. `None` for a library or auth instance, which is what makes
    /// those imports refuse rather than act on a run they are not inside.
    pub script_run: Option<super::script::ScriptRun>,
    /// The full instance, carried from `PluginExecutor::app_state` so a
    /// caller-acting import can run without a `CallerSession` to source it
    /// from. `Some` for every real instantiation; `None` only for the
    /// direct-construction test and external-auth call sites that never
    /// reach such an import.
    pub app_state: Option<crate::AppState>,
    /// Answers the preview-1 imports of a module that links them. Built for
    /// every instance, because the linker only reaches it through the store
    /// and a module that imports nothing from it never calls in.
    pub wasi: wasmtime_wasi::p1::WasiP1Ctx,
    /// The guest-execution budget the store's epoch deadline is armed from.
    pub deadline: crate::plugin::runtime::GuestDeadline,
    /// The store's linear-memory ceiling, installed as its resource limiter.
    pub limiter: crate::plugin::runtime::MemoryLimiter,
    /// Library calls started through `host_call_library_start` and not yet
    /// waited on. Dropped with the store, which aborts whatever is left.
    pub pending_calls: super::PendingCalls,
    /// The panic a guest reported before trapping, which the script runner
    /// answers the run with instead of the trap.
    pub panic: Option<happyview_plugin_sdk::wire::PanicReport>,
    /// The fuel each call into this instance starts with. The plugin type's
    /// default, except a compiled module's, which is its run's budget.
    pub call_fuel: u64,
}

/// Check that a memory access is within bounds
fn check_bounds(offset: usize, length: usize, mem_size: usize) -> Result<(usize, usize), ()> {
    if length == 0 {
        return Ok((offset, offset));
    }
    let end = offset.checked_add(length).ok_or(())?;
    if end > mem_size {
        return Err(());
    }
    Ok((offset, end))
}

/// Every host function except `host_log` starts here.
pub(super) fn require_capability(
    state: &PluginState,
    req: crate::plugin::capabilities::Requirement,
) -> Result<(), Vec<u8>> {
    if req.any_of.iter().any(|c| state.capabilities.contains(c)) {
        return Ok(());
    }
    let wanted: Vec<&str> = req.any_of.iter().map(|c| c.as_str()).collect();
    Err(error_envelope(
        "FORBIDDEN",
        format!(
            "plugin '{}' lacks the {} capability",
            state.plugin_id,
            wanted.join(" or ")
        ),
    ))
}

fn error_envelope(code: &str, message: impl std::fmt::Display) -> Vec<u8> {
    serde_json::to_vec(&serde_json::json!({
        "error": {"code": code, "message": message.to_string(), "retryable": false}
    }))
    .unwrap_or_default()
}

/// Register all host functions with the linker
pub fn register_host_functions(linker: &mut Linker<PluginState>) -> Result<(), wasmtime::Error> {
    define_host_functions(linker).map(|_| ())
}

/// Registers every async import through [`HostImports::define`], so none
/// can be added without the guest-deadline accounting, and returns the
/// names that went through it. `host_log` and `host_script_panic` are
/// synchronous and copy a few bytes; they are registered bare.
pub(crate) fn define_host_functions(
    linker: &mut Linker<PluginState>,
) -> Result<Vec<&'static str>, wasmtime::Error> {
    // Sync functions
    linker.func_wrap("env", "host_log", host_log)?;
    linker.func_wrap("env", "host_script_panic", host_script_panic)?;
    let mut imports = HostImports::new(linker);

    // host_get_secret must be async because it calls alloc on an async store
    imports.define(
        "host_get_secret",
        |caller, (name_ptr, name_len): (i32, i32)| {
            Box::pin(async move { host_get_secret_impl(caller, name_ptr, name_len).await })
        },
    )?;

    // Async functions - HTTP
    imports.define(
        "host_http_request",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move { host_http_request_impl(caller, req_ptr, req_len).await })
        },
    )?;

    // Async functions - KV
    imports.define("host_kv_get", |caller, (key_ptr, key_len): (i32, i32)| {
        Box::pin(async move { host_kv_get_impl(caller, key_ptr, key_len).await })
    })?;

    imports.define(
        "host_kv_set",
        |caller, (key_ptr, key_len, val_ptr, val_len, ttl): (i32, i32, i32, i32, i32)| {
            Box::pin(async move {
                host_kv_set_impl(caller, key_ptr, key_len, val_ptr, val_len, ttl).await
            })
        },
    )?;

    imports.define(
        "host_kv_delete",
        |caller, (key_ptr, key_len): (i32, i32)| {
            Box::pin(async move { host_kv_delete_impl(caller, key_ptr, key_len).await })
        },
    )?;

    // Async functions - Record lookup
    imports.define(
        "host_lookup_record",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move { host_lookup_record_impl(caller, req_ptr, req_len).await })
        },
    )?;

    imports.define(
        "host_records_query",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_spec_impl(
                    caller,
                    "host_records_query",
                    req_ptr,
                    req_len,
                    |db, backend, spec| async move { super::records_query(&db, backend, spec).await },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_records_count",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_spec_impl(
                    caller,
                    "host_records_count",
                    req_ptr,
                    req_len,
                    |db, backend, spec| async move { super::records_count(&db, backend, spec).await },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_records_get",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_spec_impl(
                    caller,
                    "host_records_get",
                    req_ptr,
                    req_len,
                    |db, backend, spec: GetSpec| async move {
                        super::records_get(&db, backend, &spec.uri).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_records_search",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_spec_impl(
                    caller,
                    "host_records_search",
                    req_ptr,
                    req_len,
                    |db, backend, spec| async move { super::records_search(&db, backend, spec).await },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_table_query",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_spec_impl(
                    caller,
                    "host_table_query",
                    req_ptr,
                    req_len,
                    |db, backend, spec| async move { super::table_query(&db, backend, spec).await },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_backlinks_query",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_spec_impl(
                    caller,
                    "host_backlinks_query",
                    req_ptr,
                    req_len,
                    |db, backend, spec| async move { super::backlinks_query(&db, backend, spec).await },
                )
                .await
            })
        },
    )?;

    // Async functions - acting as the calling user
    imports.define(
        "host_caller_create_record",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_caller_impl(
                    caller,
                    "host_caller_create_record",
                    req_ptr,
                    req_len,
                    |session, spec| async move { super::create_record(&session, spec).await },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_caller_put_record",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_caller_impl(
                    caller,
                    "host_caller_put_record",
                    req_ptr,
                    req_len,
                    |session, spec| async move { super::put_record(&session, spec).await },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_caller_delete_record",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_caller_impl(
                    caller,
                    "host_caller_delete_record",
                    req_ptr,
                    req_len,
                    |session, spec| async move { super::delete_record(&session, spec).await },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_caller_upload_blob",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_caller_impl(
                    caller,
                    "host_caller_upload_blob",
                    req_ptr,
                    req_len,
                    |session, spec| async move { super::upload_blob(&session, spec).await },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_caller_xrpc_query",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move { host_caller_query_impl(caller, req_ptr, req_len).await })
        },
    )?;

    imports.define(
        "host_caller_xrpc_procedure",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_caller_impl(
                    caller,
                    "host_caller_xrpc_procedure",
                    req_ptr,
                    req_len,
                    |session, spec| async move { super::xrpc_procedure(&session, spec).await },
                )
                .await
            })
        },
    )?;

    // Async functions - writing to a linked repo and enqueuing jobs
    imports.define(
        "host_linked_repos_list",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_impl(
                    caller,
                    "host_linked_repos_list",
                    req_ptr,
                    req_len,
                    super::linked_repos::LinkedRepoError::code,
                    |state, _spec: serde_json::Value| async move {
                        super::linked_repos::list(&state).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_linked_repo_create_record",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_impl(
                    caller,
                    "host_linked_repo_create_record",
                    req_ptr,
                    req_len,
                    super::linked_repos::LinkedRepoError::code,
                    |state, spec: happyview_plugin_sdk::wire::LinkedRepoRecordCreate| async move {
                        super::linked_repos::create_record(&state, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_linked_repo_put_record",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_impl(
                    caller,
                    "host_linked_repo_put_record",
                    req_ptr,
                    req_len,
                    super::linked_repos::LinkedRepoError::code,
                    |state, spec: happyview_plugin_sdk::wire::LinkedRepoRecordPut| async move {
                        super::linked_repos::put_record(&state, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_linked_repo_delete_record",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_impl(
                    caller,
                    "host_linked_repo_delete_record",
                    req_ptr,
                    req_len,
                    super::linked_repos::LinkedRepoError::code,
                    |state, spec: happyview_plugin_sdk::wire::LinkedRepoRecordDelete| async move {
                        super::linked_repos::delete_record(&state, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_linked_repo_upload_blob",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_impl(
                    caller,
                    "host_linked_repo_upload_blob",
                    req_ptr,
                    req_len,
                    super::linked_repos::LinkedRepoError::code,
                    |state, spec: happyview_plugin_sdk::wire::LinkedRepoBlobUpload| async move {
                        super::linked_repos::upload_blob(&state, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_linked_repo_call",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_impl(
                    caller,
                    "host_linked_repo_call",
                    req_ptr,
                    req_len,
                    super::linked_repos::LinkedRepoError::code,
                    |state, spec: happyview_plugin_sdk::wire::LinkedRepoCall| async move {
                        super::linked_repos::call(&state, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_jobs_create",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_caller_impl(
                    caller,
                    "host_jobs_create",
                    req_ptr,
                    req_len,
                    super::jobs::JobsError::code,
                    |state, caller_did, session, spec: happyview_plugin_sdk::wire::JobCreate| async move {
                        super::jobs::create(&state, caller_did.as_deref(), session.as_deref(), spec)
                            .await
                    },
                )
                .await
            })
        },
    )?;

    // Reads of background jobs. `host_jobs_get` takes the caller DID from the
    // call context (a plugin cannot set it) and hides other users' jobs; the
    // `_any` pair need no caller, only the `jobs:read_any` capability.
    imports.define("host_jobs_get", |caller, (req_ptr, req_len): (i32, i32)| {
        Box::pin(async move {
            host_app_caller_impl(
                caller,
                "host_jobs_get",
                req_ptr,
                req_len,
                super::jobs::JobsError::code,
                |state, caller_did, _session, spec: happyview_plugin_sdk::wire::JobGet| async move {
                    super::jobs::get(&state, caller_did.as_deref(), spec).await
                },
            )
            .await
        })
    })?;

    imports.define(
        "host_jobs_get_any",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_impl(
                    caller,
                    "host_jobs_get_any",
                    req_ptr,
                    req_len,
                    super::jobs::JobsError::code,
                    |state, spec: happyview_plugin_sdk::wire::JobGet| async move {
                        super::jobs::get_any(&state, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_jobs_list_any",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_impl(
                    caller,
                    "host_jobs_list_any",
                    req_ptr,
                    req_len,
                    super::jobs::JobsError::code,
                    |state, spec: happyview_plugin_sdk::wire::JobListAny| async move {
                        super::jobs::list_any(&state, spec).await
                    },
                )
                .await
            })
        },
    )?;

    // Async functions - spaces. Four reads need no caller; the eleven writes
    // act as `ctx.caller_did` and refuse before reaching the module when the
    // script context has none, via `require_spaces_caller`.
    imports.define(
        "host_spaces_info",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_impl(
                    caller,
                    "host_spaces_info",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state, spec: happyview_plugin_sdk::wire::SpacesInfo| async move {
                        super::spaces::info(&state, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_spaces_query",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_impl(
                    caller,
                    "host_spaces_query",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state, spec: happyview_plugin_sdk::wire::SpacesQuery| async move {
                        super::spaces::query(&state, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_spaces_members",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_impl(
                    caller,
                    "host_spaces_members",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state, spec: happyview_plugin_sdk::wire::SpacesMembers| async move {
                        super::spaces::members(&state, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_spaces_access",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_impl(
                    caller,
                    "host_spaces_access",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state, spec: happyview_plugin_sdk::wire::SpacesAccess| async move {
                        super::spaces::access(&state, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_spaces_create",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_caller_impl(
                    caller,
                    "host_spaces_create",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state,
                     caller_did,
                     _session,
                     spec: happyview_plugin_sdk::wire::SpacesCreate| async move {
                        let caller_did = require_spaces_caller(caller_did)?;
                        super::spaces::create(&state, &caller_did, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_spaces_accept_invite",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_caller_impl(
                    caller,
                    "host_spaces_accept_invite",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state,
                     caller_did,
                     _session,
                     spec: happyview_plugin_sdk::wire::SpacesAcceptInvite| async move {
                        let caller_did = require_spaces_caller(caller_did)?;
                        super::spaces::accept_invite(&state, &caller_did, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_spaces_write_record",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_caller_impl(
                    caller,
                    "host_spaces_write_record",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state,
                     caller_did,
                     _session,
                     spec: happyview_plugin_sdk::wire::SpaceRecordWrite| async move {
                        let caller_did = require_spaces_caller(caller_did)?;
                        super::spaces::write_record(&state, &caller_did, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_spaces_put_record",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_caller_impl(
                    caller,
                    "host_spaces_put_record",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state,
                     caller_did,
                     _session,
                     spec: happyview_plugin_sdk::wire::SpaceRecordPut| async move {
                        let caller_did = require_spaces_caller(caller_did)?;
                        super::spaces::put_record(&state, &caller_did, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_spaces_delete_record",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_caller_impl(
                    caller,
                    "host_spaces_delete_record",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state,
                     caller_did,
                     _session,
                     spec: happyview_plugin_sdk::wire::SpaceRecordDelete| async move {
                        let caller_did = require_spaces_caller(caller_did)?;
                        super::spaces::delete_record(&state, &caller_did, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_spaces_add_member",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_caller_impl(
                    caller,
                    "host_spaces_add_member",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state,
                     caller_did,
                     _session,
                     spec: happyview_plugin_sdk::wire::SpaceMemberAdd| async move {
                        let caller_did = require_spaces_caller(caller_did)?;
                        super::spaces::add_member(&state, &caller_did, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_spaces_set_member",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_caller_impl(
                    caller,
                    "host_spaces_set_member",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state,
                     caller_did,
                     _session,
                     spec: happyview_plugin_sdk::wire::SpaceMemberAdd| async move {
                        let caller_did = require_spaces_caller(caller_did)?;
                        super::spaces::set_member(&state, &caller_did, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_spaces_remove_member",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_caller_impl(
                    caller,
                    "host_spaces_remove_member",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state,
                     caller_did,
                     _session,
                     spec: happyview_plugin_sdk::wire::SpaceMemberRemove| async move {
                        let caller_did = require_spaces_caller(caller_did)?;
                        super::spaces::remove_member(&state, &caller_did, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_spaces_update",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_caller_impl(
                    caller,
                    "host_spaces_update",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state, caller_did, _session, spec: happyview_plugin_sdk::wire::SpaceUpdate| async move {
                        let caller_did = require_spaces_caller(caller_did)?;
                        super::spaces::update(&state, &caller_did, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_spaces_delete",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_caller_impl(
                    caller,
                    "host_spaces_delete",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state, caller_did, _session, spec: happyview_plugin_sdk::wire::SpaceDelete| async move {
                        let caller_did = require_spaces_caller(caller_did)?;
                        super::spaces::delete(&state, &caller_did, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_spaces_create_invite",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_app_caller_impl(
                    caller,
                    "host_spaces_create_invite",
                    req_ptr,
                    req_len,
                    super::spaces::SpacesError::code,
                    |state,
                     caller_did,
                     _session,
                     spec: happyview_plugin_sdk::wire::SpaceInviteCreate| async move {
                        let caller_did = require_spaces_caller(caller_did)?;
                        super::spaces::create_invite(&state, &caller_did, spec).await
                    },
                )
                .await
            })
        },
    )?;

    // Async functions - local index writes and lexicon reads
    imports.define(
        "host_records_index_put",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                // Plugin unit tests build no app state: their own pool then.
                let shared = shared_records(caller.data());
                host_spec_impl(
                    caller,
                    "host_records_index_put",
                    req_ptr,
                    req_len,
                    |db, backend, spec| async move {
                        let store = shared
                            .unwrap_or_else(|| crate::record_store::RecordStore::new(db, backend));
                        super::index_put(&store, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_records_index_delete",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                // Plugin unit tests build no app state: their own pool then.
                let shared = shared_records(caller.data());
                host_spec_impl(
                    caller,
                    "host_records_index_delete",
                    req_ptr,
                    req_len,
                    |db, backend, spec: happyview_plugin_sdk::wire::IndexDelete| async move {
                        let store = shared
                            .unwrap_or_else(|| crate::record_store::RecordStore::new(db, backend));
                        super::index_delete(&store, &spec.uri).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_lexicon_get",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move { host_lexicon_get_impl(caller, req_ptr, req_len).await })
        },
    )?;

    imports.define(
        "host_allowed_hosts",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move { host_allowed_hosts_impl(caller, req_ptr, req_len).await })
        },
    )?;

    imports.define(
        "host_call_library",
        |caller,
         (lib_ptr, lib_len, fn_ptr, fn_len, args_ptr, args_len): (i32, i32, i32, i32, i32, i32)| {
            Box::pin(async move {
                host_call_library_impl(caller, lib_ptr, lib_len, fn_ptr, fn_len, args_ptr, args_len)
                    .await
            })
        },
    )?;

    imports.define(
        "host_call_library_start",
        |caller,
         (lib_ptr, lib_len, fn_ptr, fn_len, args_ptr, args_len): (i32, i32, i32, i32, i32, i32)| {
            Box::pin(async move {
                host_call_library_start_impl(
                    caller, lib_ptr, lib_len, fn_ptr, fn_len, args_ptr, args_len,
                )
            })
        },
    )?;

    imports.define(
        "host_call_library_wait_any",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move { host_call_library_wait_any_impl(caller, req_ptr, req_len).await })
        },
    )?;

    imports.define(
        "host_get_api_surface",
        |caller, (lib_ptr, lib_len): (i32, i32)| {
            Box::pin(async move { host_get_api_surface_impl(caller, lib_ptr, lib_len).await })
        },
    )?;

    imports.define(
        "host_db_query",
        |caller, (sql_ptr, sql_len, params_ptr, params_len): (i32, i32, i32, i32)| {
            Box::pin(async move {
                host_db_impl(caller, false, sql_ptr, sql_len, params_ptr, params_len).await
            })
        },
    )?;
    imports.define(
        "host_db_execute",
        |caller, (sql_ptr, sql_len, params_ptr, params_len): (i32, i32, i32, i32)| {
            Box::pin(async move {
                host_db_impl(caller, true, sql_ptr, sql_len, params_ptr, params_len).await
            })
        },
    )?;

    // Async functions - AT Protocol network reads and attestation
    imports.define("host_blob_put", |caller, (req_ptr, req_len): (i32, i32)| {
        Box::pin(async move { host_blob_put_impl(caller, req_ptr, req_len).await })
    })?;

    imports.define("host_blob_get", |caller, (req_ptr, req_len): (i32, i32)| {
        Box::pin(async move { host_blob_get_impl(caller, req_ptr, req_len).await })
    })?;

    imports.define(
        "host_blob_stat",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move { host_blob_stat_impl(caller, req_ptr, req_len).await })
        },
    )?;

    imports.define(
        "host_atproto_resolve_service",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(
                async move { host_atproto_resolve_service_impl(caller, req_ptr, req_len).await },
            )
        },
    )?;

    imports.define(
        "host_atproto_resolve_identity",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(
                async move { host_atproto_resolve_identity_impl(caller, req_ptr, req_len).await },
            )
        },
    )?;

    imports.define(
        "host_atproto_blob_download",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move { host_atproto_blob_download_impl(caller, req_ptr, req_len).await })
        },
    )?;

    imports.define(
        "host_labels_get",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move { host_labels_get_impl(caller, req_ptr, req_len).await })
        },
    )?;

    imports.define(
        "host_attest_sign",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move { host_attest_sign_impl(caller, req_ptr, req_len).await })
        },
    )?;

    imports.define(
        "host_attest_verify",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move { host_attest_verify_impl(caller, req_ptr, req_len).await })
        },
    )?;

    // The imports only an interpreter reaches: they act on the script run in
    // progress, so each refuses outside one.
    imports.define(
        "host_script_log",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_script_impl(
                    caller,
                    "host_script_log",
                    req_ptr,
                    req_len,
                    |state, run, spec: happyview_plugin_sdk::wire::ScriptLogRequest| async move {
                        super::script::log(&state, &run, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_job_progress",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_script_impl(
                    caller,
                    "host_job_progress",
                    req_ptr,
                    req_len,
                    |state, run, spec: happyview_plugin_sdk::wire::JobProgressRequest| async move {
                        super::script::progress(&state, &run, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_job_should_stop",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_script_impl(
                    caller,
                    "host_job_should_stop",
                    req_ptr,
                    req_len,
                    |state,
                     run,
                     _spec: happyview_plugin_sdk::wire::JobShouldStopRequest| async move {
                        super::script::should_stop(&state, &run).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define("host_job_wait", |caller, (req_ptr, req_len): (i32, i32)| {
        Box::pin(async move {
            host_script_impl(
                caller,
                "host_job_wait",
                req_ptr,
                req_len,
                |state, run, spec: happyview_plugin_sdk::wire::JobWaitRequest| async move {
                    super::script::wait(&state, &run, spec).await
                },
            )
            .await
        })
    })?;

    imports.define(
        "host_module_resolve",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_script_impl(
                    caller,
                    "host_module_resolve",
                    req_ptr,
                    req_len,
                    |_state, run, spec: happyview_plugin_sdk::wire::ModuleResolve| async move {
                        super::script::module_resolve(&run, spec)
                    },
                )
                .await
            })
        },
    )?;

    // The test runner's imports. Each answers only inside a test file's run
    // on a state the test command built.
    imports.define(
        "host_test_report",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_test_impl(
                    caller,
                    "host_test_report",
                    req_ptr,
                    req_len,
                    |_state, harness, report: crate::plugin::TestReport| async move {
                        harness.report(report);
                        Ok(())
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_test_trigger",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_test_impl(
                    caller,
                    "host_test_trigger",
                    req_ptr,
                    req_len,
                    |state, harness, spec: crate::plugin::TestTrigger| async move {
                        crate::testing::triggers::fire(&state, &harness, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_test_mock",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_test_impl(
                    caller,
                    "host_test_mock",
                    req_ptr,
                    req_len,
                    |state, harness, spec: crate::plugin::TestMock| async move {
                        crate::testing::mocks::mock(&state, &harness, spec).await
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_test_calls",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_test_impl(
                    caller,
                    "host_test_calls",
                    req_ptr,
                    req_len,
                    |_state, harness, spec: crate::plugin::TestCalls| async move {
                        Ok(harness.calls(&spec.namespace))
                    },
                )
                .await
            })
        },
    )?;

    imports.define(
        "host_test_pds_writes",
        |caller, (req_ptr, req_len): (i32, i32)| {
            Box::pin(async move {
                host_test_impl(
                    caller,
                    "host_test_pds_writes",
                    req_ptr,
                    req_len,
                    |_state, harness, _spec: crate::plugin::TestPdsWritesRequest| async move {
                        Ok(harness.pds().writes())
                    },
                )
                .await
            })
        },
    )?;

    Ok(imports.wrapped)
}

pub(crate) struct HostImports<'l> {
    linker: &'l mut Linker<PluginState>,
    wrapped: Vec<&'static str>,
}

impl<'l> HostImports<'l> {
    pub(crate) fn new(linker: &'l mut Linker<PluginState>) -> Self {
        Self {
            linker,
            wrapped: Vec::new(),
        }
    }

    /// One `env` import. The guest's deadline stops when it enters and
    /// starts again when it returns, so an import's own latency is never
    /// charged to the script.
    pub(crate) fn define<Params, R, F>(
        &mut self,
        name: &'static str,
        f: F,
    ) -> Result<(), wasmtime::Error>
    where
        Params: wasmtime::WasmTyList + Send + 'static,
        R: wasmtime::WasmRet + Send + 'static,
        F: for<'a, 'c> Fn(
                &'a mut wasmtime::Caller<'c, PluginState>,
                Params,
            ) -> Pin<Box<dyn Future<Output = R> + Send + 'a>>
            + Copy
            + Send
            + Sync
            + 'static,
    {
        self.wrapped.push(name);
        self.linker.func_wrap_async(
            "env",
            name,
            move |mut caller: wasmtime::Caller<'_, PluginState>, params: Params| {
                Box::new(async move {
                    suspend_guest_deadline(&mut caller);
                    let out = f(&mut caller, params).await;
                    resume_guest_deadline(&mut caller);
                    out
                })
            },
        )?;
        Ok(())
    }
}

/// The guest has entered a host import: stop its clock.
fn suspend_guest_deadline(caller: &mut wasmtime::Caller<'_, PluginState>) {
    caller.data_mut().deadline.suspend();
}

/// The guest is about to run again: re-arm the store with what is left.
fn resume_guest_deadline(caller: &mut wasmtime::Caller<'_, PluginState>) {
    let remaining = caller.data_mut().deadline.resume();
    caller.as_context_mut().set_epoch_deadline(remaining);
}

/// One entry point for the five imports that act on the script run in
/// progress. The capability is checked before the run, so a plugin that never
/// asked for `script:host` hears about the grant rather than about a run it
/// could not have reached anyway.
async fn host_script_impl<S, R, F, Fut>(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    import: &'static str,
    req_ptr: i32,
    req_len: i32,
    run: F,
) -> i64
where
    S: serde::de::DeserializeOwned,
    R: serde::Serialize,
    F: FnOnce(crate::AppState, super::script::ScriptRun, S) -> Fut,
    Fut: std::future::Future<Output = Result<R, super::script::ScriptHostError>>,
{
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(import).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let Some(script_run) = caller.data().script_run.clone() else {
        return write_guest_response(
            caller,
            &error_envelope(
                "UNSUPPORTED",
                format!("{import} acts on a script run, and no script run is in progress"),
            ),
        )
        .await;
    };
    let Some(app_state) = caller.data().app_state.clone() else {
        return write_guest_response(
            caller,
            &error_envelope("HOST_ERROR", "this instance has no app state"),
        )
        .await;
    };
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let spec: S = match serde_json::from_slice(&bytes) {
        Ok(s) => s,
        Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
    };
    let response = match run(app_state, script_run, spec).await {
        Ok(value) => serde_json::to_vec(&serde_json::json!({"ok": value})).unwrap_or_default(),
        Err(e) => error_envelope(e.code(), e),
    };
    write_guest_response(caller, &response).await
}

/// One entry point for the test runner's imports. Both the run and the state
/// have to say "test": a test-kind input reaching a serving instance's state,
/// or a handler's run on a test state, is refused alike.
async fn host_test_impl<S, R, F, Fut>(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    import: &'static str,
    req_ptr: i32,
    req_len: i32,
    run: F,
) -> i64
where
    S: serde::de::DeserializeOwned,
    R: serde::Serialize,
    F: FnOnce(crate::AppState, Arc<crate::testing::TestHarness>, S) -> Fut,
    Fut: std::future::Future<Output = Result<R, crate::testing::TestHostError>>,
{
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(import).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let in_test_file = caller
        .data()
        .script_run
        .as_ref()
        .is_some_and(|run| run.test);
    let app_state = caller.data().app_state.clone();
    let harness = app_state.as_ref().and_then(|state| state.test.clone());
    let (true, Some(app_state), Some(harness)) = (in_test_file, app_state, harness) else {
        return write_guest_response(
            caller,
            &error_envelope(
                "UNSUPPORTED",
                format!("{import} is available only to a test file run by `happyview test`"),
            ),
        )
        .await;
    };
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let spec: S = match serde_json::from_slice(&bytes) {
        Ok(s) => s,
        Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
    };
    let response = match run(app_state, harness, spec).await {
        Ok(value) => serde_json::to_vec(&serde_json::json!({"ok": value})).unwrap_or_default(),
        Err(e) => error_envelope(e.code(), e),
    };
    write_guest_response(caller, &response).await
}

/// Read a string from guest memory
fn read_guest_string(
    caller: &wasmtime::Caller<'_, PluginState>,
    ptr: i32,
    len: i32,
) -> Option<String> {
    let memory = caller.data().memory?;
    let mem_data = memory.data(caller);
    let (start, end) = check_bounds(ptr as usize, len as usize, mem_data.len()).ok()?;
    std::str::from_utf8(&mem_data[start..end])
        .ok()
        .map(String::from)
}

/// Read raw bytes from guest memory
fn read_guest_bytes(
    caller: &wasmtime::Caller<'_, PluginState>,
    ptr: i32,
    len: i32,
) -> Option<Vec<u8>> {
    let memory = caller.data().memory?;
    let mem_data = memory.data(caller);
    let (start, end) = check_bounds(ptr as usize, len as usize, mem_data.len()).ok()?;
    Some(mem_data[start..end].to_vec())
}

/// Write response data to guest memory, returning packed (ptr << 32) | len
async fn write_guest_response(caller: &mut wasmtime::Caller<'_, PluginState>, data: &[u8]) -> i64 {
    let memory = match caller.data().memory {
        Some(m) => m,
        None => return 0,
    };
    let alloc = match &caller.data().alloc {
        Some(a) => a.clone(),
        None => return 0,
    };

    // The allocator is guest code: the clock has to be running again, and
    // armed from what remains, before it is entered.
    resume_guest_deadline(caller);
    crate::plugin::runtime::grant_fuel_for_bytes(caller, data.len());
    let len = data.len() as u32;
    let ptr = match alloc.call_async(&mut *caller, len).await {
        Ok(p) if p != 0 => p,
        _ => return 0,
    };

    let mem_data = memory.data_mut(caller);
    if check_bounds(ptr as usize, len as usize, mem_data.len()).is_err() {
        return 0;
    }

    mem_data[ptr as usize..(ptr as usize + len as usize)].copy_from_slice(data);
    ((ptr as i64) << 32) | (len as i64)
}

/// Host function: log a message from the plugin
fn host_log(
    caller: wasmtime::Caller<'_, PluginState>,
    level_ptr: i32,
    level_len: i32,
    msg_ptr: i32,
    msg_len: i32,
) {
    let memory = match caller.data().memory {
        Some(m) => m,
        None => return,
    };

    let mem_data = memory.data(&caller);
    let mem_size = mem_data.len();

    let (level_start, level_end) =
        match check_bounds(level_ptr as usize, level_len as usize, mem_size) {
            Ok(bounds) => bounds,
            Err(_) => return,
        };

    let (msg_start, msg_end) = match check_bounds(msg_ptr as usize, msg_len as usize, mem_size) {
        Ok(bounds) => bounds,
        Err(_) => return,
    };

    let level = std::str::from_utf8(&mem_data[level_start..level_end]).unwrap_or("info");
    let msg = std::str::from_utf8(&mem_data[msg_start..msg_end]).unwrap_or("");

    let plugin_id = caller.data().plugin_id.clone();
    let db = caller.data().db.clone();
    let db_backend = caller.data().db_backend;
    let log_level: super::LogLevel = level.parse().unwrap_or_default();
    super::log(&plugin_id, log_level, msg, db, db_backend);
}

/// The longest panic report the host reads. The message reaches an API
/// caller's error body and the logs, so its size is not the guest's to choose.
pub(crate) const MAX_PANIC_REPORT_BYTES: usize = 64 * 1024;

/// Host function: record the panic a guest is about to trap on. A report
/// that cannot be read, or is longer than [`MAX_PANIC_REPORT_BYTES`], is
/// dropped unparsed, and the run then ends as the plain trap it would have
/// been anyway.
fn host_script_panic(mut caller: wasmtime::Caller<'_, PluginState>, ptr: i32, len: i32) {
    if len as u32 as usize > MAX_PANIC_REPORT_BYTES {
        return;
    }
    let Some(memory) = caller.data().memory else {
        return;
    };
    let data = memory.data(&caller);
    let Ok((start, end)) = check_bounds(ptr as usize, len as usize, data.len()) else {
        return;
    };
    if let Ok(report) = serde_json::from_slice(&data[start..end]) {
        caller.data_mut().panic = Some(report);
    }
}

/// Host function: get a secret value by name
/// Returns a packed i64: (ptr << 32) | len, or 0 on error
///
/// The response is JSON-encoded as `{"ok": "<value>"}` so plugins can
/// deserialize it with the same `Response<String>` envelope they use for
/// other host calls.
async fn host_get_secret_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    name_ptr: i32,
    name_len: i32,
) -> i64 {
    if let Err(envelope) = require_capability(
        caller.data(),
        requirement_for_import("host_get_secret").unwrap(),
    ) {
        return write_guest_response(caller, &envelope).await;
    }

    let name = match read_guest_string(caller, name_ptr, name_len) {
        Some(n) => n,
        None => return 0,
    };

    let value = match caller.data().secrets.get(&name) {
        Some(v) => v.clone(),
        None => return 0,
    };

    let response_bytes = serde_json::to_vec(&serde_json::json!({"ok": value})).unwrap_or_default();
    write_guest_response(caller, &response_bytes).await
}

// ============================================================================
// Async host function implementations
// ============================================================================

/// Build a HostContext from PluginState, requires db to be present
fn build_host_context(state: &PluginState) -> Option<super::HostContext> {
    let db = state.db.clone()?;
    Some(super::HostContext {
        plugin_id: state.plugin_id.clone(),
        scope: state.scope.clone(),
        secrets: state.secrets.clone(),
        config: state.config.clone(),
        records: records_or_local(state, &db),
        db,
        db_backend: state.db_backend,
        http_client: state.http_client.clone(),
        lexicons: state.lexicons.clone(),
    })
}

/// Host function: make an HTTP request
async fn host_http_request_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    req_ptr: i32,
    req_len: i32,
) -> i64 {
    if let Err(envelope) = require_capability(
        caller.data(),
        requirement_for_import("host_http_request").unwrap(),
    ) {
        return write_guest_response(caller, &envelope).await;
    }

    let req_bytes = match read_guest_bytes(caller, req_ptr, req_len) {
        Some(b) => b,
        None => {
            tracing::error!(
                "host_http_request: failed to read guest memory (ptr={req_ptr}, len={req_len})"
            );
            return 0;
        }
    };

    let request: super::HttpRequest = match serde_json::from_slice(&req_bytes) {
        Ok(r) => r,
        Err(e) => {
            tracing::error!("host_http_request: failed to parse request JSON: {e}");
            return 0;
        }
    };

    let url = request.url.clone();
    let method = request.method.clone();

    let unrestricted = {
        let state = caller.data();
        let unrestricted = state
            .capabilities
            .contains(&PluginCapability::NetworkRequestUnrestricted);
        if !unrestricted {
            let host = reqwest::Url::parse(&request.url)
                .ok()
                .and_then(|u| u.host_str().map(String::from));
            let ok = host
                .as_deref()
                .map(|h| super::host_allowed(&state.allowed_hosts, h))
                .unwrap_or(false);
            if !ok {
                let defined = state
                    .capabilities
                    .contains(&PluginCapability::NetworkRequestDefined);
                let message =
                    super::allowed_hosts_denial(&state.allowed_hosts, defined, host.as_deref());
                return write_guest_response(caller, &error_envelope("FORBIDDEN", message)).await;
            }
        }
        unrestricted
    };

    let ctx = match build_host_context(caller.data()) {
        Some(c) => c,
        None => {
            tracing::error!("host_http_request: failed to build host context (db missing?)");
            return 0;
        }
    };

    let result = {
        let usage = &mut caller.data_mut().usage;
        super::http_request(&ctx, usage, request, unrestricted).await
    };

    let response_bytes = match result {
        Ok(resp) => serde_json::to_vec(&serde_json::json!({"ok": resp})).unwrap_or_default(),
        Err(e) => {
            tracing::warn!("host_http_request: HTTP {method} {url} failed: {e}");
            serde_json::to_vec(&serde_json::json!({
                "error": {"code": "HTTP_ERROR", "message": e.to_string(), "retryable": false}
            }))
            .unwrap_or_default()
        }
    };

    let packed = write_guest_response(caller, &response_bytes).await;
    if packed == 0 {
        tracing::error!(
            "host_http_request: write_guest_response returned 0 for {} {} (response_len={})",
            method,
            url,
            response_bytes.len()
        );
    }
    packed
}

/// Host function: get a value from KV store
async fn host_kv_get_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    key_ptr: i32,
    key_len: i32,
) -> i64 {
    if let Err(envelope) = require_capability(
        caller.data(),
        requirement_for_import("host_kv_get").unwrap(),
    ) {
        return write_guest_response(caller, &envelope).await;
    }

    let key = match read_guest_string(caller, key_ptr, key_len) {
        Some(k) => k,
        None => return 0,
    };

    let ctx = match build_host_context(caller.data()) {
        Some(c) => c,
        None => return 0,
    };

    let result = super::kv_get(&ctx, &key).await;

    let response_bytes = match result {
        Ok(Some(value)) => {
            serde_json::to_vec(&serde_json::json!({"ok": value})).unwrap_or_default()
        }
        Ok(None) => return 0,
        Err(e) => serde_json::to_vec(&serde_json::json!({
            "error": {"code": "KV_ERROR", "message": e.to_string(), "retryable": false}
        }))
        .unwrap_or_default(),
    };

    write_guest_response(caller, &response_bytes).await
}

/// Host function: set a value in KV store. Returns `0` on success and `-1`
/// on any failure, including a missing `kv:write` capability — the `env`
/// import signature is a bare `i32`, so there's no envelope to carry a
/// distinct "forbidden" code back to the guest.
async fn host_kv_set_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    key_ptr: i32,
    key_len: i32,
    val_ptr: i32,
    val_len: i32,
    ttl: i32,
) -> i32 {
    if require_capability(
        caller.data(),
        requirement_for_import("host_kv_set").unwrap(),
    )
    .is_err()
    {
        tracing::warn!(
            plugin_id = %caller.data().plugin_id,
            "host_kv_set: missing kv:write capability"
        );
        return -1;
    }

    let key = match read_guest_string(caller, key_ptr, key_len) {
        Some(k) => k,
        None => return -1,
    };
    let value = match read_guest_bytes(caller, val_ptr, val_len) {
        Some(v) => v,
        None => return -1,
    };

    let ttl_secs = if ttl > 0 { Some(ttl as u32) } else { None };

    let ctx = match build_host_context(caller.data()) {
        Some(c) => c,
        None => return -1,
    };

    let usage = &mut caller.data_mut().usage;
    match super::kv_set(&ctx, usage, &key, value, ttl_secs).await {
        Ok(()) => 0,
        // The import's return is a bare `i32`, so the reason cannot reach the
        // guest; without this it reaches nobody.
        Err(e) => {
            tracing::warn!(plugin_id = %caller.data().plugin_id, error = %e, "host_kv_set failed");
            -1
        }
    }
}

/// Host function: delete a value from KV store. Returns `0` on success and
/// `-1` on any failure, including a missing `kv:write` capability — the
/// `env` import signature is a bare `i32`, so there's no envelope to carry a
/// distinct "forbidden" code back to the guest.
async fn host_kv_delete_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    key_ptr: i32,
    key_len: i32,
) -> i32 {
    if require_capability(
        caller.data(),
        requirement_for_import("host_kv_delete").unwrap(),
    )
    .is_err()
    {
        tracing::warn!(
            plugin_id = %caller.data().plugin_id,
            "host_kv_delete: missing kv:write capability"
        );
        return -1;
    }

    let key = match read_guest_string(caller, key_ptr, key_len) {
        Some(k) => k,
        None => return -1,
    };

    let ctx = match build_host_context(caller.data()) {
        Some(c) => c,
        None => return -1,
    };

    match super::kv_delete(&ctx, &key).await {
        Ok(()) => 0,
        Err(_) => -1,
    }
}

/// Host function: look up an AT Protocol record
async fn host_lookup_record_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    req_ptr: i32,
    req_len: i32,
) -> i64 {
    if let Err(envelope) = require_capability(
        caller.data(),
        requirement_for_import("host_lookup_record").unwrap(),
    ) {
        return write_guest_response(caller, &envelope).await;
    }

    let req_bytes = match read_guest_bytes(caller, req_ptr, req_len) {
        Some(b) => b,
        None => return 0,
    };

    let request: super::LookupRequest = match serde_json::from_slice(&req_bytes) {
        Ok(r) => r,
        Err(_) => return 0,
    };

    let ctx = match build_host_context(caller.data()) {
        Some(c) => c,
        None => return 0,
    };

    let result = super::lookup_record_by_request(&ctx, request).await;

    let response_bytes = match result {
        Ok(record) => serde_json::to_vec(&serde_json::json!({"ok": record})).unwrap_or_default(),
        Err(e) => serde_json::to_vec(&serde_json::json!({
            "error": {"code": "LOOKUP_ERROR", "message": e.to_string(), "retryable": false}
        }))
        .unwrap_or_default(),
    };

    write_guest_response(caller, &response_bytes).await
}

/// `host_lexicon_get` reads the in-memory lexicon registry, not the database,
/// so it can't go through `host_spec_impl` — that refuses instances with no
/// pool, which a lexicon lookup never needed in the first place.
async fn host_lexicon_get_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    req_ptr: i32,
    req_len: i32,
) -> i64 {
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let spec: happyview_plugin_sdk::wire::LexiconGet = match serde_json::from_slice(&bytes) {
        Ok(s) => s,
        Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
    };
    let lexicons = caller.data().lexicons.clone();
    let value = super::lexicon_get(&lexicons, &spec.nsid).await;
    let response = serde_json::to_vec(&serde_json::json!({"ok": value})).unwrap_or_default();
    write_guest_response(caller, &response).await
}

/// The plugin's effective allowed-hosts list, resolved once at instantiation
/// and carried on `PluginState`: the operator's list under
/// `network:request:defined`, the manifest's under `network:request`, or
/// empty otherwise. A plugin reads this to behave sensibly no matter which
/// of the three network capabilities it holds.
fn resolved_allowed_hosts(state: &PluginState) -> Vec<String> {
    state.allowed_hosts.clone()
}

/// `host_allowed_hosts` is a free import: the list itself carries nothing
/// secret, and a plugin needs to be able to read it regardless of which
/// capability (if any) it holds. The request is always `{}`; nothing to
/// decode.
async fn host_allowed_hosts_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    _req_ptr: i32,
    _req_len: i32,
) -> i64 {
    let hosts = resolved_allowed_hosts(caller.data());
    let response = serde_json::to_vec(&serde_json::json!({"ok": hosts})).unwrap_or_default();
    write_guest_response(caller, &response).await
}

/// The instance's shared record store, which carries the interned-id cache,
/// the compression dictionaries and the external storage's routing; `None`
/// for an instance built without the app state (plugin unit tests), which
/// then reads and writes through a store of its own.
fn shared_records(state: &PluginState) -> Option<crate::record_store::RecordStore> {
    state.app_state.as_ref().map(|app| app.records.clone())
}

/// [`shared_records`], or, for an instance built without the app state
/// (plugin unit tests), a local store over its own pool.
fn records_or_local(state: &PluginState, db: &sqlx::AnyPool) -> crate::record_store::RecordStore {
    shared_records(state)
        .unwrap_or_else(|| crate::record_store::RecordStore::new(db.clone(), state.db_backend))
}

/// The record collection a spec reads, which picks the database it runs on
/// (an external storage's route when that claims the collection).
trait SpecCollection {
    fn collection(&self) -> Option<&str>;

    /// True when the spec reads the record views, which run where raw SQL
    /// reads of them do: on the external storage's raw read route, if any.
    fn reads_record_views(&self) -> bool {
        false
    }
}

impl SpecCollection for happyview_plugin_sdk::wire::RecordsQuery {
    fn collection(&self) -> Option<&str> {
        Some(&self.collection)
    }
}

impl SpecCollection for happyview_plugin_sdk::wire::RecordsCount {
    fn collection(&self) -> Option<&str> {
        Some(&self.collection)
    }
}

impl SpecCollection for happyview_plugin_sdk::wire::RecordsSearch {
    fn collection(&self) -> Option<&str> {
        Some(&self.collection)
    }
}

impl SpecCollection for happyview_plugin_sdk::wire::BacklinksQuery {
    fn collection(&self) -> Option<&str> {
        Some(&self.collection)
    }
}

impl SpecCollection for GetSpec {
    fn collection(&self) -> Option<&str> {
        crate::record_store::codec::parse_record_uri(&self.uri).map(|(_, collection, _)| collection)
    }
}

/// A table query reads the instance's own tables, or the record views.
impl SpecCollection for happyview_plugin_sdk::wire::TableQuery {
    fn collection(&self) -> Option<&str> {
        None
    }

    fn reads_record_views(&self) -> bool {
        crate::raw_sql_guard::is_record_view(&self.table)
    }
}

/// Index writes route inside the shared store, not by pool; their collection
/// says whose storage a failure belongs to.
impl SpecCollection for happyview_plugin_sdk::wire::IndexPut {
    fn collection(&self) -> Option<&str> {
        Some(&self.collection)
    }
}

impl SpecCollection for happyview_plugin_sdk::wire::IndexDelete {
    fn collection(&self) -> Option<&str> {
        crate::record_store::codec::parse_record_uri(&self.uri).map(|(_, collection, _)| collection)
    }
}

/// One entry point for every spec-taking record/table query import: gate on
/// the import's capability, decode the spec, run it against the database, and
/// wrap the result in the standard envelope. `RecordsError::InvalidSpec`
/// becomes `INVALID_SPEC` rather than `DB_ERROR` so a plugin can tell a bad
/// query from a database failure.
async fn host_spec_impl<S, R, F, Fut>(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    import: &'static str,
    req_ptr: i32,
    req_len: i32,
    run: F,
) -> i64
where
    S: serde::de::DeserializeOwned + SpecCollection,
    R: serde::Serialize,
    F: FnOnce(sqlx::AnyPool, crate::db::DatabaseBackend, S) -> Fut,
    Fut: std::future::Future<Output = Result<R, super::RecordsError>>,
{
    // A free import has no row in the requirements table and nothing to gate
    // on; anything else missing one is a programming error, not a grant.
    if !is_free_import(import)
        && let Err(envelope) =
            require_capability(caller.data(), requirement_for_import(import).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let spec: S = match serde_json::from_slice(&bytes) {
        Ok(s) => s,
        Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
    };
    let Some(db) = caller.data().db.clone() else {
        return write_guest_response(caller, &error_envelope("HOST_ERROR", "no database")).await;
    };
    let backend = caller.data().db_backend;
    // A claimed collection is read on the external storage's route, and the
    // record views on its raw read route; every other spec runs on the
    // instance's own database, as it always has.
    let external = caller.data().app_state.as_ref().and_then(|app| {
        if spec.reads_record_views() {
            let ext = app.records.external()?;
            return ext.raw_route().map(|route| (ext.clone(), route));
        }
        let collection = spec.collection()?;
        app.records
            .external_route(collection)
            .map(|ext| (ext.clone(), ext.read_route(collection)))
    });
    let (db, backend) = match &external {
        Some((_, route)) => (route.pool.clone(), route.backend),
        None => (db, backend),
    };
    let response = match run(db, backend, spec).await {
        Ok(value) => serde_json::to_vec(&serde_json::json!({"ok": value})).unwrap_or_default(),
        Err(super::RecordsError::InvalidSpec(msg)) => error_envelope("INVALID_SPEC", msg),
        // The claiming storage is down: a code of its own, so a script that
        // passes it on answers 503 rather than failing.
        Err(super::RecordsError::Database(e))
            if external
                .as_ref()
                .is_some_and(|(ext, _)| ext.unavailable(&e)) =>
        {
            error_envelope(crate::error::STORAGE_UNAVAILABLE, e)
        }
        Err(e) => error_envelope("DB_ERROR", e),
    };
    write_guest_response(caller, &response).await
}

/// One entry point for every import that acts as the calling user: gate on
/// the import's capability, then on there being a user at all, then decode the
/// spec and run it. The session comes off the instance rather than the spec,
/// so a library cannot name a user it was not lent.
async fn host_caller_impl<S, R, F, Fut>(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    import: &'static str,
    req_ptr: i32,
    req_len: i32,
    run: F,
) -> i64
where
    S: serde::de::DeserializeOwned,
    R: serde::Serialize,
    F: FnOnce(Arc<CallerSession>, S) -> Fut,
    Fut: std::future::Future<Output = Result<R, super::CallerError>>,
{
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(import).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let Some(session) = caller.data().caller.clone() else {
        return write_guest_response(
            caller,
            &error_envelope(
                "NO_SESSION",
                "this script context has no caller session, so nothing can be done as a user here",
            ),
        )
        .await;
    };
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let spec: S = match serde_json::from_slice(&bytes) {
        Ok(s) => s,
        Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
    };
    let response = match run(session, spec).await {
        Ok(value) => serde_json::to_vec(&serde_json::json!({"ok": value})).unwrap_or_default(),
        Err(e) => error_envelope(e.code(), caller_error_message(&e)),
    };
    write_guest_response(caller, &response).await
}

/// The gate/decode steps [`host_app_impl`] and [`host_app_caller_impl`]
/// share: capability, then an app state to act against at all — only the
/// direct-construction test and external-auth call sites leave it unset —
/// then the decoded spec. `Err` carries the return value a failure has
/// already written, so both callers can return it directly.
async fn host_app_gate<S>(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    import: &'static str,
    req_ptr: i32,
    req_len: i32,
) -> Result<(crate::AppState, S), i64>
where
    S: serde::de::DeserializeOwned,
{
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(import).unwrap())
    {
        return Err(write_guest_response(caller, &envelope).await);
    }
    let Some(app_state) = caller.data().app_state.clone() else {
        return Err(write_guest_response(
            caller,
            &error_envelope("HOST_ERROR", "this instance has no app state"),
        )
        .await);
    };
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return Err(0);
    };
    let spec: S = match serde_json::from_slice(&bytes) {
        Ok(s) => s,
        Err(e) => return Err(write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await),
    };
    Ok((app_state, spec))
}

/// One entry point for the six linked-repo imports: each spec already names
/// the DID whose linked repo it acts on, so none of them needs the runner's
/// own identity — only the whole instance to act against.
async fn host_app_impl<S, R, E, F, Fut>(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    import: &'static str,
    req_ptr: i32,
    req_len: i32,
    code: fn(&E) -> &'static str,
    run: F,
) -> i64
where
    S: serde::de::DeserializeOwned,
    R: serde::Serialize,
    E: std::fmt::Display,
    F: FnOnce(crate::AppState, S) -> Fut,
    Fut: std::future::Future<Output = Result<R, E>>,
{
    let (app_state, spec) = match host_app_gate(caller, import, req_ptr, req_len).await {
        Ok(pair) => pair,
        Err(early_return) => return early_return,
    };
    let response = match run(app_state, spec).await {
        Ok(value) => serde_json::to_vec(&serde_json::json!({"ok": value})).unwrap_or_default(),
        Err(e) => error_envelope(code(&e), e),
    };
    write_guest_response(caller, &response).await
}

/// As [`host_app_impl`], but also threads the runner's own caller DID and
/// (when it has one) its caller session through to `run` — what
/// `host_jobs_create` needs to enqueue as the script's runner, rather than as
/// whoever a spec field might name.
async fn host_app_caller_impl<S, R, E, F, Fut>(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    import: &'static str,
    req_ptr: i32,
    req_len: i32,
    code: fn(&E) -> &'static str,
    run: F,
) -> i64
where
    S: serde::de::DeserializeOwned,
    R: serde::Serialize,
    E: std::fmt::Display,
    F: FnOnce(crate::AppState, Option<String>, Option<Arc<CallerSession>>, S) -> Fut,
    Fut: std::future::Future<Output = Result<R, E>>,
{
    let (app_state, spec) = match host_app_gate(caller, import, req_ptr, req_len).await {
        Ok(pair) => pair,
        Err(early_return) => return early_return,
    };
    let caller_did = caller.data().call_ctx.caller_did.clone();
    let session = caller.data().caller.clone();
    let response = match run(app_state, caller_did, session, spec).await {
        Ok(value) => serde_json::to_vec(&serde_json::json!({"ok": value})).unwrap_or_default(),
        Err(e) => error_envelope(code(&e), e),
    };
    write_guest_response(caller, &response).await
}

/// Every space write acts as `ctx.caller_did` and has no other identity to
/// fall back to. Refusing here, before the module runs, keeps that rule in
/// one place rather than in each of the eleven write functions.
fn require_spaces_caller(caller_did: Option<String>) -> Result<String, super::spaces::SpacesError> {
    caller_did.ok_or_else(|| {
        super::spaces::SpacesError::BadInput(
            "a space write needs a caller; this script context has none".into(),
        )
    })
}

/// `host_caller_xrpc_query` on its own: gate on the capability, decode the
/// spec, and run it, but — unlike [`host_caller_impl`] — never refuse for
/// lacking a session. A query needs no PDS auth to run, and a query,
/// record-event or label script has no session to lend. The session, when the
/// runner has one, still supplies its claims; otherwise the call context's
/// `caller_did` stands in.
async fn host_caller_query_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    req_ptr: i32,
    req_len: i32,
) -> i64 {
    const IMPORT: &str = "host_caller_xrpc_query";
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(IMPORT).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let spec: happyview_plugin_sdk::wire::CallerXrpcQuery = match serde_json::from_slice(&bytes) {
        Ok(s) => s,
        Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
    };
    let Some(app_state) = caller.data().app_state.clone() else {
        return write_guest_response(
            caller,
            &error_envelope(
                "HOST_ERROR",
                "this instance has no app state to query against",
            ),
        )
        .await;
    };
    let session = caller.data().caller.clone();
    let caller_did = caller.data().call_ctx.caller_did.clone();
    let response = match super::xrpc_query(
        &app_state,
        session.as_deref(),
        caller_did.as_deref(),
        spec,
    )
    .await
    {
        Ok(value) => serde_json::to_vec(&serde_json::json!({"ok": value})).unwrap_or_default(),
        Err(e) => error_envelope(e.code(), caller_error_message(&e)),
    };
    write_guest_response(caller, &response).await
}

/// A dead session has to survive the trip out through the guest and the Lua
/// bridge, which flatten everything into one error string. The prefix is what
/// the script executor recognises there to answer 401 rather than 500.
fn caller_error_message(e: &super::CallerError) -> String {
    match e {
        super::CallerError::Auth(_) => {
            format!("{}{e}", crate::error::LUA_AUTH_ERROR_PREFIX)
        }
        other => other.to_string(),
    }
}

/// The host-side input to `host_records_get`: the SDK wrapper sends
/// `{"uri": ...}` rather than a bare string so it shares `host_spec_impl`'s
/// JSON-spec envelope with every other record/table import.
#[derive(serde::Deserialize)]
struct GetSpec {
    uri: String,
}

/// Gate on `library:call`, then hand back the executor.
fn library_access(
    state: &PluginState,
    import: &str,
) -> Result<crate::plugin::PluginExecutor, Vec<u8>> {
    require_capability(
        state,
        crate::plugin::capabilities::requirement_for_import(import).unwrap(),
    )?;
    state
        .executor
        .clone()
        .ok_or_else(|| error_envelope("HOST_ERROR", "no executor attached to this instance"))
}

async fn host_call_library_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    lib_ptr: i32,
    lib_len: i32,
    fn_ptr: i32,
    fn_len: i32,
    args_ptr: i32,
    args_len: i32,
) -> i64 {
    let (Some(lib), Some(function), Some(args_bytes)) = (
        read_guest_string(caller, lib_ptr, lib_len),
        read_guest_string(caller, fn_ptr, fn_len),
        read_guest_bytes(caller, args_ptr, args_len),
    ) else {
        return 0;
    };
    let outcome = match library_call(
        caller.data(),
        "host_call_library",
        lib,
        function,
        &args_bytes,
    ) {
        Ok(call) => call.await,
        Err(refused) => refused,
    };
    write_guest_response(caller, &serde_json::to_vec(&outcome).unwrap_or_default()).await
}

/// One library call, checked and ready to run: everything it needs is owned,
/// so `host_call_library` can await it in place and
/// `host_call_library_start` can hand it to a task of its own. It runs one
/// hop deeper than `state`, as `state`'s caller, under `state`'s context.
/// Anything wrong with the request is the `Err`, already an envelope.
fn library_call(
    state: &PluginState,
    import: &str,
    lib: String,
    function: String,
    args_bytes: &[u8],
) -> Result<
    impl Future<Output = super::pending_calls::CallOutcome> + Send + 'static,
    super::pending_calls::CallOutcome,
> {
    let refused = |envelope: Vec<u8>| {
        serde_json::from_slice(&envelope)
            .unwrap_or_else(|_| call_error("HOST_ERROR", "unreadable refusal"))
    };
    let args: Vec<serde_json::Value> = match serde_json::from_slice(args_bytes) {
        Ok(serde_json::Value::Array(a)) => a,
        Ok(_) => return Err(call_error("BAD_INPUT", "args must be a JSON array")),
        Err(e) => return Err(call_error("BAD_INPUT", e)),
    };
    let executor = library_access(state, import).map_err(refused)?;
    let ctx = state.call_ctx.clone();
    let depth = state.depth + 1;
    // A library reached through another library acts as the same user, so the
    // session travels the whole chain rather than stopping at the first hop.
    let session = state.caller.clone();
    Ok(async move {
        match executor
            .call_library_as(&lib, &function, &args, &ctx, session, depth)
            .await
        {
            Ok(ok) => super::pending_calls::CallOutcome::Ok { ok },
            Err(crate::plugin::ExecutionError::PluginError {
                code,
                message,
                retryable,
            }) => super::pending_calls::CallOutcome::Err {
                error: crate::plugin::PluginEnvelopeError {
                    code,
                    message,
                    retryable,
                },
            },
            Err(e) => call_error("LIBRARY_ERROR", e),
        }
    })
}

/// [`error_envelope`], as the typed outcome a started call settles with.
fn call_error(code: &str, message: impl std::fmt::Display) -> super::pending_calls::CallOutcome {
    super::pending_calls::CallOutcome::Err {
        error: crate::plugin::PluginEnvelopeError::new(code, message.to_string()),
    }
}

/// Start a library call and hand back its handle without waiting for it. The
/// guest is never suspended on the call's work, and the only failure answered
/// here is guest memory that cannot be read (`0`, as `host_call_library`
/// answers it); every other one settles through the handle, so an interpreter
/// reads every failure of a call from one place.
fn host_call_library_start_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    lib_ptr: i32,
    lib_len: i32,
    fn_ptr: i32,
    fn_len: i32,
    args_ptr: i32,
    args_len: i32,
) -> i32 {
    let (Some(lib), Some(function), Some(args_bytes)) = (
        read_guest_string(caller, lib_ptr, lib_len),
        read_guest_string(caller, fn_ptr, fn_len),
        read_guest_bytes(caller, args_ptr, args_len),
    ) else {
        return 0;
    };
    let call = library_call(
        caller.data(),
        "host_call_library_start",
        lib,
        function,
        &args_bytes,
    );
    let pending = &mut caller.data_mut().pending_calls;
    let handle = match call {
        Ok(call) => pending.start(call),
        Err(refused) => pending.settled(refused),
    };
    handle.map_or(0, |h| h as i32)
}

/// Suspend the guest until one of the listed calls settles and answer its
/// envelope with its handle beside it. A request that could never be answered
/// — malformed, empty, or naming a handle that is not outstanding — is refused
/// at once with no `handle`, rather than left to hang.
async fn host_call_library_wait_any_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    req_ptr: i32,
    req_len: i32,
) -> i64 {
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    if let Err(envelope) = require_capability(
        caller.data(),
        requirement_for_import("host_call_library_wait_any").unwrap(),
    ) {
        return write_guest_response(caller, &envelope).await;
    }
    let handles: Vec<u32> = match serde_json::from_slice(&bytes) {
        Ok(handles) => handles,
        Err(e) => {
            return write_guest_response(
                caller,
                &error_envelope("BAD_INPUT", format!("expected an array of handles: {e}")),
            )
            .await;
        }
    };
    let mut waiting = match caller.data_mut().pending_calls.take(&handles) {
        Ok(waiting) => waiting,
        Err(message) => {
            return write_guest_response(caller, &error_envelope("BAD_INPUT", message)).await;
        }
    };
    let (handle, outcome) = waiting.first().await;
    caller.data_mut().pending_calls.restore(waiting);
    let settled = happyview_plugin_sdk::wire::LibraryCallSettled { handle, outcome };
    write_guest_response(caller, &serde_json::to_vec(&settled).unwrap_or_default()).await
}

async fn host_get_api_surface_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    lib_ptr: i32,
    lib_len: i32,
) -> i64 {
    let Some(lib) = read_guest_string(caller, lib_ptr, lib_len) else {
        return 0;
    };
    let executor = match library_access(caller.data(), "host_get_api_surface") {
        Ok(e) => e,
        Err(envelope) => return write_guest_response(caller, &envelope).await,
    };
    let response = match executor.api_surface(&lib).await {
        Ok(surface) => {
            serde_json::to_vec(&serde_json::json!({"ok": &*surface})).unwrap_or_default()
        }
        Err(e) => error_envelope("LIBRARY_ERROR", e),
    };
    write_guest_response(caller, &response).await
}

/// Host function: run raw SQL. `execute` selects `host_db_execute` (needs
/// `database:write`) vs `host_db_query` (needs either `database:read` or
/// `database:write`, but read-only capability restricts it to read-only SQL).
/// SQL is passed through untranslated with backend-native placeholders
/// (`?` on SQLite, `$1`… on Postgres).
async fn host_db_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    execute: bool,
    sql_ptr: i32,
    sql_len: i32,
    params_ptr: i32,
    params_len: i32,
) -> i64 {
    let import = if execute {
        "host_db_execute"
    } else {
        "host_db_query"
    };
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(import).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let (Some(sql), Some(params_bytes)) = (
        read_guest_string(caller, sql_ptr, sql_len),
        read_guest_bytes(caller, params_ptr, params_len),
    ) else {
        return 0;
    };
    let params: Vec<serde_json::Value> = if params_bytes.is_empty() {
        Vec::new()
    } else {
        match serde_json::from_slice(&params_bytes) {
            Ok(serde_json::Value::Array(a)) => a,
            _ => {
                return write_guest_response(
                    caller,
                    &error_envelope("BAD_INPUT", "params must be a JSON array"),
                )
                .await;
            }
        }
    };
    // database:read alone may only run queries; database:write may run anything through either import.
    if !caller
        .data()
        .capabilities
        .contains(&PluginCapability::DatabaseWrite)
    {
        match crate::raw_sql_guard::is_read_only(&sql) {
            Ok(true) => {}
            Ok(false) => {
                return write_guest_response(
                    caller,
                    &error_envelope(
                        "FORBIDDEN",
                        "database:read permits only read-only statements; declare database:write to modify data",
                    ),
                )
                .await;
            }
            Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
        }
    }
    let external = caller
        .data()
        .app_state
        .as_ref()
        .and_then(|app| app.records.external().cloned());
    if let Some(external) = &external
        && let Err(e) = crate::raw_sql_guard::check_claimed_record_writes(&sql, &params, &|c| {
            external.claims(c)
        })
    {
        return write_guest_response(caller, &error_envelope("FORBIDDEN", e)).await;
    }
    // SQL that reads the record views runs on the external storage's raw read
    // route, where those views show its records. It is classified before the
    // read-only gate: a read the gate cannot vouch for (a locking clause, SQL
    // the parser rejects) must not fall through to the local views, which hold
    // none of the claimed records. A write runs locally as it always has, the
    // claimed-write guard above covering claimed collections; one that also
    // reads the views sees only the local rows.
    if !execute
        && let Some(ext) = &external
        && let Some(route) = ext.raw_route()
        && !crate::raw_sql_guard::is_write(&sql)
    {
        match crate::raw_sql_guard::record_view_use(&sql) {
            Ok(crate::raw_sql_guard::RecordViews::None) => {}
            Err(e) => return write_guest_response(caller, &error_envelope("FORBIDDEN", e)).await,
            Ok(_) if crate::raw_sql_guard::is_read_only(&sql) != Ok(true) => {
                return write_guest_response(
                    caller,
                    &error_envelope(
                        "FORBIDDEN",
                        "raw SQL that reads claimed records must be a plain read-only query",
                    ),
                )
                .await;
            }
            Ok(crate::raw_sql_guard::RecordViews::Mixed(_)) => {
                return write_guest_response(
                    caller,
                    &error_envelope("FORBIDDEN", crate::raw_sql_guard::RAW_MIXED_REFUSED),
                )
                .await;
            }
            Ok(crate::raw_sql_guard::RecordViews::Only) => {
                let response = match super::run_read_only_query_on(
                    &route.pool,
                    route.backend,
                    &sql,
                    &params,
                )
                .await
                {
                    Ok(rows) => {
                        serde_json::to_vec(&serde_json::json!({"ok": rows})).unwrap_or_default()
                    }
                    Err(super::QueryError::Database(e)) if ext.unavailable(&e) => {
                        error_envelope(crate::error::STORAGE_UNAVAILABLE, e)
                    }
                    Err(e) => {
                        let backend = match route.backend {
                            crate::db::DatabaseBackend::Postgres => "postgres",
                            crate::db::DatabaseBackend::Sqlite => "sqlite",
                        };
                        error_envelope(
                            "DB_ERROR",
                            format!(
                                "{e} (record views on this instance are served by a {backend} database)"
                            ),
                        )
                    }
                };
                return write_guest_response(caller, &response).await;
            }
        }
    }
    let Some(db) = caller.data().db.clone() else {
        return write_guest_response(caller, &error_envelope("HOST_ERROR", "no database")).await;
    };
    let response = if execute {
        match super::run_execute(&db, &sql, &params).await {
            Ok(n) => serde_json::to_vec(&serde_json::json!({"ok": {"rows_affected": n}}))
                .unwrap_or_default(),
            Err(e) => error_envelope("DB_ERROR", e),
        }
    } else {
        match super::run_query(&db, &sql, &params).await {
            Ok(rows) => serde_json::to_vec(&serde_json::json!({"ok": rows})).unwrap_or_default(),
            Err(e) => error_envelope("DB_ERROR", e),
        }
    };
    write_guest_response(caller, &response).await
}

/// `AtprotoError` → envelope code. Distinct codes so a plugin can tell "no
/// signer configured" from "signature could not be checked" from an ordinary
/// database or network failure, rather than seeing one undifferentiated
/// `HOST_ERROR`.
fn atproto_error_envelope(e: super::AtprotoError) -> Vec<u8> {
    use super::AtprotoError;
    match e {
        AtprotoError::Resolve(msg) => error_envelope("RESOLVE_ERROR", msg),
        AtprotoError::Blob { status, body } => {
            error_envelope("BLOB_ERROR", format!("PDS returned {status}: {body}"))
        }
        AtprotoError::NoSigner => error_envelope(
            "NO_SIGNER",
            "no attestation signer is configured on this instance",
        ),
        AtprotoError::Unverifiable(msg) => error_envelope("UNVERIFIABLE", msg),
        AtprotoError::Database(inner) => error_envelope("HOST_ERROR", inner),
        AtprotoError::Other(msg) => error_envelope("HOST_ERROR", msg),
    }
}

/// Host function: resolve the AT Protocol service a DID's document
/// advertises. Needs no session — it reads the public network, not the
/// caller's own repo — so it runs off `app_state` alone.
async fn host_atproto_resolve_service_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    req_ptr: i32,
    req_len: i32,
) -> i64 {
    const IMPORT: &str = "host_atproto_resolve_service";
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(IMPORT).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let spec: happyview_plugin_sdk::wire::AtprotoResolveService =
        match serde_json::from_slice(&bytes) {
            Ok(s) => s,
            Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
        };
    let Some(app_state) = caller.data().app_state.clone() else {
        return write_guest_response(
            caller,
            &error_envelope(
                "HOST_ERROR",
                "this instance has no app state to resolve against",
            ),
        )
        .await;
    };
    let response =
        match super::resolve_service(&app_state.http, &app_state.config.plc_url, &spec.did).await {
            Ok(endpoint) => {
                serde_json::to_vec(&serde_json::json!({"ok": endpoint})).unwrap_or_default()
            }
            Err(e) => atproto_error_envelope(e),
        };
    write_guest_response(caller, &response).await
}

/// Host function: resolve a handle or a DID to both.
///
/// `handle` comes back only when it was confirmed in both directions, which
/// is what makes it safe to display; the DID is whatever the identifier
/// resolved to.
async fn host_atproto_resolve_identity_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    req_ptr: i32,
    req_len: i32,
) -> i64 {
    const IMPORT: &str = "host_atproto_resolve_identity";
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(IMPORT).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let spec: happyview_plugin_sdk::wire::AtprotoResolveIdentity =
        match serde_json::from_slice(&bytes) {
            Ok(s) => s,
            Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
        };
    let Some(app_state) = caller.data().app_state.clone() else {
        return write_guest_response(
            caller,
            &error_envelope(
                "HOST_ERROR",
                "this instance has no app state to resolve against",
            ),
        )
        .await;
    };
    let response = match crate::identity::resolve_verified(
        &app_state.http,
        &app_state.config.plc_url,
        &spec.identifier,
    )
    .await
    {
        Ok(identity) => serde_json::to_vec(&serde_json::json!({
            "ok": happyview_plugin_sdk::wire::ResolvedIdentity {
                did: identity.did,
                handle: identity.handle,
            }
        }))
        .unwrap_or_default(),
        Err(e) => error_envelope("HOST_ERROR", e),
    };
    write_guest_response(caller, &response).await
}

/// Host function: store bytes under the CID of their content.
///
/// The caller does not choose the key, so a plugin cannot file bytes under a
/// CID they do not hash to, and storing the same content twice is the row
/// already there.
async fn host_blob_put_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    req_ptr: i32,
    req_len: i32,
) -> i64 {
    const IMPORT: &str = "host_blob_put";
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(IMPORT).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let req: happyview_plugin_sdk::wire::BlobPut = match serde_json::from_slice(&bytes) {
        Ok(r) => r,
        Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
    };
    let Some(ctx) = build_host_context(caller.data()) else {
        return 0;
    };

    let response =
        match crate::blobs::put(&ctx.db, ctx.db_backend, &req.bytes, &req.mime_type).await {
            Ok(cid) => serde_json::to_vec(&serde_json::json!({ "ok": { "cid": cid } })),
            Err(e) => serde_json::to_vec(&serde_json::json!({
                "error": { "code": "BLOB_ERROR", "message": e.to_string(), "retryable": false }
            })),
        }
        .unwrap_or_default();
    write_guest_response(caller, &response).await
}

/// Host function: read a stored blob's bytes.
async fn host_blob_get_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    req_ptr: i32,
    req_len: i32,
) -> i64 {
    const IMPORT: &str = "host_blob_get";
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(IMPORT).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let req: happyview_plugin_sdk::wire::BlobLookup = match serde_json::from_slice(&bytes) {
        Ok(r) => r,
        Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
    };
    let Some(ctx) = build_host_context(caller.data()) else {
        return 0;
    };

    let response = match crate::blobs::get(&ctx.db, ctx.db_backend, &req.cid).await {
        // Absent is `0`, the same "nothing here" a `kv` miss answers, rather
        // than an error: asking for a blob that is not held is an ordinary
        // outcome.
        Ok(None) => return 0,
        Ok(Some(blob)) => {
            let size = blob.bytes.len() as u64;
            serde_json::to_vec(
                &serde_json::json!({ "ok": happyview_plugin_sdk::wire::BlobData {
                bytes: blob.bytes,
                mime_type: blob.mime_type,
                size,
            } }),
            )
        }
        Err(e) => serde_json::to_vec(&serde_json::json!({
            "error": { "code": "BLOB_ERROR", "message": e.to_string(), "retryable": false }
        })),
    }
    .unwrap_or_default();
    write_guest_response(caller, &response).await
}

/// Host function: a stored blob's media type and size, without its bytes.
///
/// Answering `0` for an absent blob is what lets a caller ask whether one is
/// held without transferring it, so there is no separate existence import to
/// drift from this one.
async fn host_blob_stat_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    req_ptr: i32,
    req_len: i32,
) -> i64 {
    const IMPORT: &str = "host_blob_stat";
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(IMPORT).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let req: happyview_plugin_sdk::wire::BlobLookup = match serde_json::from_slice(&bytes) {
        Ok(r) => r,
        Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
    };
    let Some(ctx) = build_host_context(caller.data()) else {
        return 0;
    };

    let response = match crate::blobs::stat(&ctx.db, ctx.db_backend, &req.cid).await {
        Ok(None) => return 0,
        Ok(Some(stat)) => serde_json::to_vec(
            &serde_json::json!({ "ok": happyview_plugin_sdk::wire::BlobInfo {
                cid: req.cid,
                mime_type: stat.mime_type,
                size: stat.size as u64,
            } }),
        ),
        Err(e) => serde_json::to_vec(&serde_json::json!({
            "error": { "code": "BLOB_ERROR", "message": e.to_string(), "retryable": false }
        })),
    }
    .unwrap_or_default();
    write_guest_response(caller, &response).await
}

/// Host function: download a blob from a repo. Same "no session needed" shape
/// as service resolution — the DID is a parameter of the request, not the
/// caller's own identity.
async fn host_atproto_blob_download_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    req_ptr: i32,
    req_len: i32,
) -> i64 {
    const IMPORT: &str = "host_atproto_blob_download";
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(IMPORT).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let spec: happyview_plugin_sdk::wire::AtprotoBlobDownload = match serde_json::from_slice(&bytes)
    {
        Ok(s) => s,
        Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
    };
    let Some(app_state) = caller.data().app_state.clone() else {
        return write_guest_response(
            caller,
            &error_envelope(
                "HOST_ERROR",
                "this instance has no app state to resolve against",
            ),
        )
        .await;
    };

    // A blob download is an HTTP fetch like any other and counts against the
    // same shared budget `host_http_request` does — it just goes through PDS
    // resolution instead of a plugin-supplied URL. There is no
    // `allowed_hosts` check to make here: the host, not the plugin, chose
    // the PDS host, via DID resolution inside `super::blob_download`.
    let requests_over_budget = {
        let usage = &mut caller.data_mut().usage;
        usage.http_requests += 1;
        (usage.http_requests > super::MAX_HTTP_REQUESTS).then_some(usage.http_requests)
    };
    if let Some(requests) = requests_over_budget {
        return write_guest_response(
            caller,
            &error_envelope(
                "BLOB_ERROR",
                format!(
                    "too many requests: {requests} > {}",
                    super::MAX_HTTP_REQUESTS
                ),
            ),
        )
        .await;
    }

    // `host_http_request` refuses once the transfer budget is already spent,
    // before sending — checked here too, not just after the fetch below, so
    // an exhausted budget can't still push through one more multi-hundred-MB
    // blob before the next call finally sees it.
    let already_over_budget =
        caller.data().usage.http_bytes_transferred > super::MAX_HTTP_TOTAL_TRANSFER;
    if already_over_budget {
        return write_guest_response(
            caller,
            &error_envelope(
                "BLOB_ERROR",
                format!(
                    "transfer limit exceeded: {} > {}",
                    caller.data().usage.http_bytes_transferred,
                    super::MAX_HTTP_TOTAL_TRANSFER
                ),
            ),
        )
        .await;
    }

    let blob = match super::blob_download(&app_state.http, &app_state.config.plc_url, spec).await {
        Ok(blob) => blob,
        Err(e) => return write_guest_response(caller, &atproto_error_envelope(e)).await,
    };

    let transfer_over_budget = {
        let usage = &mut caller.data_mut().usage;
        usage.http_bytes_transferred += blob.size;
        (usage.http_bytes_transferred > super::MAX_HTTP_TOTAL_TRANSFER)
            .then_some(usage.http_bytes_transferred)
    };
    if let Some(transferred) = transfer_over_budget {
        return write_guest_response(
            caller,
            &error_envelope(
                "BLOB_ERROR",
                format!(
                    "transfer limit exceeded: {transferred} > {}",
                    super::MAX_HTTP_TOTAL_TRANSFER
                ),
            ),
        )
        .await;
    }

    let response = serde_json::to_vec(&serde_json::json!({"ok": blob})).unwrap_or_default();
    write_guest_response(caller, &response).await
}

/// Host function: look up labels for a set of URIs, keyed by URI.
async fn host_labels_get_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    req_ptr: i32,
    req_len: i32,
) -> i64 {
    const IMPORT: &str = "host_labels_get";
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(IMPORT).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let spec: happyview_plugin_sdk::wire::LabelsGet = match serde_json::from_slice(&bytes) {
        Ok(s) => s,
        Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
    };
    let Some(db) = caller.data().db.clone() else {
        return write_guest_response(caller, &error_envelope("HOST_ERROR", "no database")).await;
    };
    let records = records_or_local(caller.data(), &db);
    let response = match super::labels_get(&records, &spec.uris).await {
        Ok(map) => serde_json::to_vec(&serde_json::json!({"ok": map})).unwrap_or_default(),
        // `labels_get`'s only source of `Other` is the URI-count cap — a
        // caller-fixable mistake, not a host failure, so it gets `BAD_INPUT`
        // here rather than the `HOST_ERROR` the shared envelope maps `Other`
        // to for the other imports.
        Err(super::AtprotoError::Other(msg)) => error_envelope("BAD_INPUT", msg),
        Err(e) => atproto_error_envelope(e),
    };
    write_guest_response(caller, &response).await
}

/// Host function: sign a record with this instance's attestation key. Reads
/// `attestation_signer` off `app_state` — this import needs no session, so it
/// cannot get the signer any other way.
async fn host_attest_sign_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    req_ptr: i32,
    req_len: i32,
) -> i64 {
    const IMPORT: &str = "host_attest_sign";
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(IMPORT).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let spec: happyview_plugin_sdk::wire::AttestSign = match serde_json::from_slice(&bytes) {
        Ok(s) => s,
        Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
    };
    let signer = caller
        .data()
        .app_state
        .as_ref()
        .and_then(|s| s.attestation_signer.clone());
    let caller_did = caller.data().call_ctx.caller_did.clone();

    // The DID is part of the signed content, so an absent caller can't fall
    // back to signing as `""` — that produces a signature that can never
    // verify against a real repo. A missing signer is checked first (and
    // reported as `NO_SIGNER` below) so a plugin with neither configured
    // still gets the more specific reason for the failure it actually hit.
    if signer.is_some() && caller_did.is_none() {
        return write_guest_response(
            caller,
            &error_envelope("BAD_INPUT", "signing needs a caller"),
        )
        .await;
    }
    let did = caller_did.unwrap_or_default();
    let response = match super::attest_sign(signer.as_deref(), &did, spec.record) {
        Ok(sig) => serde_json::to_vec(&serde_json::json!({"ok": sig})).unwrap_or_default(),
        Err(e) => atproto_error_envelope(e),
    };
    write_guest_response(caller, &response).await
}

/// Host function: verify a record's attestation signature.
async fn host_attest_verify_impl(
    caller: &mut wasmtime::Caller<'_, PluginState>,
    req_ptr: i32,
    req_len: i32,
) -> i64 {
    const IMPORT: &str = "host_attest_verify";
    if let Err(envelope) =
        require_capability(caller.data(), requirement_for_import(IMPORT).unwrap())
    {
        return write_guest_response(caller, &envelope).await;
    }
    let Some(bytes) = read_guest_bytes(caller, req_ptr, req_len) else {
        return 0;
    };
    let spec: happyview_plugin_sdk::wire::AttestVerify = match serde_json::from_slice(&bytes) {
        Ok(s) => s,
        Err(e) => return write_guest_response(caller, &error_envelope("BAD_INPUT", e)).await,
    };
    let signer = caller
        .data()
        .app_state
        .as_ref()
        .and_then(|s| s.attestation_signer.clone());
    let response = match super::attest_verify(signer.as_deref(), spec) {
        Ok(valid) => serde_json::to_vec(&serde_json::json!({"ok": valid})).unwrap_or_default(),
        Err(e) => atproto_error_envelope(e),
    };
    write_guest_response(caller, &response).await
}

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

    #[test]
    fn test_plugin_state_fields_exist() {
        fn _check_fields(state: &PluginState) {
            let _ = &state.plugin_id;
            let _ = &state.scope;
            let _ = &state.secrets;
            let _ = &state.config;
            let _ = &state.usage;
            let _ = &state.memory;
            let _ = &state.alloc;
            let _ = &state.dealloc;
        }
    }

    /// Every field filled by hand, the way `PluginExecutor::instantiate`
    /// does — no `Default` impl exists for `PluginState`, since a real
    /// instance is never built any other way. Tests that only care about one
    /// or two fields (here, `allowed_hosts`) use this rather than each
    /// re-deriving the rest.
    fn test_plugin_state(allowed_hosts: Vec<String>) -> PluginState {
        crate::db::install_drivers();
        PluginState {
            plugin_id: "test-plugin".into(),
            scope: "did:example:test".into(),
            secrets: HashMap::new(),
            config: serde_json::json!({}),
            db: None,
            db_backend: crate::db::DatabaseBackend::Sqlite,
            http_client: reqwest::Client::new(),
            lexicons: Arc::new(crate::catalog::LexiconCatalog::detached()),
            usage: Default::default(),
            memory: None,
            alloc: None,
            dealloc: None,
            capabilities: std::collections::HashSet::new(),
            allowed_hosts,
            plugin_type: crate::plugin::PluginType::Library,
            executor: None,
            call_ctx: crate::plugin::library::LibraryCallContext::default(),
            depth: 0,
            caller: None,
            script_run: None,
            app_state: None,
            wasi: super::super::build_wasi_context(
                "test-plugin",
                None,
                crate::db::DatabaseBackend::Sqlite,
            ),
            deadline: crate::plugin::runtime::GuestDeadline::unbounded(Arc::new(
                std::sync::atomic::AtomicU64::new(0),
            )),
            limiter: crate::plugin::runtime::MemoryLimiter::new(
                crate::plugin::runtime::LIBRARY_MEMORY_CEILING,
            ),
            pending_calls: Default::default(),
            panic: None,
            call_fuel: u64::MAX,
        }
    }

    /// Direct construction with no hosts set — the default a plugin sees
    /// before an operator has configured `network:request:defined`, or under
    /// any capability that grants none.
    #[test]
    fn resolved_allowed_hosts_defaults_to_empty() {
        let state = test_plugin_state(Vec::new());
        assert!(resolved_allowed_hosts(&state).is_empty());
    }

    #[test]
    fn resolved_allowed_hosts_returns_the_resolved_list() {
        let hosts = vec![
            "api.example.com".to_string(),
            "*.cdn.example.com".to_string(),
        ];
        let state = test_plugin_state(hosts.clone());
        assert_eq!(resolved_allowed_hosts(&state), hosts);
    }

    /// The executor recovers a 401 by finding the prefix anywhere in the
    /// error string, so the envelope has to carry it out of the guest.
    #[test]
    fn an_auth_failure_leaves_with_the_prefix_the_executor_looks_for() {
        let envelope = error_envelope(
            super::super::CallerError::Auth("expired".into()).code(),
            caller_error_message(&super::super::CallerError::Auth("expired".into())),
        );
        let parsed: serde_json::Value = serde_json::from_slice(&envelope).unwrap();
        assert_eq!(parsed["error"]["code"], "AUTH_REQUIRED");
        let message = parsed["error"]["message"].as_str().unwrap();
        assert!(
            message.contains(crate::error::LUA_AUTH_ERROR_PREFIX),
            "{message}"
        );
        assert!(message.contains("expired"), "{message}");
    }

    /// Only an auth failure gets the prefix; anything else wearing it would
    /// turn an unrelated error into a spurious logout.
    #[test]
    fn other_failures_are_not_dressed_as_auth_failures() {
        let message = caller_error_message(&super::super::CallerError::Pds {
            status: 400,
            body: "InvalidSwap".into(),
        });
        assert!(
            !message.contains(crate::error::LUA_AUTH_ERROR_PREFIX),
            "{message}"
        );
    }

    #[test]
    fn test_pack_ptr_len() {
        let ptr: u32 = 0x1000;
        let len: u32 = 0x0100;
        let packed: i64 = ((ptr as i64) << 32) | (len as i64);
        let unpacked_ptr = (packed >> 32) as u32;
        let unpacked_len = (packed & 0xFFFFFFFF) as u32;
        assert_eq!(unpacked_ptr, ptr);
        assert_eq!(unpacked_len, len);
    }

    #[test]
    fn test_bounds_check_helper() {
        assert!(check_bounds(0, 10, 100).is_ok());
        assert!(check_bounds(90, 10, 100).is_ok());
        assert!(check_bounds(91, 10, 100).is_err());
        assert!(check_bounds(0, 0, 100).is_ok());
    }
}