//! runModule (an entry is an event handler), line for line with
//! javascript/src/run-module/index.js: the same validation, messages, allowlist, rkey guard ledger,
//! budgets, queue overflow, cancel and drain rules.

use std::collections::{HashMap, HashSet, VecDeque};
use std::sync::Arc;
use std::time::Duration;

use serde_json::{Map, Value, json};
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinHandle;

use crate::{BoxFut, Signal, iso_now, js_num, js_string};

const DRAIN_MS: u64 = 1000;
/// The longest timer either host sets (JS setTimeout's limit, 2^31 - 1 ms): past it Node fires at once.
pub const MAX_TIMER_MS: f64 = 2_147_483_647.0;

/// A module- or host-given ms as a timer Duration, clamped to [0, MAX_TIMER_MS] (NaN is 0): a huge
/// value must not panic a spawned task (Duration::from_secs_f64 does past u64 seconds).
pub fn ms_duration(ms: f64) -> Duration {
    Duration::try_from_secs_f64(ms.clamp(0.0, MAX_TIMER_MS) / 1000.0).unwrap_or_default()
}
/// A send the stream cannot take (a JS stream with no send()): refused, not a failure.
pub const NOSEND: &str = "\u{0}nosend";

/// ctx.wasm.run bound to one module: run(entry, input) -> the export's JSON output.
pub trait Runner: Send + Sync {
    /// `io_bytes`: the input and output cap for this call (the run's budget ioBytes).
    fn run(&self, entry: String, input: Value, io_bytes: usize) -> BoxFut<'_, Result<Value, String>>;
}

/// ctx.storage.managed_records: the at:// URIs a module created, each { createdAt, module: owner }.
pub trait Ledger: Send + Sync {
    /// The entry's value; None when absent (or null).
    fn get(&self, uri: &str) -> BoxFut<'_, Result<Option<Value>, String>>;
    fn put(&self, uri: String, value: Value) -> BoxFut<'_, Result<(), String>>;
    fn delete(&self, uri: String) -> BoxFut<'_, Result<(), String>>;
}

/// The ledger as one module sees it (JS ledgerFor): an entry counts only when this module made it
/// (its `module` is the owner); puts record the owner. Entries with no module count for none.
struct Owned {
    inner: Arc<dyn Ledger>,
    owner: String,
}

impl Ledger for Owned {
    fn get(&self, uri: &str) -> BoxFut<'_, Result<Option<Value>, String>> {
        let uri = uri.to_string();
        Box::pin(async move {
            let value = self.inner.get(&uri).await?;
            Ok(value.filter(|v| v.get("module").and_then(Value::as_str) == Some(&self.owner)))
        })
    }
    fn put(&self, uri: String, mut value: Value) -> BoxFut<'_, Result<(), String>> {
        if let Value::Object(map) = &mut value {
            map.insert("module".into(), json!(self.owner));
        }
        self.inner.put(uri, value)
    }
    fn delete(&self, uri: String) -> BoxFut<'_, Result<(), String>> {
        self.inner.delete(uri)
    }
}

impl Ctx {
    /// This ctx with its ledger seen as `module` (Owned).
    fn owned(&self, module: &Module) -> Ctx {
        Ctx {
            runner: self.runner.clone(),
            ledger: self.ledger.clone().map(|inner| {
                Arc::new(Owned {
                    inner,
                    owner: module.owner(),
                }) as Arc<dyn Ledger>
            }),
            now: self.now.clone(),
        }
    }
}

/// The host `call`: one XRPC (or pseudo-) call, or a subscription.
pub trait Call: Send + Sync {
    /// Ok(None) is a call that resolved to undefined.
    fn call(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Option<Value>, String>>;
    fn subscribe(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Sub, String>>;
    /// The URL schemes (no colon: "iroh") whose HTTP_NSID requests this call serves with fetch(): a
    /// transport's (transport::fetchers). None by default.
    fn schemes(&self) -> &'static [&'static str] {
        &[]
    }
    /// One HTTP_NSID request to a URL of one of schemes() (JS: a fetchers entry), as host::Fetch::fetch.
    fn fetch(&self, _method: &str, url: &str, _headers: Vec<(String, String)>, _body: Option<Vec<u8>>) -> BoxFut<'_, Result<crate::host::Response, String>> {
        let error = format!("no fetch for {url}");
        Box::pin(async move { Err(error) })
    }
}

#[derive(Clone, Default)]
pub struct Request {
    pub params: Option<Map<String, Value>>,
    pub input: Option<Value>,
    /// The content-type of a `{ $bytes }` input (sent as those bytes); none: application/octet-stream.
    pub encoding: Option<String>,
    pub proxy: Option<String>,
    pub service: Option<String>,
    /// With a did#id service: where that DID serves it ([{ kind: iroh|webrtc|https, address }], its
    /// record's order); the host picks one it can reach (a transport's scheme, transport::TransportCall; else https,
    /// host::HostCall) and sends service auth aud did#id.
    pub endpoints: Option<Value>,
    pub subscribe: bool,
    pub framing: Option<String>,
    pub keepalive: Option<Value>,
    /// Authenticated subscriptions: {} = Authorization header, { param } = query parameter.
    pub service_auth: Option<Value>,
    /// Set by host wrappers (DidkeyCall) that resolved service_auth; never by a module.
    pub headers: Vec<(String, String)>,
    pub signal: Signal,
    /// Queue a `callback` event into the calling run while this call is pending.
    pub emit: Option<Emit>,
}

impl Request {
    /// The request as the JS host passes it (undefined fields left out).
    pub fn to_json(&self) -> Value {
        let mut out = Map::new();
        if let Some(p) = &self.params {
            out.insert("params".into(), Value::Object(p.clone()));
        }
        for (key, value) in [
            ("input", self.input.clone()),
            ("encoding", self.encoding.clone().map(Value::String)),
            ("proxy", self.proxy.clone().map(Value::String)),
            ("service", self.service.clone().map(Value::String)),
            ("endpoints", self.endpoints.clone()),
        ] {
            if let Some(v) = value {
                out.insert(key.into(), v);
            }
        }
        if self.subscribe {
            out.insert("subscribe".into(), Value::Bool(true));
            if let Some(f) = &self.framing {
                out.insert("framing".into(), Value::String(f.clone()));
            }
            if let Some(k) = &self.keepalive {
                out.insert("keepalive".into(), k.clone());
            }
            if let Some(a) = &self.service_auth {
                out.insert("serviceAuth".into(), a.clone());
            }
        }
        Value::Object(out)
    }
}

/// An open subscription: messages (Err = it failed; channel closed = it ended) and commands.
pub struct Sub {
    pub messages: mpsc::UnboundedReceiver<Result<Value, String>>,
    pub commands: mpsc::UnboundedSender<SubCmd>,
}

pub enum SubCmd {
    Send(Value, oneshot::Sender<Result<(), String>>),
    Close,
}

#[derive(Clone)]
pub struct Emit {
    tx: mpsc::UnboundedSender<Msg>,
    id: String,
    generation: u64,
}

impl Emit {
    pub fn emit(&self, event: Value) {
        let _ = self
            .tx
            .send(Msg::Callback(self.id.clone(), self.generation, event));
    }
}

/// A module as runModule takes it (loadModule's shape).
#[derive(Clone, Default)]
pub struct Module {
    pub bytes: Arc<Vec<u8>>,
    pub sha256: Option<String>,
    pub calls: Option<Vec<String>>,
    pub any_records: Option<Vec<String>>,
    /// module.json http: the origins (or "*") HTTP_NSID may reach, when the host allows them too.
    pub http: Option<Vec<String>>,
    pub name: Option<String>,
    pub didkey: Option<Value>,
    pub session: Option<Value>,
    /// Its identity (JS mod.id): the at:// module URI, or a local module's absolute path.
    pub id: Option<String>,
    /// Who owns its ledger entries when not its own (JS mod.owner): a composition's.
    pub owner: Option<String>,
    /// module.json hooks: NSIDs another party serves.
    pub hooks: Vec<Hook>,
    /// module.json implements: (hook NSID, entry) pairs this module serves.
    pub implements: Vec<(String, String)>,
    /// module.json config: the instance input a composition may give it (None: none).
    pub config: Option<Vec<Field>>,
    /// module.json anyHooks: a relay whose routed NSIDs its input names; edges from it are not checked.
    pub any_hooks: bool,
    /// What a pin covers (module_pin): its wasm and its grant fields.
    pub pin: String,
    /// Its key (rkey, lockfile key, directory name or wasm file stem): what a refused call names (oauth::ScopeCall).
    pub key: Option<String>,
    /// Its session record's (rkey, scope): all a signed-in run of it may call (oauth::ScopeCall). None: no record.
    pub scope: Option<(String, String)>,
}

/// A module.json config field (fields as in actions): a key of an instance's input, of its type.
#[derive(Clone, Debug)]
pub struct Field {
    pub key: String,
    /// string (default), integer, number, boolean, object or array
    pub kind: String,
    pub required: bool,
}

pub const FIELD_TYPES: [&str; 6] = ["string", "integer", "number", "boolean", "object", "array"];

/// A module.json hook.
#[derive(Clone, Debug)]
pub struct Hook {
    pub nsid: String,
    pub required: bool,
    pub reason: String,
    /// combine: "all-allow": routes answer kcp Verdicts, ANDed (first deny wins).
    pub all_allow: bool,
    /// A module key that serves the hook when nothing routes it.
    pub default: Option<String>,
    /// The entries that need it (None: every entry).
    pub entries: Option<Vec<String>>,
}

/// A hook default: [A-Za-z0-9_-]{1,200} (an rkey and a directory name, never a path).
pub fn module_key_ok(s: &str) -> bool {
    (1..=200).contains(&s.len()) && s.chars().all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
}

/// The module record fields that grant or route authority; a pin covers them with the wasm.
pub const GRANT_FIELDS: [&str; 7] = ["calls", "anyRecords", "hooks", "implements", "anyHooks", "didkey", "http"];

/// A module's pin (JS modulePin): hex sha256 of its wasm's hex sha256, a newline, and its GRANT_FIELDS
/// (those present) as JSON with sorted keys, so a record whose calls or hooks change no longer matches.
pub fn module_pin(bytes: &[u8], spec: &Value) -> String {
    pin_of_sha256(&crate::wasm::sha256_hex(bytes), spec)
}

/// module_pin from the wasm's hex sha256 (a published record's blob CID carries it: host::blob_sha256).
pub fn pin_of_sha256(sha256: &str, spec: &Value) -> String {
    fn sorted(v: &Value) -> Value {
        match v {
            Value::Object(m) => {
                let mut keys: Vec<&String> = m.keys().collect();
                keys.sort();
                Value::Object(keys.into_iter().map(|k| (k.clone(), sorted(&m[k]))).collect())
            }
            Value::Array(a) => Value::Array(a.iter().map(sorted).collect()),
            other => other.clone(),
        }
    }
    let grants: Map<String, Value> = GRANT_FIELDS.iter().filter_map(|k| spec.get(*k).map(|v| (k.to_string(), v.clone()))).collect();
    crate::wasm::sha256_hex(format!("{sha256}\n{}", sorted(&Value::Object(grants))).as_bytes())
}

impl Hook {
    fn parse(v: &Value) -> Option<Hook> {
        let nsid = v.get("nsid")?.as_str().filter(|n| nsid_ok(n))?.to_string();
        let reason = v.get("reason")?.as_str()?.to_string();
        let required = match v.get("required") {
            None => false,
            Some(r) => r.as_bool()?,
        };
        let default = match v.get("default") {
            None => None,
            Some(d) => Some(d.as_str().filter(|d| module_key_ok(d))?.to_string()),
        };
        let all_allow = match v.get("combine") {
            None => false,
            Some(c) => c.as_str().filter(|c| *c == "all-allow").map(|_| true)?,
        };
        let entries = match v.get("entries") {
            None => None,
            Some(e) => Some(e.as_array()?.iter().map(|x| x.as_str().map(String::from)).collect::<Option<Vec<_>>>()?),
        };
        Some(Hook { nsid, required, reason, all_allow, default, entries })
    }
}

/// "no edge routes it": a required hook that only a route (or a default) can serve.
pub const NO_ROUTE: &str = "no edge routes it";

/// A hook the host serves itself (JS hostsHook): no default, a host pseudo-call NSID (com.fedfork.exec.*) and in the
/// module's calls. Every other hook is routed, served by its default, or (optional, unrouted) null.
pub fn hosts_hook(hook: &Hook, calls: Option<&Vec<String>>) -> bool {
    hook.default.is_none() && hook.nsid.starts_with(crate::transport::PSEUDO) && calls.is_some_and(|c| c.contains(&hook.nsid))
}

/// Why a hook nothing routes cannot be served on this host (JS unservedHook), or None. A required hook the
/// host serves (hosts_hook: a transport) is served when `provides` (the NSIDs this host serves, from its
/// transports; None: none) has it. `entries`: the module's entries that run (None: not known yet); a hook
/// naming entries binds only those.
pub fn unserved_hook(hook: &Hook, calls: Option<&Vec<String>>, provides: Option<&HashSet<String>>, entries: Option<&[String]>) -> Option<String> {
    if !hook.required || hook.default.is_some() {
        return None;
    }
    if let Some(needs) = &hook.entries
        && !entries.is_some_and(|run| run.iter().any(|e| needs.contains(e)))
    {
        return None;
    }
    if !hosts_hook(hook, calls) {
        return Some(NO_ROUTE.into());
    }
    if provides.is_some_and(|p| p.contains(&hook.nsid)) {
        return None;
    }
    Some(format!("this host was built without {}", crate::transport::name(&hook.nsid).unwrap_or(&hook.nsid)))
}

/// A hook route's answer as a kcp Verdict (com.publicdomainrelay.temp.market.evaluatePolicy): anything but allow: true, an
/// error included, is a deny; a deny without violations of its own names `policy_id` (JS verdictOf).
pub fn verdict_of(result: Option<&Value>, policy_id: &str, error: Option<&str>) -> Value {
    if error.is_none() && result.is_some_and(|r| r.get("allow") == Some(&Value::Bool(true))) {
        return json!({ "allow": true, "violations": [] });
    }
    let own = result.filter(|_| error.is_none()).and_then(|r| r.get("violations")).and_then(Value::as_array).filter(|v| !v.is_empty());
    match own {
        Some(v) => json!({ "allow": false, "violations": v }),
        None => json!({ "allow": false, "violations": [{ "msg": error.unwrap_or("no verdict"), "policyId": policy_id }] }),
    }
}

impl Module {
    /// JS moduleOwner: owner, else id, else sha256 (given, or of the bytes).
    pub fn owner(&self) -> String {
        self.owner
            .clone()
            .or_else(|| self.id.clone())
            .or_else(|| self.sha256.clone())
            .unwrap_or_else(|| crate::wasm::sha256_hex(&self.bytes))
    }
    /// module.json fields (calls, anyRecords, name, didkey). calls and anyRecords are
    /// absent or lists of strings (JS checkLists), else the module is refused.
    pub fn from_manifest(bytes: Vec<u8>, manifest: &Value) -> Result<Self, String> {
        let strings = |key: &str| -> Result<Option<Vec<String>>, String> {
            let Some(list) = manifest.get(key) else {
                return Ok(None);
            };
            list.as_array()
                .and_then(|a| a.iter().map(|v| v.as_str().map(String::from)).collect())
                .map(Some)
                .ok_or_else(|| format!("module {key} must be a list of strings"))
        };
        let hooks = match manifest.get("hooks") {
            None => vec![],
            Some(list) => list
                .as_array()
                .and_then(|l| l.iter().map(Hook::parse).collect())
                .ok_or("module hooks must be a list of { nsid, reason, required?: boolean, entries?: [entry], default?: module key, combine?: all-allow }")?,
        };
        let implements = match manifest.get("implements") {
            None => vec![],
            Some(list) => list
                .as_array()
                .and_then(|l| {
                    l.iter()
                        .map(|i| Some((i.get("nsid")?.as_str().filter(|n| nsid_ok(n))?.to_string(), i.get("entry")?.as_str()?.to_string())))
                        .collect()
                })
                .ok_or("module implements must be a list of { nsid, entry }")?,
        };
        let field = |f: &Value| -> Option<Field> {
            let kind = match f.get("type") {
                None => "string",
                Some(t) => t.as_str().filter(|t| FIELD_TYPES.contains(t))?,
            };
            let required = match f.get("required") {
                None => false,
                Some(r) => r.as_bool()?,
            };
            Some(Field { key: f.get("key")?.as_str()?.to_string(), kind: kind.into(), required })
        };
        let config = match manifest.get("config") {
            None => None,
            Some(list) => Some(list.as_array().and_then(|l| l.iter().map(field).collect()).ok_or(format!("module config must be a list of {{ key, label, type?: {}, required?: boolean }}", FIELD_TYPES.join(" | ")))?),
        };
        Ok(Self {
            config,
            pin: module_pin(&bytes, manifest),
            bytes: Arc::new(bytes),
            sha256: None,
            hooks,
            implements,
            any_hooks: manifest.get("anyHooks") == Some(&Value::Bool(true)),
            calls: strings("calls")?,
            any_records: strings("anyRecords")?,
            http: strings("http")?,
            name: manifest
                .get("name")
                .and_then(Value::as_str)
                .map(String::from),
            didkey: manifest.get("didkey").cloned(),
            session: manifest.get("session").cloned(),
            id: None,
            owner: None,
            key: None,
            scope: None,
        })
    }
}

pub struct Ctx {
    pub runner: Arc<dyn Runner>,
    pub ledger: Option<Arc<dyn Ledger>>,
    /// A fixed `now` for every event (tests); None is the clock.
    pub now: Option<String>,
}

impl Ctx {
    fn now(&self) -> String {
        self.now.clone().unwrap_or_else(iso_now)
    }
}

// --- validation (parseCall) ---

fn nsid_ok(s: &str) -> bool {
    let parts: Vec<&str> = s.split('.').collect();
    let seg = |p: &str| !p.is_empty() && p.chars().all(|c| c.is_ascii_alphanumeric() || c == '-');
    parts.len() >= 3
        && parts[0].starts_with(|c: char| c.is_ascii_alphabetic())
        && parts.iter().all(|p| seg(p))
}

fn proxy_ok(s: &str) -> bool {
    let Some(rest) = s.strip_prefix("did:") else {
        return false;
    };
    let Some((method, rest)) = rest.split_once(':') else {
        return false;
    };
    let Some((id, service)) = rest.split_once('#') else {
        return false;
    };
    !method.is_empty()
        && method.chars().all(|c| c.is_ascii_lowercase())
        && !id.is_empty()
        && !id.chars().any(|c| c.is_whitespace() || c == '\u{feff}')
        && !service.is_empty()
        && service
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
}

/// JS MIME: type/subtype, no parameters.
fn mime_ok(s: &str) -> bool {
    let part = |p: &str| p.chars().next().is_some_and(|c| c.is_ascii_alphanumeric()) && p.chars().all(|c| c.is_ascii_alphanumeric() || "!#$&^_.+-".contains(c));
    s.split_once('/').is_some_and(|(t, sub)| part(t) && part(sub))
}

const ENDPOINTS: &str = "Module call endpoints must be 1 to 8 { kind, address } (kind a URL scheme name, address up to 2048 printable ASCII)";

/// [{ kind, address }]: kind a lowercase scheme name, address printable ASCII (JS ENDPOINT).
fn endpoints_ok(value: &Value) -> bool {
    let Some(list) = value.as_array().filter(|l| (1..=8).contains(&l.len())) else { return false };
    list.iter().all(|e| {
        let (Some(kind), Some(address)) = (e.get("kind").and_then(Value::as_str), e.get("address").and_then(Value::as_str)) else { return false };
        e.as_object().is_some_and(|o| o.len() == 2)
            && (1..=16).contains(&kind.len())
            && kind.starts_with(|c: char| c.is_ascii_lowercase())
            && kind.chars().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit())
            && (1..=2048).contains(&address.len())
            && address.bytes().all(|b| (0x21..0x7f).contains(&b))
    })
}

fn service_ok(s: &str) -> bool {
    if let Some(host) = s.strip_prefix("https://") {
        let (name, port) = host
            .split_once(':')
            .map_or((host, None), |(n, p)| (n, Some(p)));
        return !name.is_empty()
            && name
                .chars()
                .all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '-')
            && port.is_none_or(|p| !p.is_empty() && p.chars().all(|c| c.is_ascii_digit()));
    }
    let id = s
        .strip_prefix("did:plc:")
        .or_else(|| s.strip_prefix("did:web:"));
    // did:plc:x or did:web:x, with an optional #service id (JS SERVICE)
    let (id, fragment) = id.map_or((None, None), |id| id.split_once('#').map_or((Some(id), None), |(i, f)| (Some(i), Some(f))));
    id.is_some_and(|id| !id.is_empty() && id.chars().all(|c| c.is_ascii_alphanumeric() || "._:%-".contains(c)))
        && fragment.is_none_or(|f| !f.is_empty() && f.chars().all(|c| c.is_ascii_alphanumeric() || "_-".contains(c)))
}

pub fn id_ok(s: &str) -> bool {
    (1..=256).contains(&s.len())
        && s.chars()
            .all(|c| c.is_ascii_alphanumeric() || "_.:-".contains(c))
}

#[derive(Clone, Debug, Default)]
pub struct Planned {
    pub nsid: String,
    pub params: Option<Map<String, Value>>,
    pub input: Option<Value>,
    pub encoding: Option<String>,
    pub proxy: Option<String>,
    pub service: Option<String>,
    pub endpoints: Option<Value>,
    pub subscribe: bool,
    pub timeout_ms: Option<f64>,
    pub framing: Option<String>,
    pub keepalive: Option<Value>,
    pub overflow: Option<String>,
    pub service_auth: Option<Value>,
}

impl Planned {
    pub fn request(&self, signal: Signal) -> Request {
        Request {
            params: self.params.clone(),
            input: self.input.clone(),
            encoding: self.encoding.clone(),
            proxy: self.proxy.clone(),
            service: self.service.clone(),
            endpoints: self.endpoints.clone(),
            subscribe: self.subscribe,
            framing: self.framing.clone(),
            keepalive: self.keepalive.clone(),
            service_auth: self.service_auth.clone(),
            headers: vec![],
            signal,
            emit: None,
        }
    }
}

fn opt_string(raw: &Value, key: &str, error: &str) -> Result<Option<String>, String> {
    match raw.get(key) {
        None => Ok(None),
        Some(Value::String(s)) => Ok(Some(s.clone())),
        Some(_) => Err(error.into()),
    }
}

pub fn parse_call(raw: &Value, allowed: Option<&[String]>) -> Result<Planned, String> {
    let nsid = raw
        .get("nsid")
        .and_then(Value::as_str)
        .filter(|s| nsid_ok(s))
        .ok_or("Module call needs an NSID")?;
    if allowed.is_some_and(|a| !a.iter().any(|n| n == nsid)) {
        return Err(format!("Module does not allow {nsid}"));
    }
    let params = match raw.get("params") {
        Some(Value::Object(p)) => Some(
            p.iter()
                .map(|(k, v)| (k.clone(), Value::String(js_string(Some(v)))))
                .collect(),
        ),
        Some(Value::Array(a)) => Some(
            a.iter()
                .enumerate()
                .map(|(i, v)| (i.to_string(), Value::String(js_string(Some(v)))))
                .collect(),
        ),
        _ => None,
    };
    let proxy = opt_string(raw, "proxy", "Module call proxy must be did#service")?;
    if proxy.as_deref().is_some_and(|p| !proxy_ok(p)) {
        return Err("Module call proxy must be did#service".into());
    }
    let service = opt_string(
        raw,
        "service",
        "Module call service must be an https origin, a DID or a did#service",
    )?;
    if service.as_deref().is_some_and(|s| !service_ok(s)) {
        return Err("Module call service must be an https origin, a DID or a did#service".into());
    }
    if proxy.is_some() && service.is_some() {
        return Err("Module call takes proxy or service, not both".into());
    }
    let input = raw.get("input").cloned();
    let endpoints = raw.get("endpoints").cloned();
    if let Some(e) = &endpoints {
        if !service.as_deref().is_some_and(|s| s.starts_with("did:") && s.contains('#')) {
            return Err("Module call endpoints need a did#service service".into());
        }
        if !endpoints_ok(e) {
            return Err(ENDPOINTS.into());
        }
    }
    if service.as_deref().is_some_and(|s| s.starts_with("did:")) && input.is_some() && endpoints.is_none() {
        return Err("Module call to a DID service is a read: no input".into());
    }
    // JS: the content-type of a { $bytes } input
    let encoding = opt_string(raw, "encoding", "Module call encoding must be a MIME type")?;
    if encoding.as_deref().is_some_and(|e| !mime_ok(e)) {
        return Err("Module call encoding must be a MIME type".into());
    }
    let bytes = input.as_ref().and_then(Value::as_object).is_some_and(|o| o.len() == 1 && o.get("$bytes").is_some_and(Value::is_string));
    if encoding.is_some() && !bytes {
        return Err("Module call encoding needs a $bytes input".into());
    }
    let mut planned = Planned {
        nsid: nsid.into(),
        params,
        input,
        encoding,
        proxy,
        service,
        endpoints,
        ..Default::default()
    };
    planned.subscribe = raw.get("subscribe") == Some(&Value::Bool(true));
    if let Some(t) = raw.get("timeoutMs") {
        match t.as_f64() {
            Some(ms) if ms > 0.0 => planned.timeout_ms = Some(ms),
            _ => return Err("Module call timeoutMs must be a positive number".into()),
        }
    }
    if planned.subscribe {
        // A host pseudo-call session (iroh.connect) takes input; a WebSocket subscription is a GET.
        if planned.input.is_some() && !planned.nsid.starts_with("com.fedfork.exec.") {
            return Err("Module subscribe takes params, not input".into());
        }
        if planned.proxy.is_some() {
            return Err(
                "Module subscribe cannot go through atproto-proxy; call the service directly"
                    .into(),
            );
        }
        match raw.get("framing") {
            None => {}
            Some(Value::String(f)) if f == "cbor" || f == "json" => {
                planned.framing = Some(f.clone())
            }
            Some(_) => return Err("Module subscribe framing must be cbor or json".into()),
        }
        if let Some(k) = raw.get("keepalive") {
            let ms = k
                .get("ms")
                .and_then(Value::as_f64)
                .filter(|ms| *ms >= 1000.0)
                .ok_or("Module subscribe keepalive needs ms >= 1000")?;
            let mut keepalive = Map::new();
            keepalive.insert("ms".into(), json!(ms_value(ms)));
            if let Some(data) = k.get("data") {
                keepalive.insert("data".into(), data.clone());
            }
            planned.keepalive = Some(Value::Object(keepalive));
        }
        match raw.get("overflow") {
            None => {}
            Some(Value::String(o)) if o == "drop" || o == "close" => {
                planned.overflow = Some(o.clone())
            }
            Some(_) => return Err("Module subscribe overflow must be drop or close".into()),
        }
        if let Some(auth) = raw.get("serviceAuth") {
            let bad = || "Module subscribe serviceAuth must be {} or { param }".to_string();
            let obj = auth.as_object().ok_or_else(bad)?;
            if obj.keys().any(|k| k != "param") {
                return Err(bad());
            }
            let param = match obj.get("param") {
                None => None,
                Some(Value::String(p))
                    if (1..=256).contains(&p.len())
                        && p.bytes().all(|b| b.is_ascii_alphanumeric() || b"_.-".contains(&b)) =>
                {
                    Some(p.clone())
                }
                Some(_) => return Err(bad()),
            };
            if !planned.service.as_deref().is_some_and(|s| s.starts_with("https://")) {
                return Err("Module subscribe serviceAuth needs an https service".into());
            }
            planned.service_auth = Some(match param {
                Some(p) => json!({ "param": p }),
                None => json!({}),
            });
        }
    }
    Ok(planned)
}

fn ms_value(ms: f64) -> Value {
    if ms.fract() == 0.0 {
        json!(ms as i64)
    } else {
        json!(ms)
    }
}

// --- the rkey guard ---

const CREATE: &str = "com.atproto.repo.createRecord";
const PUT: &str = "com.atproto.repo.putRecord";
const DELETE: &str = "com.atproto.repo.deleteRecord";
const APPLY: &str = "com.atproto.repo.applyWrites";

struct Write {
    action: Option<String>,
    collection: Option<Value>,
    uri: Option<String>,
}

fn writes_of(planned: &Planned) -> Vec<Write> {
    let empty = Value::Object(Map::new());
    let input = match &planned.input {
        Some(v @ (Value::Object(_) | Value::Array(_))) => v,
        _ => &empty,
    };
    let at = |collection: Option<&Value>, rkey: Option<&Value>| {
        rkey.map(|rkey| {
            format!(
                "at://{}/{}/{}",
                js_string(input.get("repo")),
                js_string(collection),
                js_string(Some(rkey))
            )
        })
    };
    let one = |action: &str, uri: bool| {
        let collection = input.get("collection");
        vec![Write {
            action: Some(action.into()),
            collection: collection.cloned(),
            uri: if uri {
                at(collection, input.get("rkey"))
            } else {
                None
            },
        }]
    };
    match planned.nsid.as_str() {
        CREATE => one("create", false),
        PUT => one("update", true),
        DELETE => one("delete", true),
        APPLY => input
            .get("writes")
            .and_then(Value::as_array)
            .map(|writes| {
                writes
                    .iter()
                    .map(|w| Write {
                        action: js_string(w.get("$type"))
                            .split('#')
                            .nth(1)
                            .map(String::from),
                        collection: w.get("collection").cloned(),
                        uri: at(w.get("collection"), w.get("rkey")),
                    })
                    .collect()
            })
            .unwrap_or_default(),
        _ => vec![],
    }
}

async fn refusal(
    ledger: Option<&dyn Ledger>,
    any_records: Option<&[String]>,
    planned: &Planned,
) -> Result<Option<String>, String> {
    for Write {
        action,
        collection,
        uri,
    } in writes_of(planned)
    {
        let listed = any_records.is_some_and(|a| {
            collection
                .as_ref()
                .and_then(Value::as_str)
                .is_some_and(|c| a.iter().any(|x| x == c))
        });
        if action.as_deref() == Some("create") || listed {
            continue;
        }
        let managed = match (&uri, ledger) {
            (Some(uri), Some(ledger)) => ledger.get(uri).await?.is_some(),
            _ => false,
        };
        if !managed {
            let what = uri
                .clone()
                .unwrap_or_else(|| js_string(collection.as_ref()));
            return Ok(Some(format!(
                "{} refused: {what} was not created by this module (list {} in module.json anyRecords to allow)",
                planned.nsid,
                js_string(collection.as_ref())
            )));
        }
    }
    Ok(None)
}

async fn remember(
    ledger: Option<&dyn Ledger>,
    planned: &Planned,
    output: Option<&Value>,
) -> Result<(), String> {
    let Some(ledger) = ledger else { return Ok(()) };
    for (i, Write { action, uri, .. }) in writes_of(planned).into_iter().enumerate() {
        let created = if planned.nsid == APPLY {
            output
                .and_then(|o| o.get("results"))
                .and_then(|r| r.get(i))
                .and_then(|r| r.get("uri"))
        } else {
            output.and_then(|o| o.get("uri"))
        };
        match (action.as_deref(), created, uri) {
            (Some("create"), Some(Value::String(created)), _) => {
                ledger
                    .put(created.clone(), json!({ "createdAt": iso_now() }))
                    .await?
            }
            (Some("delete"), _, Some(uri)) => ledger.delete(uri).await?,
            _ => {}
        }
    }
    Ok(())
}

// --- the run ---

#[derive(Clone, Copy, Debug)]
pub struct Budget {
    pub in_flight: f64,
    pub subscriptions: f64,
    pub calls: f64,
    /// calls per one-minute window (fixed, from the run's start), the guard for a run with calls unlimited
    pub calls_per_minute: f64,
    pub events: f64,
    pub wall_ms: f64,
    pub queue: f64,
    /// call records kept in the result: the last `log` (a long run, a server, stays bounded)
    pub log: f64,
    /// the wasm's input and output per event (JSON bytes), at most IO_BYTES_MAX
    pub io_bytes: f64,
    /// the run's held values together (HELD_NSID), JSON bytes
    pub held_bytes: f64,
}

/// JS IO_BYTES_MAX: the ceiling on ioBytes, whatever a host sets (an event's input sits in the 64 MiB
/// memory beside its parsed copy, what it decodes and its output).
pub const IO_BYTES_MAX: f64 = (16 << 20) as f64;

pub const BUDGET: Budget = Budget {
    in_flight: 6.0,
    subscriptions: 4.0,
    calls: 100.0,
    calls_per_minute: f64::INFINITY,
    events: 1000.0,
    wall_ms: 60_000.0,
    queue: 256.0,
    log: 1000.0,
    io_bytes: (1 << 20) as f64,
    held_bytes: (16 << 20) as f64,
};

/// The budget's JSON keys (JS BUDGET).
pub const BUDGET_KEYS: [&str; 10] = ["inFlight", "subscriptions", "calls", "callsPerMinute", "events", "wallMs", "queue", "log", "ioBytes", "heldBytes"];
/// JS UNLIMITED: the keys a host may set to "unlimited" (JSON has no Infinity); a manifest never can.
pub const UNLIMITED: [&str; 4] = ["calls", "callsPerMinute", "events", "wallMs"];

/// JS budgetValue: a positive finite number, or "unlimited" (INFINITY) for an UNLIMITED `key`
/// (anything else leaves the default). Without a key (a manifest's values), finite only.
pub fn budget_value(v: Option<&Value>, key: Option<&str>) -> Option<f64> {
    if v.and_then(Value::as_str) == Some("unlimited") && key.is_some_and(|k| UNLIMITED.contains(&k)) {
        return Some(f64::INFINITY);
    }
    v.and_then(Value::as_f64).filter(|n| n.is_finite() && *n > 0.0)
}

impl Budget {
    /// BUDGET with a JSON object's budget values (budget_value) over it.
    pub fn merged(limits: Option<&Value>) -> Budget {
        let mut b = BUDGET;
        if let Some(Value::Object(l)) = limits {
            for (key, slot) in [
                ("inFlight", &mut b.in_flight),
                ("subscriptions", &mut b.subscriptions),
                ("calls", &mut b.calls),
                ("callsPerMinute", &mut b.calls_per_minute),
                ("events", &mut b.events),
                ("wallMs", &mut b.wall_ms),
                ("queue", &mut b.queue),
                ("log", &mut b.log),
                ("ioBytes", &mut b.io_bytes),
                ("heldBytes", &mut b.held_bytes),
            ] {
                if let Some(n) = budget_value(l.get(key), Some(key)) {
                    *slot = n;
                }
            }
        }
        b.io_bytes = b.io_bytes.min(IO_BYTES_MAX);
        b
    }
}

pub type OnEvent = Arc<dyn Fn(&Value, &Value) + Send + Sync>;

#[derive(Default)]
pub struct Options {
    pub signal: Option<Signal>,
    /// Each Ok value becomes an `input` event; Err ends inputs with "inputs failed"; close ends them.
    pub inputs: Option<mpsc::UnboundedReceiver<Result<Value, String>>>,
    pub budget: Option<Value>,
    /// Called after each handler call with (event, output), for hosts that show progress.
    pub on_event: Option<OnEvent>,
    /// Resolves a hook default's module key; without one a hook with a default fails.
    pub load: Option<Loader>,
    /// The host pseudo-call NSIDs (transports) this host serves, checked against required hooks.
    pub provides: Option<Arc<HashSet<String>>>,
    /// The origins HTTP_NSID may reach, with the headers the host adds (None: none).
    pub http: Option<Arc<Http>>,
    /// Held values the run starts with (JS runModule's held): name -> value, counted against heldBytes.
    pub held: Option<Map<String, Value>>,
    /// The call that mints http's serviceAuth tokens (JS runModule's session; None: the run's call). A
    /// composition passes the instance's session (call_for), not its call, which holds the module to its calls.
    pub session: Option<Arc<dyn Call>>,
}

/// The host pseudo-call for plain HTTP (not XRPC): input { method?, url, headers?, body? } -> { status, headers, body }.
pub const HTTP_NSID: &str = "com.fedfork.exec.http.fetch";

/// JS HELD_NSID: held values, JSON a run keeps in the host by name instead of in its state. put { name,
/// value }, append { name, items } -> { length, bytes }, get { name } -> { value }, delete { name }; answered at
/// plan time, in order; { "$held": name } in an http.fetch body or the result is replaced by the value.
/// Per run, at most budget heldBytes, gone when the run ends.
pub const HELD_NSID: &str = "com.fedfork.exec.held.";

#[derive(Default)]
struct Held {
    values: HashMap<String, (Value, usize)>,
    total: usize,
    limit: usize,
}

fn json_len(value: &Value) -> usize {
    serde_json::to_vec(value).map_or(0, |b| b.len())
}

impl Held {
    fn keep(&mut self, nsid: &str, name: &str, bytes: usize) -> Result<(), String> {
        let was = self.values.get(name).map_or(0, |v| v.1);
        if self.total - was + bytes > self.limit {
            return Err(format!("{nsid}: held values past {} bytes", self.limit));
        }
        self.total = self.total - was + bytes;
        Ok(())
    }

    fn op(&mut self, nsid: &str, input: Option<&Value>) -> Result<Value, String> {
        let name = input.and_then(|i| i.get("name")).and_then(Value::as_str).filter(|n| id_ok(n)).ok_or_else(|| format!("{nsid}: input needs a name"))?.to_string();
        let input = input.cloned().unwrap_or_default();
        match &nsid[HELD_NSID.len()..] {
            "put" => {
                let value = input.get("value").cloned().ok_or_else(|| format!("{nsid}: input needs a value"))?;
                let bytes = json_len(&value);
                self.keep(nsid, &name, bytes)?;
                self.values.insert(name, (value, bytes));
                Ok(json!({}))
            }
            "append" => {
                let items = input.get("items").and_then(Value::as_array).ok_or_else(|| format!("{nsid}: items must be a list"))?;
                let (len, was) = match self.values.get(&name) {
                    Some((Value::Array(list), bytes)) => (list.len(), *bytes),
                    Some(_) => return Err(format!("{nsid}: {name} is not a list")),
                    None => (0, 2),
                };
                // JSON bytes of the longer list: "[" + items joined by "," + "]"
                let bytes = was + items.iter().map(json_len).sum::<usize>() + (len + items.len()).saturating_sub(1) - len.saturating_sub(1);
                self.keep(nsid, &name, bytes)?;
                let entry = self.values.entry(name).or_insert((Value::Array(vec![]), 2));
                if let Value::Array(list) = &mut entry.0 {
                    list.extend(items.iter().cloned());
                }
                entry.1 = bytes;
                Ok(json!({ "length": entry.0.as_array().map_or(0, Vec::len), "bytes": bytes }))
            }
            "get" => Ok(json!({ "value": self.values.get(&name).map_or(Value::Null, |v| v.0.clone()) })),
            "delete" => {
                if let Some((_, bytes)) = self.values.remove(&name) {
                    self.total -= bytes;
                }
                Ok(json!({}))
            }
            _ => Err(format!("{nsid} is not a held-value call (put, append, get, delete)")),
        }
    }

    /// { "$held": name } anywhere in value, replaced by that held value; { "$held": [names] }: those
    /// held lists, concatenated.
    fn splice(&self, value: &Value) -> Result<Value, String> {
        let get = |name: &str| self.values.get(name).map(|v| &v.0).ok_or_else(|| format!("no held value {name}"));
        Ok(match value {
            Value::Array(items) => Value::Array(items.iter().map(|v| self.splice(v)).collect::<Result<_, _>>()?),
            Value::Object(o) if o.len() == 1 && o.get("$held").is_some_and(Value::is_string) => get(o["$held"].as_str().unwrap_or_default())?.clone(),
            Value::Object(o) if o.len() == 1 && o.get("$held").and_then(Value::as_array).is_some_and(|n| n.iter().all(Value::is_string)) => {
                let mut out = vec![];
                for name in o["$held"].as_array().into_iter().flatten().filter_map(Value::as_str) {
                    out.extend(get(name)?.as_array().ok_or_else(|| format!("held value {name} is not a list"))?.iter().cloned());
                }
                Value::Array(out)
            }
            Value::Object(o) => Value::Object(o.iter().map(|(k, v)| Ok((k.clone(), self.splice(v)?))).collect::<Result<_, String>>()?),
            other => other.clone(),
        })
    }
}
const METHODS: [&str; 6] = ["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE"];

/// The host's HTTP allowlist (JS runModule's `http`): origin -> the headers it adds (API keys the wasm
/// never sees) and the service auth it mints per request, (aud, lxm); and the fetch that makes the
/// requests. That fetch must not follow redirects. An entry "<scheme>://*" allows that scheme's URLs,
/// fetched by `schemes` ("<scheme>:" -> fetch; JS runModule's `fetchers`; transport::fetchers), never
/// with the host's headers or service auth.
pub struct Http {
    pub fetch: Arc<dyn crate::host::Fetch>,
    pub allow: Vec<(String, Vec<(String, String)>, Option<(String, String)>)>,
    pub schemes: HashMap<String, Arc<dyn crate::host::Fetch>>,
}

const GET_SERVICE_AUTH: &str = "com.atproto.server.getServiceAuth";

impl Http {
    /// From the JSON the hosts take (ATLIBS_HTTP, the bridge): { origin: { headers?: { name: value }, serviceAuth?: { aud, lxm } } }.
    pub fn from_json(fetch: Arc<dyn crate::host::Fetch>, value: &Value) -> Result<Self, String> {
        let bad = || "http must be { origin: { headers?: { name: value }, serviceAuth?: { aud, lxm } } }".to_string();
        let mut allow = vec![];
        for (origin, grant) in value.as_object().ok_or_else(bad)? {
            let headers = match grant.get("headers") {
                None => vec![],
                Some(h) => h.as_object().ok_or_else(bad)?.iter().map(|(k, v)| v.as_str().map(|v| (k.clone(), v.to_string())).ok_or_else(bad)).collect::<Result<_, _>>()?,
            };
            let service_auth = match grant.get("serviceAuth") {
                None => None,
                Some(a) => match (a["aud"].as_str(), a["lxm"].as_str()) {
                    (Some(aud), Some(lxm)) if aud.starts_with("did:") && !lxm.is_empty() => Some((aud.to_string(), lxm.to_string())),
                    _ => return Err(format!("http: {origin} serviceAuth must be {{ aud: a DID, lxm: an NSID }}")),
                },
            };
            allow.push((entry_key(origin).ok_or_else(|| format!("http: {origin} is not an http(s) origin or <scheme>://*"))?, headers, service_auth));
        }
        Ok(Self { fetch, allow, schemes: HashMap::new() })
    }
}

/// An http(s) URL's origin, None for anything else or a URL carrying credentials (JS httpOrigin).
pub fn http_origin(url: &str) -> Option<String> {
    let u = url::Url::parse(url).ok()?;
    (matches!(u.scheme(), "http" | "https") && u.username().is_empty() && u.password().is_none()).then(|| u.origin().ascii_serialization())
}

/// An http list or allowlist entry (JS entryKey): an http(s) origin, or "<scheme>://*" for a fetcher's scheme.
pub fn entry_key(entry: &str) -> Option<String> {
    match entry.strip_suffix("://*") {
        Some(scheme) => {
            let ok = scheme.starts_with(|c: char| c.is_ascii_lowercase()) && scheme.chars().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || "+.-".contains(c));
            (ok && !matches!(scheme, "http" | "https")).then(|| entry.to_string())
        }
        None => http_origin(entry),
    }
}

/// The entry a request URL needs (JS allowKey): its origin for http(s), else "<scheme>://*". None with credentials.
pub fn allow_key(url: &str) -> Option<String> {
    let u = url::Url::parse(url).ok()?;
    if !u.username().is_empty() || u.password().is_some() {
        return None;
    }
    Some(if matches!(u.scheme(), "http" | "https") { u.origin().ascii_serialization() } else { format!("{}://*", u.scheme()) })
}

fn token_name(s: &str) -> bool {
    !s.is_empty() && s.chars().all(|c| c.is_ascii_alphanumeric() || "!#$%&'*+.^_`|~-".contains(c))
}

/// JS withHttp: HTTP_NSID served by the engine for the origins the module declares and the host allows.
struct HttpCall {
    inner: Arc<dyn Call>,
    /// mints serviceAuth tokens: Options session, else inner
    mint: Arc<dyn Call>,
    declared: Vec<String>,
    http: Option<Arc<Http>>,
}

impl HttpCall {
    async fn fetch(&self, request: Request) -> Result<Value, String> {
        let fail = |why: String| format!("{HTTP_NSID}: {why}");
        let input = request.input.as_ref().and_then(Value::as_object).ok_or_else(|| fail("input must be { method?, url, headers?, body? }".into()))?;
        let method = match input.get("method") {
            None => "GET",
            Some(m) => m.as_str().filter(|m| METHODS.contains(m)).ok_or_else(|| fail(format!("method must be one of {}", METHODS.join(", "))))?,
        };
        let url = input.get("url").and_then(Value::as_str).unwrap_or_default();
        let origin = allow_key(url).ok_or_else(|| fail("url must be a URL without credentials".into()))?;
        let headers: Vec<(String, String)> = match input.get("headers") {
            None => vec![],
            Some(h) => h
                .as_object()
                .and_then(|h| h.iter().map(|(k, v)| v.as_str().filter(|v| token_name(k) && !v.contains(['\r', '\n', '\0'])).map(|v| (k.clone(), v.to_string()))).collect())
                .ok_or_else(|| fail("headers must map header names to strings".into()))?,
        };
        let body = match input.get("body") {
            None => None,
            Some(Value::String(b)) => Some(b.clone()),
            Some(_) => return Err(fail("body must be a string".into())),
        };
        if body.is_some() && matches!(method, "GET" | "HEAD") {
            return Err(fail(format!("{method} takes no body")));
        }
        let plain = http_origin(url).is_some();
        if !self.declared.iter().any(|d| (d == "*" && plain) || entry_key(d).as_ref() == Some(&origin)) {
            return Err(fail(format!("{origin} is not in the module's http list")));
        }
        let http = self.http.as_ref().ok_or_else(|| fail(format!("this host does not allow {origin}")))?;
        let (_, injected, service_auth) = http.allow.iter().find(|(o, ..)| *o == origin).ok_or_else(|| fail(format!("this host does not allow {origin}")))?;
        // another scheme: its fetcher, with the module's headers only
        let (fetch, injected, service_auth) = if plain {
            (&http.fetch, injected.clone(), service_auth.as_ref())
        } else {
            let scheme = format!("{}:", origin.trim_end_matches("://*"));
            (http.schemes.get(&scheme).ok_or_else(|| fail(format!("no fetcher for {scheme}")))?, vec![], None)
        };
        let mut injected = injected;
        if let Some((aud, lxm)) = service_auth {
            // minted per request with the run's session (its scope must grant rpc:<lxm>?aud=<aud>); never cached: a token may carry a jti nonce
            let params = json!({ "aud": aud, "lxm": lxm }).as_object().cloned();
            let answer = self.mint.call(GET_SERVICE_AUTH.into(), Request { params, signal: request.signal.clone(), ..Default::default() }).await.map_err(|e| {
                fail(format!("{origin} needs a service-auth token (aud {aud}, lxm {lxm}) from the run's session: {e} (an OAuth session's scope must include rpc:{lxm}?aud={aud})"))
            })?;
            let token = answer.as_ref().and_then(|a| a["token"].as_str()).filter(|t| !t.is_empty()).ok_or_else(|| fail(format!("{GET_SERVICE_AUTH} answered no service-auth token")))?;
            injected.push(("authorization".into(), format!("Bearer {token}")));
        }
        let mut sent: Vec<(String, String)> = headers.into_iter().filter(|(k, _)| !injected.iter().any(|(i, _)| i.eq_ignore_ascii_case(k))).collect();
        sent.extend(injected.iter().cloned());
        let response = tokio::select! {
            r = fetch.fetch(method, url, sent, body.map(String::into_bytes)) => r?,
            reason = request.signal.aborted() => return Err(reason),
        };
        // JS Headers: lowercase names, repeated ones joined with ", "
        let mut out = Map::new();
        for (k, v) in response.headers {
            let k = k.to_ascii_lowercase();
            let joined = match out.get(&k).and_then(Value::as_str) {
                Some(prev) => format!("{prev}, {v}"),
                None => v,
            };
            out.insert(k, Value::String(joined));
        }
        Ok(json!({ "status": response.status, "headers": out, "body": String::from_utf8_lossy(&response.body) }))
    }
}

impl Call for HttpCall {
    fn call(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Option<Value>, String>> {
        Box::pin(async move {
            if nsid != HTTP_NSID {
                return self.inner.call(nsid, request).await;
            }
            self.fetch(request).await.map(Some)
        })
    }
    fn subscribe(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Sub, String>> {
        Box::pin(async move {
            if nsid == HTTP_NSID {
                return Err(format!("{HTTP_NSID}: cannot be subscribed to (a response arrives whole)"));
            }
            self.inner.subscribe(nsid, request).await
        })
    }
}

/// A hook default's module key -> the module, the runner for its bytes, and the call it runs on: its own
/// (what the host gives that module for its calls), never the caller's.
pub type Loader = Arc<dyn Fn(String) -> BoxFut<'static, Result<(Module, Arc<dyn Runner>, Arc<dyn Call>), String>> + Send + Sync>;

// A hook in a run with no composition: one with a default is served by that module's
// implements entry, as a nested run on the default's own call (the loader's); one without answers null
// (skip). Never sent to `inner`.
struct HookCall {
    inner: Arc<dyn Call>,
    module: Module,
    ctx: Ctx,
    load: Option<Loader>,
    provides: Option<Arc<HashSet<String>>>,
    budget: Option<Value>,
    http: Option<Arc<Http>>,
    loaded: tokio::sync::Mutex<HashMap<String, (Module, Arc<dyn Runner>, Arc<dyn Call>)>>,
}

impl HookCall {
    async fn serve(&self, hook: &Hook, key: &str, nsid: &str, request: Request) -> Result<Value, String> {
        let (target, runner, call) = {
            let mut loaded = self.loaded.lock().await;
            match loaded.get(key) {
                Some(m) => m.clone(),
                None => {
                    let from = self.module.id.as_deref().unwrap_or("bytes");
                    let load = self.load.clone().ok_or_else(|| format!("hook default {key}: no loader for a module loaded from {from} (pass the host's load option)"))?;
                    let m = load(key.to_string()).await?;
                    // a default runs on its own calls allowlist; one that lists none would get them all
                    if m.0.calls.is_none() {
                        return Err(format!("hook default {key}: its module lists no calls (a hook default must)"));
                    }
                    loaded.insert(key.to_string(), m.clone());
                    m
                }
            }
        };
        let ctx = Ctx { runner, ledger: self.ctx.ledger.clone(), now: self.ctx.now.clone() };
        let entry = target
            .implements
            .iter()
            .find(|(n, _)| n == nsid)
            .map(|(_, e)| e.clone())
            .ok_or_else(|| format!("hook default {key} does not implement {nsid}"))?;
        let mut body = request.params.clone().unwrap_or_default();
        match request.input {
            Some(Value::Object(m)) => body.extend(m),
            Some(other) => drop(body.insert("input".into(), other)),
            None => {}
        }
        let instance = self.module.name.clone().unwrap_or_else(|| "module".into());
        body.insert("caller".into(), json!({ "instance": instance, "nsid": nsid, "reason": hook.reason }));
        let options = Options { signal: Some(request.signal.clone()), budget: self.budget.clone(), load: self.load.clone(), provides: self.provides.clone(), http: self.http.clone(), ..Default::default() };
        let ran: BoxFut<'_, Value> = Box::pin(run_module(&ctx, &target, &entry, &body, call, options));
        let ran = ran.await;
        match ran.get("error") {
            Some(e) if !e.is_null() => Err(js_string(Some(e))),
            _ => Ok(ran.get("result").cloned().unwrap_or(Value::Null)),
        }
    }
}

impl Call for HookCall {
    fn call(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Option<Value>, String>> {
        Box::pin(async move {
            // a transport the module lists in calls is the host's (hosts_hook)
            let Some(hook) = self.module.hooks.iter().find(|h| h.nsid == nsid).filter(|h| !hosts_hook(h, self.module.calls.as_ref())) else {
                return self.inner.call(nsid, request).await;
            };
            if request.proxy.is_some() || request.service.is_some() || request.subscribe {
                return Err(format!("{nsid} is a hook: no proxy, service or subscribe"));
            }
            let Some(key) = &hook.default else {
                return Ok(Some(Value::Null));
            };
            let out = self.serve(hook, key, &nsid, request).await;
            if hook.all_allow {
                return Ok(Some(match &out {
                    Ok(r) => verdict_of(Some(r), key, None),
                    Err(e) => verdict_of(None, key, Some(e)),
                }));
            }
            out.map(Some)
        })
    }

    fn subscribe(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Sub, String>> {
        Box::pin(async move {
            if self.module.hooks.iter().any(|h| h.nsid == nsid && !hosts_hook(h, self.module.calls.as_ref())) {
                return Err(format!("{nsid} is a hook: no proxy, service or subscribe"));
            }
            self.inner.subscribe(nsid, request).await
        })
    }
}

enum Msg {
    Result(String, u64, Value),
    Message(String, u64, Value),
    Closed(String, u64, Option<String>),
    Timer(String, u64),
    Refuse(String, String, String),
    Callback(String, u64, Value),
    Input(Value),
    InputsEnd(Option<String>),
    Deadline,
    Abort,
}

#[derive(PartialEq, Clone, Copy)]
enum Kind {
    Queued,
    Call,
    Sub,
    Timer,
}

enum Ctl {
    Send(Value),
    Close(String),
}

struct Slot {
    generation: u64,
    kind: Kind,
    planned: Option<Planned>,
    abort: Signal,
    messages: u64,
    ctl: Option<mpsc::UnboundedSender<Ctl>>,
    task: Option<JoinHandle<()>>,
    timer: Option<JoinHandle<()>>,
}

struct Run<'a> {
    ctx: &'a Ctx,
    module: &'a Module,
    call: Arc<dyn Call>,
    budget: Budget,
    tx: mpsc::UnboundedSender<Msg>,
    ended: Signal,
    pending: HashMap<String, Slot>,
    /// cancelled subscriptions still flushing their sends: the run's end waits for them as for open ones
    closing: Vec<JoinHandle<()>>,
    held: Held,
    waiting: VecDeque<String>,
    queue: VecDeque<Value>,
    dropped: HashMap<String, u64>,
    log: VecDeque<Value>,
    log_dropped: u64,
    error: Option<String>,
    total: f64,
    /// the callsPerMinute window: its start and the calls in it
    minute: (std::time::Instant, f64),
    generation: u64,
    inputs_open: bool,
}

fn event_id(event: &Value) -> Option<&str> {
    event.get("id").and_then(Value::as_str)
}

fn list(value: Option<&Value>) -> Vec<Value> {
    value.and_then(Value::as_array).cloned().unwrap_or_default()
}

impl Run<'_> {
    fn current(&self, id: &str, generation: u64) -> bool {
        self.pending
            .get(id)
            .is_some_and(|s| s.generation == generation)
    }
    fn count(&self, kind: Kind) -> usize {
        self.pending.values().filter(|s| s.kind == kind).count()
    }
    fn closes(&self, id: Option<&str>) -> bool {
        id.and_then(|id| self.pending.get(id))
            .and_then(|s| s.planned.as_ref())
            .and_then(|p| p.overflow.as_deref())
            == Some("close")
    }

    // Only the last budget.log call records are kept; the result counts the rest (callsDropped).
    fn record(&mut self, entry: Value) {
        self.log.push_back(entry);
        if self.log.len() as f64 > self.budget.log {
            self.log.pop_front();
            self.log_dropped += 1;
        }
    }

    // Past budget.queue the oldest message is dropped (results, closes and timers never are) and the
    // next message on its subscription carries the count; overflow "close" closes instead.
    fn push(&mut self, event: Value) {
        let message = event.get("type").and_then(Value::as_str) == Some("message");
        if self.queue.len() as f64 >= self.budget.queue && message {
            if self.closes(event_id(&event)) {
                return self.overflow(event_id(&event).unwrap_or_default().to_string());
            }
            let oldest = self.queue.iter().position(|e| {
                e.get("type").and_then(Value::as_str) == Some("message")
                    && !self.closes(event_id(e))
            });
            let victim = match oldest {
                Some(i) => self.queue.remove(i).expect("index from position"),
                None => event.clone(),
            };
            *self
                .dropped
                .entry(event_id(&victim).unwrap_or_default().to_string())
                .or_default() += 1;
            if oldest.is_none() {
                return;
            }
        }
        self.queue.push_back(event);
    }

    fn overflow(&mut self, id: String) {
        if let Some(slot) = self.pending.get(&id) {
            slot.abort.abort("overflow");
            self.settle(
                &id,
                json!({ "type": "closed", "id": id, "error": "overflow" }),
            );
        }
    }

    fn refuse(&mut self, id: Option<Value>, nsid: Option<Value>, error: String) {
        let mut entry = Map::new();
        if let Some(id) = id.clone() {
            entry.insert("id".into(), id);
        }
        if let Some(nsid) = nsid.clone() {
            entry.insert("nsid".into(), nsid);
        }
        entry.insert("ok".into(), json!(false));
        entry.insert("error".into(), json!(error));
        self.record(Value::Object(entry.clone()));
        let mut event = Map::new();
        event.insert("type".into(), json!("result"));
        event.extend(entry);
        self.push(Value::Object(event));
    }

    // A call or subscription ended on its own (not cancelled).
    fn settle(&mut self, id: &str, event: Value) {
        let Some(slot) = self.pending.remove(id) else {
            return;
        };
        if let Some(timer) = &slot.timer {
            timer.abort();
        }
        let planned = slot.planned.as_ref().expect("calls have a plan");
        let error = event.get("error").filter(|e| !e.is_null()).cloned();
        let mut entry = Map::new();
        entry.insert("id".into(), json!(id));
        entry.insert("nsid".into(), json!(planned.nsid));
        if planned.subscribe {
            entry.insert("subscribe".into(), json!(true));
            entry.insert("messages".into(), json!(slot.messages));
            entry.insert("ok".into(), json!(error.is_none()));
            if let Some(e) = error {
                entry.insert("error".into(), e);
            }
        } else {
            let ok = event.get("ok") == Some(&Value::Bool(true));
            entry.insert("ok".into(), json!(ok));
            if !ok {
                entry.insert("error".into(), error.unwrap_or(Value::Null));
            }
        }
        self.record(Value::Object(entry));
        self.push(event);
        self.pump();
    }

    fn start(&mut self, id: String) {
        let tx = self.tx.clone();
        let call = self.call.clone();
        let ledger = self.ctx.ledger.clone();
        let ended = self.ended.clone();
        let slot = self
            .pending
            .get_mut(&id)
            .expect("started slots are pending");
        let planned = slot.planned.clone().expect("calls have a plan");
        let generation = slot.generation;
        slot.kind = if planned.subscribe {
            Kind::Sub
        } else {
            Kind::Call
        };
        let signal = Signal::any(&slot.abort, &ended);
        if let Some(ms) = planned.timeout_ms {
            let (abort, nsid) = (slot.abort.clone(), planned.nsid.clone());
            // a call cannot outlive the run: at most wallMs
            let wait = ms_duration(ms.min(self.budget.wall_ms));
            slot.timer = Some(tokio::spawn(async move {
                tokio::time::sleep(wait).await;
                abort.abort(format!("{nsid} timed out after {} ms", js_num(ms)));
            }));
        }
        let mut request = planned.request(signal.clone());
        request.emit = Some(Emit {
            tx: tx.clone(),
            id: id.clone(),
            generation,
        });
        if !planned.subscribe {
            // The write is remembered even when the module stopped waiting for it.
            let made = tokio::spawn(async move {
                let output = call.call(planned.nsid.clone(), request).await?;
                remember(ledger.as_deref(), &planned, output.as_ref()).await?;
                Ok::<_, String>((planned.nsid, output))
            });
            let nsid = self.pending[&id]
                .planned
                .as_ref()
                .map(|p| p.nsid.clone())
                .unwrap_or_default();
            tokio::spawn(async move {
                let event = tokio::select! {
                    made = made => match made {
                        Ok(Ok((nsid, output))) => {
                            let mut e = json!({ "type": "result", "id": id, "nsid": nsid, "ok": true });
                            if let Some(output) = output { e["output"] = output; }
                            e
                        }
                        Ok(Err(error)) => json!({ "type": "result", "id": id, "nsid": nsid, "ok": false, "error": error }),
                        Err(join) => json!({ "type": "result", "id": id, "nsid": nsid, "ok": false, "error": join.to_string() }),
                    },
                    reason = signal.aborted() => json!({ "type": "result", "id": id, "nsid": nsid, "ok": false, "error": reason }),
                };
                let _ = tx.send(Msg::Result(id, generation, event));
            });
            return;
        }
        let (ctl_tx, ctl_rx) = mpsc::unbounded_channel();
        slot.ctl = Some(ctl_tx);
        let abort = slot.abort.clone();
        slot.task = Some(tokio::spawn(subscription(
            id, generation, planned, request, call, signal, abort, ctl_rx, tx,
        )));
    }

    fn pump(&mut self) {
        while !self.waiting.is_empty() && (self.count(Kind::Call) as f64) < self.budget.in_flight {
            let id = self.waiting.pop_front().expect("not empty");
            self.start(id);
        }
    }

    async fn plan(&mut self, raw: &Value) {
        let id = raw.get("id").cloned();
        let nsid = raw.get("nsid").cloned();
        let Some(id_str) = id
            .as_ref()
            .and_then(Value::as_str)
            .filter(|s| id_ok(s))
            .map(String::from)
        else {
            return self.refuse(id, nsid, "Module call needs an id".into());
        };
        if self.pending.contains_key(&id_str) {
            return self.refuse(
                id,
                nsid,
                format!("Module call id {id_str} is already pending"),
            );
        }
        let mut planned = match parse_call(raw, self.module.calls.as_deref()) {
            Ok(p) => p,
            Err(e) => return self.refuse(id, nsid, e),
        };
        self.total += 1.0;
        if self.total > self.budget.calls {
            return self.refuse(
                id,
                nsid,
                format!("Module exceeded {} calls", js_num(self.budget.calls)),
            );
        }
        if self.minute.0.elapsed() >= Duration::from_secs(60) {
            self.minute = (std::time::Instant::now(), 0.0);
        }
        self.minute.1 += 1.0;
        if self.minute.1 > self.budget.calls_per_minute {
            return self.refuse(
                id,
                nsid,
                format!("Module exceeded {} calls per minute", js_num(self.budget.calls_per_minute)),
            );
        }
        if planned.subscribe && self.count(Kind::Sub) as f64 >= self.budget.subscriptions {
            return self.refuse(
                id,
                nsid,
                format!(
                    "Module exceeded {} subscriptions",
                    js_num(self.budget.subscriptions)
                ),
            );
        }
        if planned.nsid.starts_with(HELD_NSID) {
            let out = if planned.subscribe { Err(format!("{} cannot be subscribed to", planned.nsid)) } else { self.held.op(&planned.nsid, planned.input.as_ref()) };
            return match out {
                Err(e) => self.refuse(id, nsid, e),
                Ok(output) => {
                    self.record(json!({ "id": id, "nsid": nsid, "ok": true }));
                    self.push(json!({ "type": "result", "id": id, "nsid": nsid, "ok": true, "output": output }));
                }
            };
        }
        // An http.fetch body that is JSON (not text) goes as its JSON, held values spliced in.
        if planned.nsid == HTTP_NSID
            && let Some(body) = planned.input.as_ref().and_then(|i| i.get("body")).filter(|b| !b.is_string())
        {
            match self.held.splice(body) {
                Ok(v) => planned.input.as_mut().expect("has a body")["body"] = json!(v.to_string()),
                Err(e) => return self.refuse(id, nsid, format!("{HTTP_NSID}: {e}")),
            }
        }
        match refusal(
            self.ctx.ledger.as_deref(),
            self.module.any_records.as_deref(),
            &planned,
        )
        .await
        {
            Ok(None) => {}
            Ok(Some(refused)) | Err(refused) => return self.refuse(id, nsid, refused),
        }
        self.generation += 1;
        let subscribe = planned.subscribe;
        self.pending.insert(
            id_str.clone(),
            Slot {
                generation: self.generation,
                kind: Kind::Queued,
                planned: Some(planned),
                abort: Signal::new(),
                messages: 0,
                ctl: None,
                task: None,
                timer: None,
            },
        );
        if subscribe {
            self.start(id_str);
        } else {
            self.waiting.push_back(id_str);
        }
    }

    // Events already queued for `id` of these types are dropped (a cancelled call yields
    // no result; a re-armed timer fires once), and its dropped-message count with them.
    fn purge(&mut self, id: &str, types: &[&str]) {
        self.queue.retain(|e| event_id(e) != Some(id) || !e.get("type").and_then(Value::as_str).is_some_and(|t| types.contains(&t)));
        self.dropped.remove(id);
    }

    // A cancelled call yields nothing; a cancelled subscription yields closed now.
    fn cancel(&mut self, id: &Value) {
        let Some(id) = id.as_str() else { return };
        self.purge(id, &["result", "message", "timer", "callback"]);
        let Some(mut slot) = self.pending.remove(id) else {
            return;
        };
        if let Some(timer) = &slot.timer {
            timer.abort();
        }
        if let Some(task) = slot.task.take() {
            self.closing.push(task);
        }
        match slot.kind {
            Kind::Timer => return,
            Kind::Queued => return self.waiting.retain(|w| w != id),
            _ => {}
        }
        let planned = slot.planned.as_ref().expect("calls have a plan");
        let mut entry = json!({ "id": id, "nsid": planned.nsid, "cancelled": true });
        if slot.kind == Kind::Sub {
            entry["subscribe"] = json!(true);
            entry["messages"] = json!(slot.messages);
        }
        self.record(entry);
        if slot.kind != Kind::Sub {
            return slot.abort.abort("cancelled");
        }
        self.push(json!({ "type": "closed", "id": id, "error": "cancelled" }));
        // Frames already sent go out first (bounded by DRAIN_MS), then the stream closes.
        if let Some(ctl) = &slot.ctl {
            let _ = ctl.send(Ctl::Close("cancelled".into()));
        }
        let abort = slot.abort.clone();
        tokio::spawn(async move {
            tokio::time::sleep(Duration::from_millis(DRAIN_MS)).await;
            abort.abort("cancelled");
        });
    }

    // Frames go out in order per subscription; a send that fails closes the subscription.
    fn send(&mut self, id: Option<&Value>, data: Option<&Value>) {
        let slot = id
            .and_then(Value::as_str)
            .and_then(|id| self.pending.get(id))
            .filter(|s| s.kind == Kind::Sub);
        match slot.and_then(|s| s.ctl.as_ref()) {
            Some(ctl) => {
                let _ = ctl.send(Ctl::Send(data.cloned().unwrap_or(Value::Null)));
            }
            None => self.refuse(
                id.cloned(),
                None,
                format!("send: {} is not an open subscription", js_string(id)),
            ),
        }
    }

    fn set_timer(&mut self, timer: &Value) -> Option<String> {
        let id = timer.get("id").and_then(Value::as_str).filter(|s| id_ok(s));
        let ms = timer
            .get("ms")
            .and_then(Value::as_f64)
            .filter(|ms| *ms >= 0.0);
        let (Some(id), Some(ms)) = (id, ms) else {
            return Some("Module timer needs an id and ms".into());
        };
        if let Some(old) = self.pending.get(id) {
            if old.kind != Kind::Timer {
                return Some(format!("Module timer id {id} is already pending"));
            }
            if let Some(t) = &old.timer {
                t.abort();
            }
        }
        self.purge(id, &["timer"]);
        self.generation += 1;
        let (tx, generation, owned) = (self.tx.clone(), self.generation, id.to_string());
        let wait = ms_duration(ms.min(self.budget.wall_ms));
        let handle = tokio::spawn(async move {
            tokio::time::sleep(wait).await;
            let _ = tx.send(Msg::Timer(owned, generation));
        });
        self.pending.insert(
            id.into(),
            Slot {
                generation,
                kind: Kind::Timer,
                planned: None,
                abort: Signal::new(),
                messages: 0,
                ctl: None,
                task: None,
                timer: Some(handle),
            },
        );
        None
    }

    fn ingest(&mut self, msg: Msg) {
        match msg {
            Msg::Result(id, g, event) if self.current(&id, g) => self.settle(&id, event),
            Msg::Closed(id, g, error) if self.current(&id, g) => {
                let mut event = json!({ "type": "closed", "id": id });
                if let Some(e) = error {
                    event["error"] = json!(e);
                }
                self.settle(&id, event);
            }
            Msg::Message(id, g, data) if self.current(&id, g) => {
                self.pending.get_mut(&id).expect("current").messages += 1;
                self.push(json!({ "type": "message", "id": id, "data": data }));
            }
            Msg::Timer(id, g) if self.current(&id, g) => {
                self.pending.remove(&id);
                self.push(json!({ "type": "timer", "id": id }));
            }
            Msg::Callback(id, g, mut event) if self.current(&id, g) => {
                if !event.is_object() {
                    event = json!({});
                }
                event["type"] = json!("callback");
                event["id"] = json!(id);
                self.push(event);
            }
            Msg::Refuse(id, nsid, error) => self.refuse(Some(json!(id)), Some(json!(nsid)), error),
            Msg::Input(data) => self.push(json!({ "type": "input", "data": data })),
            Msg::InputsEnd(failure) => {
                self.inputs_open = false;
                if let Some(f) = failure {
                    self.error.get_or_insert(format!("inputs failed: {f}"));
                }
            }
            Msg::Deadline => {
                let message = format!("Module exceeded {} ms", js_num(self.budget.wall_ms));
                self.error.get_or_insert(message);
            }
            Msg::Abort => {
                self.error.get_or_insert("aborted".into());
            }
            _ => {}
        }
    }
}

#[allow(clippy::too_many_arguments)]
async fn subscription(
    id: String,
    generation: u64,
    planned: Planned,
    request: Request,
    call: Arc<dyn Call>,
    signal: Signal,
    abort: Signal,
    mut ctl: mpsc::UnboundedReceiver<Ctl>,
    tx: mpsc::UnboundedSender<Msg>,
) {
    // A Close while still opening (cancel, the run's end) stops the open now, unless a Send came
    // first (it goes out once open, then the Close, as for an open stream).
    let mut early = None;
    let opening = call.subscribe(planned.nsid.clone(), request);
    tokio::pin!(opening);
    let opened = loop {
        tokio::select! {
            sub = &mut opening => break sub,
            reason = signal.aborted() => break Err(reason),
            command = ctl.recv(), if early.is_none() => match command {
                Some(Ctl::Send(data)) => early = Some(data),
                Some(Ctl::Close(reason)) => return abort.abort(reason),
                None => break opening.await,
            },
        }
    };
    let mut sub = match opened {
        Ok(sub) => sub,
        Err(error) => {
            let _ = tx.send(Msg::Closed(id, generation, Some(error)));
            return;
        }
    };
    let mut failure = None;
    let mut ctl_open = true;
    loop {
        tokio::select! {
            message = sub.messages.recv() => match message {
                Some(Ok(data)) => { let _ = tx.send(Msg::Message(id.clone(), generation, data)); }
                Some(Err(error)) => { failure = Some(error); break; }
                None => break,
            },
            command = async { match early.take() { Some(data) => Some(Ctl::Send(data)), None => ctl.recv().await } }, if ctl_open => match command {
                Some(Ctl::Send(data)) => {
                    let (ack, acked) = oneshot::channel();
                    let _ = sub.commands.send(SubCmd::Send(data, ack));
                    let sent = tokio::select! { a = acked => a.unwrap_or(Err(format!("{} is closed", planned.nsid))), r = signal.aborted() => Err(r) };
                    match sent {
                        Ok(()) => {}
                        Err(e) if e == NOSEND => { let _ = tx.send(Msg::Refuse(id.clone(), planned.nsid.clone(), format!("send: {} cannot send", planned.nsid))); }
                        Err(e) => { failure = Some(signal.reason().unwrap_or(format!("send failed: {e}"))); abort.abort(format!("send failed: {e}")); break; }
                    }
                }
                Some(Ctl::Close(reason)) => {
                    abort.abort(reason);
                    let _ = sub.commands.send(SubCmd::Close);
                    return;
                }
                None => ctl_open = false,
            },
            reason = signal.aborted() => { failure = Some(reason); break; }
        }
    }
    let _ = sub.commands.send(SubCmd::Close);
    let _ = tx.send(Msg::Closed(id, generation, failure));
}

/// runModule: run a handler until done, nothing is pending, or a budget runs out.
/// Never fails: module and call errors come back as `error` or as events.
pub async fn run_module(
    ctx: &Ctx,
    module: &Module,
    entry: &str,
    input: &Map<String, Value>,
    call: Arc<dyn Call>,
    mut options: Options,
) -> Value {
    for h in &module.hooks {
        if let Some(why) = unserved_hook(h, module.calls.as_ref(), options.provides.as_deref(), Some(&[entry.to_string()])) {
            let name = module.name.as_deref().unwrap_or("module");
            let why = if why == NO_ROUTE { "run it in a composition that routes it".into() } else { why };
            let error = format!("{name} needs hook {} ({}): {why}", h.nsid, h.reason);
            return json!({ "entry": entry, "result": null, "error": error, "events": 0, "calls": [] });
        }
    }
    let hooked;
    let (module, call) = if module.hooks.is_empty() {
        (module, call)
    } else {
        let hooks = module.hooks.iter().map(|h| h.nsid.clone());
        hooked = Module { calls: module.calls.clone().map(|c| c.into_iter().chain(hooks).collect()), ..module.clone() };
        let wrapped = HookCall { inner: call, module: module.clone(), ctx: Ctx { runner: ctx.runner.clone(), ledger: ctx.ledger.clone(), now: ctx.now.clone() }, load: options.load.clone(), provides: options.provides.clone(), budget: options.budget.clone(), http: options.http.clone(), loaded: Default::default() };
        (&hooked, Arc::new(wrapped) as Arc<dyn Call>)
    };
    let mint = options.session.clone().unwrap_or_else(|| call.clone());
    let call: Arc<dyn Call> = Arc::new(HttpCall { inner: call, mint, declared: module.http.clone().unwrap_or_default(), http: options.http.clone() });
    let ctx = &ctx.owned(module);
    let budget = Budget::merged(options.budget.as_ref());
    let (tx, mut rx) = mpsc::unbounded_channel();
    let mut run = Run {
        ctx,
        module,
        call,
        budget,
        tx: tx.clone(),
        ended: Signal::new(),
        pending: HashMap::new(),
        closing: vec![],
        held: Held { limit: budget.held_bytes as usize, ..Default::default() },
        waiting: VecDeque::new(),
        queue: VecDeque::new(),
        dropped: HashMap::new(),
        log: VecDeque::new(),
        log_dropped: 0,
        error: None,
        total: 0.0,
        minute: (std::time::Instant::now(), 0.0),
        generation: 0,
        inputs_open: false,
    };
    for (name, value) in options.held.take().unwrap_or_default() {
        if let Err(e) = run.held.op(&format!("{HELD_NSID}put"), Some(&json!({ "name": name, "value": value }))) {
            return json!({ "entry": entry, "result": null, "error": format!("held: {e}"), "events": 0, "calls": [] });
        }
    }
    let mut helpers = vec![];
    if let Some(mut inputs) = options.inputs.take() {
        run.inputs_open = true;
        let tx = tx.clone();
        helpers.push(tokio::spawn(async move {
            loop {
                match inputs.recv().await {
                    Some(Ok(data)) => {
                        let _ = tx.send(Msg::Input(data));
                    }
                    Some(Err(e)) => break drop(tx.send(Msg::InputsEnd(Some(e)))),
                    None => break drop(tx.send(Msg::InputsEnd(None))),
                }
            }
        }));
    }
    // wallMs unlimited: no deadline (a server runs until its signal aborts).
    if budget.wall_ms.is_finite() {
        let wall = ms_duration(budget.wall_ms);
        let deadline_tx = tx.clone();
        helpers.push(tokio::spawn(async move {
            tokio::time::sleep(wall).await;
            let _ = deadline_tx.send(Msg::Deadline);
        }));
    }
    if let Some(signal) = options.signal.clone() {
        if signal.reason().is_some() {
            run.error = Some("aborted".into());
        }
        let abort_tx = tx.clone();
        helpers.push(tokio::spawn(async move {
            signal.aborted().await;
            let _ = abort_tx.send(Msg::Abort);
        }));
    }
    run.push(json!({ "type": "start" }));
    let mut state = Value::Null;
    let mut result = Value::Null;
    let mut events: u64 = 0;
    // An idle run takes the event that woke it at once; a busy one (a handler running, outputs being
    // applied) queues what arrives meanwhile, with the drop rules, as the JS event loop does.
    let mut woken = false;
    while run.error.is_none() {
        if !woken {
            while let Ok(msg) = rx.try_recv() {
                run.ingest(msg);
            }
        }
        woken = false;
        if run.error.is_some() {
            break;
        }
        let Some(mut event) = run.queue.pop_front() else {
            if run.pending.is_empty() && !run.inputs_open {
                break;
            }
            if let Some(msg) = rx.recv().await {
                run.ingest(msg);
                woken = true;
            }
            continue;
        };
        if event.get("type").and_then(Value::as_str) == Some("message")
            && let Some(n) = event_id(&event).and_then(|id| run.dropped.remove(id))
        {
            event["dropped"] = json!(n);
        }
        events += 1;
        if events as f64 > budget.events {
            run.error = Some(format!("Module exceeded {} events", js_num(budget.events)));
            break;
        }
        let mut handler_input = input.clone();
        handler_input.insert("now".into(), json!(ctx.now()));
        handler_input.insert("state".into(), state.clone());
        handler_input.insert("event".into(), event.clone());
        let runner = ctx.runner.clone();
        let mut running = runner.run(entry.to_string(), Value::Object(handler_input), budget.io_bytes as usize);
        let out = loop {
            tokio::select! {
                out = &mut running => break out,
                Some(msg) = rx.recv() => run.ingest(msg),
            }
        };
        let out = match out {
            Ok(out) => out,
            Err(e) => {
                run.error = Some(e);
                break;
            }
        };
        if let Some(Value::String(e)) = out.get("error") {
            run.error = Some(e.clone());
            break;
        }
        let Value::Object(out) = out else {
            run.error = Some("Module output must be an object".into());
            break;
        };
        if let Some(s) = out.get("state") {
            state = s.clone();
        }
        if let Some(r) = out.get("result") {
            result = r.clone();
        }
        if let Some(on_event) = &options.on_event {
            on_event(&event, &Value::Object(out.clone()));
        }
        // Sends to an open subscription go before this output's cancels (a last frame, then close);
        // sends to a subscription this output opens go after its calls.
        let (early, late): (Vec<Value>, Vec<Value>) =
            list(out.get("send")).into_iter().partition(|f| {
                f.get("id")
                    .and_then(Value::as_str)
                    .and_then(|id| run.pending.get(id))
                    .is_some_and(|s| s.kind == Kind::Sub)
            });
        for frame in &early {
            run.send(frame.get("id"), frame.get("data"));
        }
        for id in list(out.get("cancel")) {
            run.cancel(&id);
        }
        for timer in list(out.get("timers")) {
            if let Some(e) = run.set_timer(&timer) {
                run.error.get_or_insert(e);
            }
        }
        if run.error.is_some() {
            break;
        }
        for raw in list(out.get("calls")) {
            run.plan(&raw).await;
        }
        for frame in &late {
            run.send(frame.get("id"), frame.get("data"));
        }
        run.pump();
        if out.get("done") == Some(&Value::Bool(true)) {
            break;
        }
    }
    // Frames the last handler sent (a final response) go out before the streams close, unless aborted.
    let reason = run.error.clone().unwrap_or_else(|| "run ended".into());
    let aborted = options
        .signal
        .as_ref()
        .is_some_and(|s| s.reason().is_some());
    let mut tasks = std::mem::take(&mut run.closing);
    for slot in run.pending.values_mut() {
        if let Some(timer) = &slot.timer {
            timer.abort();
        }
        if let Some(ctl) = &slot.ctl {
            let _ = ctl.send(Ctl::Close(reason.clone()));
        }
        if let Some(task) = slot.task.take() {
            tasks.push(task);
        }
    }
    if !tasks.is_empty() && !aborted {
        let _ = tokio::time::timeout(
            Duration::from_millis(DRAIN_MS),
            futures_util::future::join_all(tasks.iter_mut()),
        )
        .await;
    }
    // Every call's signal follows `ended`: in-flight calls see the run's error (or "run ended").
    run.ended.abort(reason);
    for helper in helpers {
        helper.abort();
    }
    // The result may name held values ({ "$held": name }): they go in now, before the run's are dropped.
    let result = match run.held.splice(&result) {
        Ok(r) => r,
        Err(e) => {
            run.error.get_or_insert(format!("result: {e}"));
            result
        }
    };
    let mut ran = json!({ "entry": entry, "result": result });
    if let Some(e) = run.error {
        ran["error"] = json!(e);
    }
    ran["events"] = json!(events);
    ran["calls"] = Value::Array(run.log.into());
    if run.log_dropped > 0 {
        ran["callsDropped"] = json!(run.log_dropped);
    }
    ran
}

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

    #[test]
    fn huge_or_odd_ms_clamp_instead_of_panicking() {
        assert_eq!(ms_duration(1500.0), Duration::from_millis(1500));
        assert_eq!(ms_duration(1e300), Duration::from_millis(MAX_TIMER_MS as u64));
        assert_eq!(ms_duration(f64::INFINITY), Duration::from_millis(MAX_TIMER_MS as u64));
        assert_eq!(ms_duration(-5.0), Duration::ZERO);
        assert_eq!(ms_duration(f64::NAN), Duration::ZERO);
    }

    #[test]
    fn manifest_lists_are_absent_or_strings() {
        let ok = Module::from_manifest(vec![], &json!({ "calls": ["a.b.c"] })).unwrap();
        assert_eq!(ok.calls, Some(vec!["a.b.c".to_string()]));
        assert_eq!(ok.any_records, None);
        for bad in [json!({ "calls": "a.b.c" }), json!({ "calls": null }), json!({ "anyRecords": [1] }), json!({ "anyRecords": "x" }), json!({ "http": "*" })] {
            assert!(Module::from_manifest(vec![], &bad).err().unwrap().ends_with("must be a list of strings"), "{bad}");
        }
    }

    #[test]
    fn http_allowlists_are_origins() {
        struct NoFetch;
        impl crate::host::Fetch for NoFetch {
            fn fetch(&self, _: &str, _: &str, _: Vec<(String, String)>, _: Option<Vec<u8>>) -> BoxFut<'_, Result<crate::host::Response, String>> {
                Box::pin(async { Err("no network".into()) })
            }
        }
        let http = Http::from_json(Arc::new(NoFetch), &json!({ "https://API.example:443/x": { "headers": { "x-api-key": "k" } }, "http://127.0.0.1:8080": {} })).unwrap();
        assert_eq!(http.allow, vec![("https://api.example".to_string(), vec![("x-api-key".to_string(), "k".to_string())], None), ("http://127.0.0.1:8080".to_string(), vec![], None)]);
        let minted = Http::from_json(Arc::new(NoFetch), &json!({ "https://cocore.dev": { "serviceAuth": { "aud": "did:web:cocore.dev", "lxm": "dev.cocore.inference.dispatch" } } })).unwrap();
        assert_eq!(minted.allow[0].2, Some(("did:web:cocore.dev".to_string(), "dev.cocore.inference.dispatch".to_string())));
        for bad in [json!([]), json!({ "ftp://x.example": {} }), json!({ "https://x.example": { "headers": { "a": 1 } } }), json!({ "https://x.example": { "serviceAuth": { "aud": "x", "lxm": "a.b" } } }), json!({ "https://x.example": { "serviceAuth": {} } })] {
            assert!(Http::from_json(Arc::new(NoFetch), &bad).is_err(), "{bad}");
        }
        assert_eq!(http_origin("https://u:p@x.example/"), None);
        // a fetcher's scheme: "<scheme>://*", never http(s)
        assert_eq!(Http::from_json(Arc::new(NoFetch), &json!({ "iroh://*": {} })).unwrap().allow[0].0, "iroh://*");
        assert!(Http::from_json(Arc::new(NoFetch), &json!({ "https://*": {} })).is_err());
        assert_eq!((allow_key("webrtc:///e30/shell/mcp").as_deref(), allow_key("webrtc://u:p@x/").as_deref()), (Some("webrtc://*"), None));
    }

    // Hooks with no composition (JS withHooks): a default runs nested at its implements entry,
    // a hook without one answers null, a required one fails at start; never sent to the caller's call.
    #[tokio::test]
    async fn a_hook_default_runs_nested_and_a_required_hook_fails_at_start() {
        const H: &str = "com.example.hooks.check";
        struct Steps(fn(&Value) -> Value);
        impl Runner for Steps {
            fn run(&self, _entry: String, input: Value, _io_bytes: usize) -> BoxFut<'_, Result<Value, String>> {
                let out = (self.0)(&input);
                Box::pin(async move { Ok(out) })
            }
        }
        struct Refuse;
        impl Call for Refuse {
            fn call(&self, nsid: String, _r: Request) -> BoxFut<'_, Result<Option<Value>, String>> {
                Box::pin(async move { Err(format!("{nsid} reached the caller's call")) })
            }
            fn subscribe(&self, nsid: String, _r: Request) -> BoxFut<'_, Result<Sub, String>> {
                Box::pin(async move { Err(format!("{nsid} reached the caller's call")) })
            }
        }
        fn asker(input: &Value) -> Value {
            match input["event"]["type"].as_str() {
                Some("start") => json!({ "calls": [{ "id": "q", "nsid": H, "input": { "n": input["n"] } }] }),
                _ => json!({ "result": input["event"]["output"], "done": true }),
            }
        }
        fn even(input: &Value) -> Value {
            let allow = input["n"].as_i64().is_some_and(|n| n % 2 == 0);
            json!({ "result": { "allow": allow, "violations": if allow { json!([]) } else { json!([{ "msg": "odd", "policyId": "even" }]) } }, "done": true })
        }
        // peek: a default that reads a record, on the call its loader gave it
        fn peek(input: &Value) -> Value {
            match input["event"]["type"].as_str() {
                Some("start") => json!({ "calls": [{ "id": "g", "nsid": "com.atproto.repo.getRecord" }] }),
                _ => json!({ "result": input["event"]["output"], "done": true }),
            }
        }
        struct Own;
        impl Call for Own {
            fn call(&self, nsid: String, _r: Request) -> BoxFut<'_, Result<Option<Value>, String>> {
                Box::pin(async move { Ok(Some(json!({ "own": nsid }))) })
            }
            fn subscribe(&self, nsid: String, _r: Request) -> BoxFut<'_, Result<Sub, String>> {
                Box::pin(async move { Err(format!("{nsid}: no")) })
            }
        }
        let ctx = Ctx { runner: Arc::new(Steps(asker)), ledger: None, now: Some("t".into()) };
        let module = |hook: Value| Module::from_manifest(vec![], &json!({ "name": "asker", "calls": [], "hooks": [hook] })).unwrap();
        let load: Loader = Arc::new(|key: String| {
            Box::pin(async move {
                let (spec, runner): (Value, Arc<dyn Runner>) = match key.as_str() {
                    "even" => (json!({ "calls": [], "implements": [{ "nsid": H, "entry": "even" }] }), Arc::new(Steps(even))),
                    "peek" => (json!({ "calls": ["com.atproto.repo.getRecord"], "implements": [{ "nsid": H, "entry": "peek" }] }), Arc::new(Steps(peek))),
                    // no calls: it would get every call there is, so it is refused
                    _ => (json!({ "implements": [{ "nsid": H, "entry": "even" }] }), Arc::new(Steps(even))),
                };
                Ok((Module::from_manifest(vec![1], &spec)?, runner, Arc::new(Own) as Arc<dyn Call>))
            })
        });
        let run = |hook: Value, n: i64| {
            let (ctx, load) = (&ctx, load.clone());
            async move {
                let mut input = Map::new();
                input.insert("n".into(), json!(n));
                run_module(ctx, &module(hook), "caller", &input, Arc::new(Refuse), Options { load: Some(load), ..Default::default() }).await
            }
        };
        let hook = json!({ "nsid": H, "combine": "all-allow", "default": "even", "reason": "r" });
        assert_eq!(run(hook.clone(), 2).await["result"], json!({ "allow": true, "violations": [] }));
        assert_eq!(run(hook, 3).await["result"], json!({ "allow": false, "violations": [{ "msg": "odd", "policyId": "even" }] }));
        assert_eq!(run(json!({ "nsid": H, "reason": "r" }), 2).await["result"], Value::Null);
        let refused = run(json!({ "nsid": H, "required": true, "reason": "r" }), 2).await;
        assert_eq!(refused["error"], json!(format!("asker needs hook {H} (r): run it in a composition that routes it")));
        let other = run(json!({ "nsid": H, "required": true, "entries": ["launch"], "reason": "r" }), 2).await;
        assert_eq!(other["result"], Value::Null, "a hook for other entries binds nothing here");
        // the default's calls go to its own call (the loader's), never the caller's (Refuse)
        assert_eq!(run(json!({ "nsid": H, "default": "peek", "reason": "r" }), 2).await["result"], json!({ "own": "com.atproto.repo.getRecord" }));
        let unlisted = run(json!({ "nsid": H, "combine": "all-allow", "default": "open", "reason": "r" }), 2).await;
        assert_eq!(unlisted["result"], json!({ "allow": false, "violations": [{ "msg": "hook default open: its module lists no calls (a hook default must)", "policyId": "open" }] }));
        // no loader: the same text as JS withHooks
        let mut input = Map::new();
        input.insert("n".into(), json!(2));
        let bare = run_module(&ctx, &module(json!({ "nsid": H, "combine": "all-allow", "default": "even", "reason": "r" })), "caller", &input, Arc::new(Refuse), Options::default()).await;
        assert_eq!(bare["result"]["violations"][0]["msg"], "hook default even: no loader for a module loaded from bytes (pass the host's load option)");
        // a hook needs a string reason, and required is a boolean
        for bad in [json!({ "nsid": H }), json!({ "nsid": H, "reason": 1 }), json!({ "nsid": H, "reason": "r", "required": "yes" })] {
            assert!(Module::from_manifest(vec![], &json!({ "hooks": [bad] })).err().unwrap().starts_with("module hooks must be a list of { nsid, reason, required?: boolean"), "{bad}");
        }
        // only a transport (com.fedfork.exec.<name>.connect) in calls is the host's: an XRPC method there is a hook like any other
        let xrpc = module(json!({ "nsid": "com.example.foo", "required": true, "reason": "r" }));
        assert_eq!(unserved_hook(&xrpc.hooks[0], Some(&vec!["com.example.foo".into()]), None, None).as_deref(), Some(NO_ROUTE));
        let iroh = Hook::parse(&json!({ "nsid": "com.fedfork.exec.iroh.connect", "required": true, "reason": "r" })).unwrap();
        let listed = Some(vec!["com.fedfork.exec.iroh.connect".to_string()]);
        assert_eq!(unserved_hook(&iroh, listed.as_ref(), None, None).as_deref(), Some("this host was built without iroh"), "no provides: provides nothing");
        assert_eq!(unserved_hook(&iroh, listed.as_ref(), Some(&HashSet::from(["com.fedfork.exec.iroh.connect".to_string()])), None), None);
    }
}