//! Module composition, line for line with
//! javascript/src/compose runCompose: instances calling each other over manifest edges, caller
//! handlers, callbacks as {$cap} tokens, policy modules before each call, graph limits.
//! A host here supplies handlers as Rust closures; root input is JSON (no host functions as caps).

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

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

use crate::crypto::hex;
use crate::run::{self, Call, Ctx, Ledger, Module, Options, Request, Sub};
use crate::store::random_bytes;
use crate::wasm::{Bound, Wasm, entry_ok, sha256_hex};
use crate::{BoxFut, Signal, js_num, js_string};

pub const LIMITS: (f64, f64) = (8.0, 100.0);

fn name_ok(s: &str) -> bool {
    (1..=256).contains(&s.len())
        && s.bytes()
            .all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-')
}

pub(crate) fn nsid_ok(s: &str) -> bool {
    run::parse_call(&json!({ "nsid": s }), None).is_ok()
}

fn token_ok(s: &str) -> bool {
    let parts: Vec<&str> = s.split('.').collect();
    parts.len() == 3
        && parts[0] == "cap"
        && parts[1].len() == 32
        && parts[1]
            .bytes()
            .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
        && !parts[2].is_empty()
        && parts[2].bytes().all(|b| b.is_ascii_digit())
}

fn cap_of(v: &Value) -> Option<&str> {
    v.as_object()
        .filter(|m| m.len() == 1)
        .and_then(|m| m.get("$cap"))
        .and_then(Value::as_str)
}

/// Rebuild a JSON value, replacing what `f` returns Some for.
fn walk(
    value: &Value,
    f: &mut dyn FnMut(&Value) -> Result<Option<Value>, String>,
) -> Result<Value, String> {
    if let Some(swapped) = f(value)? {
        return Ok(swapped);
    }
    Ok(match value {
        Value::Array(items) => {
            Value::Array(items.iter().map(|i| walk(i, f)).collect::<Result<_, _>>()?)
        }
        Value::Object(map) => Value::Object(
            map.iter()
                .map(|(k, v)| Ok((k.clone(), walk(v, f)?)))
                .collect::<Result<_, String>>()?,
        ),
        other => other.clone(),
    })
}

/// Check a manifest's shape and references; Err on the first problem.
pub fn validate(manifest: &Value) -> Result<(), String> {
    let fail = |m: String| Err(format!("composition: {m}"));
    let s = |v: Option<&Value>| js_string(v);
    let instances_list = manifest
        .get("instances")
        .and_then(Value::as_array)
        .filter(|l| !l.is_empty());
    let Some(list) = instances_list else {
        return fail("instances must be a non-empty list".into());
    };
    let mut instances = HashSet::new();
    for inst in list {
        let name = inst.get("name").and_then(Value::as_str).unwrap_or_default();
        if !name_ok(name) {
            return fail(format!(
                "instance name {} must match /^[A-Za-z0-9_-]{{1,256}}$/",
                s(inst.get("name"))
            ));
        }
        if !instances.insert(name.to_string()) {
            return fail(format!("instance {name} named twice"));
        }
        if !inst
            .get("module")
            .and_then(Value::as_str)
            .is_some_and(|m| !m.is_empty())
        {
            return fail(format!("instance {name} needs a module"));
        }
        if inst.get("pin").is_some_and(|p| !p.as_str().is_some_and(|p| p.len() == 64 && p.bytes().all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)))) {
            return fail(format!("instance {name} pin must be 64 lowercase hex characters"));
        }
        if let Some(calls) = inst.get("calls")
            && !calls
                .as_array()
                .is_some_and(|c| c.iter().all(|n| n.as_str().is_some_and(nsid_ok)))
        {
            return fail(format!("instance {name} calls must be NSIDs"));
        }
        let budget_ok = match inst.get("budget") {
            None | Some(Value::Null) => true,
            Some(Value::Object(b)) => b.iter().all(|(k, v)| run::BUDGET_KEYS.contains(&k.as_str()) && run::budget_value(Some(v), None).is_some()),
            Some(_) => false,
        };
        if !budget_ok {
            return fail(format!(
                "instance {name} budget must map {} to positive numbers",
                run::BUDGET_KEYS.join(", ")
            ));
        }
    }
    let mut handlers = HashSet::new();
    for h in manifest
        .get("handlers")
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
    {
        let name = h.get("name").and_then(Value::as_str).unwrap_or_default();
        if !name_ok(name) || !handlers.insert(name.to_string()) {
            return fail(format!(
                "handler name {} must be unique and match /^[A-Za-z0-9_-]{{1,256}}$/",
                s(h.get("name"))
            ));
        }
    }
    let policies = |list: Option<&Value>, at: &str| -> Result<(), String> {
        for p in list.and_then(Value::as_array).into_iter().flatten() {
            if !p
                .get("instance")
                .and_then(Value::as_str)
                .is_some_and(|i| instances.contains(i))
            {
                return Err(format!(
                    "composition: {at}: policy instance {} is not an instance",
                    s(p.get("instance"))
                ));
            }
            if p.get("entry")
                .is_some_and(|e| !e.as_str().is_some_and(entry_ok))
            {
                return Err(format!(
                    "composition: {at}: policy entry {} is not an export name",
                    s(p.get("entry"))
                ));
            }
        }
        Ok(())
    };
    for inst in list {
        policies(
            inst.get("policies"),
            &format!("instance {}", s(inst.get("name"))),
        )?;
    }
    let limits_ok = match manifest.get("limits") {
        None | Some(Value::Null) => true,
        Some(Value::Object(l)) => l.iter().all(|(k, v)| ["depth", "calls"].contains(&k.as_str()) && run::budget_value(Some(v), None).is_some()),
        Some(_) => false,
    };
    if !limits_ok {
        return fail("limits must map depth, calls to positive numbers".into());
    }
    for edge in manifest
        .get("edges")
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
    {
        let at = format!("edge {} {}", s(edge.get("from")), s(edge.get("nsid")));
        let str_of = |k: &str| edge.get(k).and_then(Value::as_str);
        if !str_of("from").is_some_and(|f| instances.contains(f)) {
            return fail(format!("{at}: from is not an instance"));
        }
        if !str_of("nsid").is_some_and(|n| nsid_ok(n) && !n.starts_with("cap.")) {
            return fail(format!("{at}: nsid must be an NSID (not cap.*)"));
        }
        if !str_of("reason").is_some_and(|r| !r.trim().is_empty()) {
            return fail(format!("{at}: reason is required"));
        }
        if edge.get("to").is_none() == edge.get("handler").is_none() {
            return fail(format!("{at}: needs exactly one of to and handler"));
        }
        if edge.get("entry").is_some() {
            return fail(format!("{at}: an edge has no entry: it reaches the entry its target's module implements for {}", s(edge.get("nsid"))));
        }
        if edge.get("to").is_some() && !str_of("to").is_some_and(|t| instances.contains(t)) {
            return fail(format!("{at}: to is not an instance"));
        }
        if let Some(h) = edge.get("handler")
            && !h.as_str().is_some_and(|h| handlers.contains(h))
        {
            return fail(format!(
                "{at}: handler {} is not declared in handlers",
                s(Some(h))
            ));
        }
        policies(edge.get("policies"), &at)?;
    }
    crate::params::validate(manifest)
}

/// The entry an edge reaches: what its target's module implements for the edge's NSID (none for a handler).
pub fn entry_of(modules: &HashMap<String, Module>, edge: &Value) -> Option<String> {
    let module = modules.get(edge.get("to")?.as_str()?)?;
    module.implements.iter().find(|(n, _)| Some(n.as_str()) == edge.get("nsid").and_then(Value::as_str)).map(|(_, e)| e.clone())
}

#[derive(Clone)]
pub struct Composition {
    pub manifest: Value,
    /// by instance name
    pub modules: HashMap<String, Module>,
}

/// An instance's `module` ref and `pin` -> its module (JS loadComposition's `load`): how this host finds modules
/// (at:// records, at the pin's version when pinned; staged or vendored keys; local builds). Hook defaults come
/// through it too, as sibling refs.
pub type ModuleLoader = Arc<dyn Fn(String, Option<String>) -> BoxFut<'static, Result<Module, String>> + Send + Sync>;

/// JS loadComposition, less its provides check (check, next): validate the manifest, load each instance's
/// module with `load` (once per ref, checked against the instance's `pin`: run::module_pin), then the hooks:
/// edges checked, defaults added and loaded the same way.
pub async fn load(manifest: &Value, load: &ModuleLoader) -> Result<Composition, String> {
    validate(manifest)?;
    let mut cache = HashMap::new();
    let mut modules = HashMap::new();
    for inst in manifest["instances"].as_array().into_iter().flatten() {
        modules.insert(js_string(inst.get("name")), load_instance(inst, load, &mut cache).await?);
    }
    let manifest = hook_routes(manifest, &mut modules, load, &mut cache).await?;
    validate(&manifest)?;
    Ok(Composition { manifest, modules })
}

/// The load checks with no wasm (JS checkComposition): `load` gives each ref's module from its module.json
/// fields (spec_module), with its pin when the loader knows it (a lockfile's; else pins are the running
/// host's to check). Transports count as provided. Returns the manifest as it would run (defaults added).
pub async fn check_specs(manifest: &Value, load_spec: &ModuleLoader) -> Result<Value, String> {
    let comp = load(manifest, load_spec).await?;
    let provides = comp.modules.values().flat_map(|m| m.hooks.iter().map(|h| h.nsid.clone())).collect();
    check(&comp, &provides)?;
    Ok(comp.manifest)
}

/// A module from its module.json fields alone (no wasm) for check_specs, with its pin if known.
pub fn spec_module(spec: &Value, pin: Option<&str>) -> Result<Module, String> {
    Ok(Module { pin: pin.unwrap_or_default().to_string(), ..Module::from_manifest(vec![], spec)? })
}

/// A loaded composition's required hooks are served on this host (run::unserved_hook): routed, or a transport
/// in `provides` (the NSIDs this host's transports serve). A hook naming entries binds only the entries edges
/// and policies reach; a root's entry is checked when it runs (run_compose).
pub fn check(comp: &Composition, provides: &HashSet<String>) -> Result<(), String> {
    let edges: Vec<Value> = comp.manifest.get("edges").and_then(Value::as_array).cloned().unwrap_or_default();
    let instances: Vec<Value> = comp.manifest["instances"].as_array().cloned().unwrap_or_default();
    let mut reached: HashMap<String, Vec<String>> = HashMap::new();
    for e in &edges {
        if let Some(to) = e.get("to").and_then(Value::as_str) {
            reached.entry(to.into()).or_default().push(entry_of(&comp.modules, e).unwrap_or_default());
        }
    }
    for p in instances.iter().chain(edges.iter()).flat_map(policies_of) {
        let entry = p.get("entry").and_then(Value::as_str).unwrap_or("evaluate").to_string();
        reached.entry(js_string(p.get("instance"))).or_default().push(entry);
    }
    for inst in &instances {
        let name = js_string(inst.get("name"));
        let module = &comp.modules[&name];
        for hook in &module.hooks {
            if edges.iter().any(|e| js_string(e.get("from")) == name && js_string(e.get("nsid")) == hook.nsid) {
                continue;
            }
            if let Some(why) = run::unserved_hook(hook, module.calls.as_ref(), Some(provides), reached.get(&name).map(Vec::as_slice)) {
                return Err(format!("composition: instance {name} needs hook {} ({}): {why}", hook.nsid, hook.reason));
            }
        }
    }
    Ok(())
}

async fn load_instance(inst: &Value, load: &ModuleLoader, cache: &mut HashMap<String, Module>) -> Result<Module, String> {
    let (name, reference) = (js_string(inst.get("name")), js_string(inst.get("module")));
    let pin = inst.get("pin").and_then(Value::as_str).map(String::from);
    let key = format!("{reference}@{}", pin.as_deref().unwrap_or_default());
    if !cache.contains_key(&key) {
        cache.insert(key.clone(), load(reference, pin).await?);
    }
    let module = cache[&key].clone();
    // a spec-only module (check_specs) has a pin only when its loader knows one (a lockfile's)
    if inst.get("pin").is_some_and(|pin| !module.pin.is_empty() && *pin != module.pin.as_str()) {
        return Err(format!("composition: instance {name} module does not match its pin"));
    }
    Ok(module)
}

/// Hooks (JS hookRoutes), once the instances' modules are loaded: an edge only for a hook the module
/// declares or an NSID in its calls; one edge per (from, nsid) unless the hook ANDs verdicts (all-allow); an
/// unrouted hook with a default gets an instance (`<from>-<default>`, its module the sibling of the instance's
/// own ref) loaded with `load` and checked in turn. An instance's hooks that default to one module share
/// that one instance, as a composition routes them (vm-mcp's provision and release, to one requester): one
/// session and one did:key (sealed per instance name), so what one hook's call made the other's can see. Returns the manifest with the default instances and edges
/// added (check, after it, sees them routed and the entries they are reached at).
async fn hook_routes(manifest: &Value, modules: &mut HashMap<String, Module>, load: &ModuleLoader, cache: &mut HashMap<String, Module>) -> Result<Value, String> {
    let mut instances: Vec<Value> = manifest["instances"].as_array().cloned().unwrap_or_default();
    let mut edges: Vec<Value> = manifest.get("edges").and_then(Value::as_array).cloned().unwrap_or_default();
    let mut defaults: Vec<Value> = vec![]; // the instances added here
    let mut i = 0;
    while i < instances.len() {
        let inst = instances[i].clone();
        i += 1;
        let name = js_string(inst.get("name"));
        let module = modules[&name].clone();
        crate::params::check_input(&inst, &module, manifest.get("parameters"))?;
        let mine: Vec<Value> = edges.iter().filter(|e| js_string(e.get("from")) == name).cloned().collect();
        for edge in &mine {
            let nsid = js_string(edge.get("nsid"));
            let hook = module.hooks.iter().find(|h| h.nsid == nsid);
            if hook.is_none() && !module.any_hooks && module.calls.as_ref().is_some_and(|c| !c.contains(&nsid)) {
                return Err(format!("composition: edge {name} {nsid}: {} neither declares hook {nsid} nor lists it in calls", js_string(inst.get("module"))));
            }
            if !hook.is_some_and(|h| h.all_allow) && mine.iter().filter(|e| js_string(e.get("nsid")) == nsid).count() > 1 {
                return Err(format!("composition: edge {name} {nsid}: routed twice"));
            }
            if let Some(to) = edge.get("to").and_then(Value::as_str)
                && entry_of(modules, edge).is_none()
            {
                let module = instances.iter().find(|x| x["name"] == to).map(|x| js_string(x.get("module"))).unwrap_or_default();
                return Err(format!("composition: edge {name} {nsid}: {to} ({module}) does not implement {nsid}"));
            }
        }
        for hook in &module.hooks {
            if mine.iter().any(|e| js_string(e.get("nsid")) == hook.nsid) {
                continue;
            }
            let Some(default) = &hook.default else { continue };
            let def_name: String = format!("{name}-{default}").chars().take(256).collect();
            let def = json!({ "name": def_name, "module": sibling(&js_string(inst.get("module")), default) });
            let shared = defaults.contains(&def);
            if !shared && instances.iter().any(|x| js_string(x.get("name")) == def_name) {
                return Err(format!("composition: instance {def_name} (the default for {name}'s hook {}) named twice", hook.nsid));
            }
            if inst.get("pin").is_some() {
                return Err(format!("composition: instance {name} is pinned, but its hook {} default {default} is not: route the hook to a pinned instance", hook.nsid));
            }
            let target = if shared { modules[&def_name].clone() } else { load_instance(&def, load, cache).await? };
            if target.calls.is_none() {
                return Err(format!("composition: instance {name} hook {}: its default {default} lists no calls (a hook default must)", hook.nsid));
            }
            if !target.implements.iter().any(|(n, _)| *n == hook.nsid) {
                return Err(format!("composition: instance {name} hook {}: its default {default} does not implement it", hook.nsid));
            }
            if !shared {
                modules.insert(def_name.clone(), target);
                defaults.push(def.clone());
                instances.push(def);
            }
            let reason = if hook.reason.is_empty() { format!("the default for {}", hook.nsid) } else { hook.reason.clone() };
            edges.push(json!({ "from": name, "nsid": hook.nsid, "to": def_name, "reason": reason }));
        }
    }
    let mut out = manifest.clone();
    out["instances"] = Value::Array(instances);
    out["edges"] = Value::Array(edges);
    Ok(out)
}

/// `key` beside `reference` (JS sibling): the same publisher's record for an at://did/collection/rkey
/// instance, else the key itself.
fn sibling(reference: &str, key: &str) -> String {
    let parts: Vec<&str> = reference.strip_prefix("at://").map(|r| r.split('/').collect()).unwrap_or_default();
    match parts[..] {
        [did, collection, rkey] if !did.is_empty() && !collection.is_empty() && !rkey.is_empty() => format!("at://{did}/{collection}/{key}"),
        _ => key.to_string(),
    }
}

/// kcp market/requester/policy.go PolicyWorkflowName: a computer.socialweb.temp.policy.ghalite record's name.
const POLICY_ALIASES: [(&str, &str); 6] = [("", "requester-tangled-vouch"), ("only-me", "requester-only-me"), ("only_me", "requester-only-me"), ("mutuals", "requester-mutuals"), ("tangled-vouch", "requester-tangled-vouch"), ("tangled_vouch", "requester-tangled-vouch")];
const POLICIES: [&str; 9] = ["requester-tangled-vouch", "bidder-tangled-vouch", "requester-only-me", "bidder-only-me", "requester-mutuals", "bidder-mutuals", "open", "bid-payload", "under-4-cpus"];
const POLICY_NSID: &str = "com.publicdomainrelay.temp.market.evaluatePolicy";

/// JS rfpPolicyRoutes: add the policies an RFP attaches to `from`'s policy hook (POLICY_NSID,
/// combine all-allow), after the routes it has: a loaded manifest's (load routes the module's default), so
/// the requester's or bidder's own policy still runs, first. Each computer.socialweb.temp.policy.ghalite
/// record's `name` (kcp-canonicalised) is the module policy-<name> (sibling of `from`'s module), its `args`
/// the instance's input.args. A hook not routed yet, a record that is not an object or a name no policy
/// module implements fails (closed). Returns the manifest with one instance and edge per record added.
pub fn rfp_policy_routes(manifest: &Value, from: &str, records: &[Value]) -> Result<Value, String> {
    let inst = manifest["instances"].as_array().into_iter().flatten().find(|i| i["name"] == from).ok_or_else(|| format!("composition: no instance {from}"))?;
    let (mut instances, mut edges) = (manifest["instances"].as_array().cloned().unwrap_or_default(), manifest.get("edges").and_then(Value::as_array).cloned().unwrap_or_default());
    if !edges.iter().any(|e| e["from"] == from && e["nsid"] == POLICY_NSID) {
        return Err(format!("composition: {from}'s {POLICY_NSID} is not routed: RFP policies add to its routes, so pass the loaded manifest (loading routes its default)"));
    }
    for (i, record) in records.iter().enumerate() {
        if !record.is_object() {
            return Err(format!("composition: no policy module for the RFP's policy {record}"));
        }
        let raw = record["name"].as_str().map(|n| n.trim().to_lowercase()).unwrap_or_default();
        let name = POLICY_ALIASES.iter().find(|(a, _)| *a == raw).map_or(raw.as_str(), |(_, n)| n);
        if !POLICIES.contains(&name) {
            return Err(format!("composition: no policy module for the RFP's policy {}", record.get("name").map_or("undefined".into(), Value::to_string)));
        }
        let mut instance = json!({ "name": format!("{from}-rfp-{i}"), "module": sibling(&js_string(inst.get("module")), &format!("policy-{name}")) });
        if let Some(args) = record.get("args") {
            instance["input"] = json!({ "args": args });
        }
        edges.push(json!({ "from": from, "nsid": POLICY_NSID, "to": instance["name"], "reason": format!("the RFP attaches policy {name}") }));
        instances.push(instance);
    }
    let mut out = manifest.clone();
    out["instances"] = Value::Array(instances);
    out["edges"] = Value::Array(edges);
    Ok(out)
}

/// What a handler is told about the call; `invoke` calls a cap token found in its input.
pub struct Info {
    pub caller: String,
    pub nsid: String,
    pub reason: String,
    pub signal: Signal,
    shared: Arc<Shared>,
}

impl Info {
    pub fn invoke(&self, token: &str, data: Value) -> BoxFut<'static, Result<Value, String>> {
        let (shared, token, signal) = (self.shared.clone(), token.to_string(), self.signal.clone());
        Box::pin(async move { shared.invoke(&token, data, signal).await })
    }
}

pub type Handler = Arc<dyn Fn(Value, Info) -> BoxFut<'static, Result<Value, String>> + Send + Sync>;
pub type CallFor = Arc<dyn Fn(&str) -> Option<Arc<dyn Call>> + Send + Sync>;
type Invoke = Arc<dyn Fn(Value, Signal) -> BoxFut<'static, Result<Value, String>> + Send + Sync>;

struct Cap {
    invoke: Invoke,
    revoked: Signal,
}

struct Shared {
    comp: Arc<Composition>,
    instances: HashMap<String, Value>,
    /// "from nsid" -> its routes, in manifest order (several only for an all-allow hook)
    edges: HashMap<String, Vec<Value>>,
    wasm: Arc<Wasm>,
    ledger: Option<Arc<dyn Ledger>>,
    call_for: CallFor,
    handlers: HashMap<String, Handler>,
    depth: f64,
    max_calls: f64,
    budget: Option<Value>,
    /// The host's HTTP allowlist (run::Http), for every instance.
    http: Option<Arc<run::Http>>,
    /// Held values the root's run starts with (run::Options held).
    held: Option<Map<String, Value>>,
    /// The root's outputs as they come (run::Options on_event): a workload's output, step by step.
    on_event: Option<run::OnEvent>,
    /// JS compositionOwner: the ledger owner every instance's run shares
    owner: String,
    caps: Mutex<HashMap<String, Cap>>,
    // the last `graph_max` entries, and how many were dropped (a long serve run stays bounded)
    graph: Mutex<(VecDeque<Value>, u64)>,
    graph_max: f64,
    calls: Mutex<f64>,
    serial: Mutex<u64>,
}

type Held = Arc<Mutex<HashSet<String>>>;
/// Tokens to revoke when a call settles (replies minted while it runs join them).
type Minted = Arc<Mutex<Vec<String>>>;

impl Shared {
    fn spend(&self) -> Result<(), String> {
        let mut calls = self.calls.lock().unwrap();
        *calls += 1.0;
        if *calls > self.max_calls {
            Err(format!(
                "composition exceeded {} calls",
                js_num(self.max_calls)
            ))
        } else {
            Ok(())
        }
    }

    fn log(&self, entry: Value) {
        let mut graph = self.graph.lock().unwrap();
        graph.0.push_back(entry);
        if graph.0.len() as f64 > self.graph_max {
            graph.0.pop_front();
            graph.1 += 1;
        }
    }

    // Tokens in a value delivered to a run become callable by that run.
    fn grant(&self, held: &Held, value: &Value) {
        let caps = self.caps.lock().unwrap();
        let _ = walk(value, &mut |v| {
            if let Some(t) = cap_of(v).filter(|t| caps.contains_key(*t)) {
                held.lock().unwrap().insert(t.to_string());
                return Ok(Some(v.clone()));
            }
            Ok(None)
        });
    }

    fn mint(&self, invoke: Invoke) -> String {
        let n = {
            let mut serial = self.serial.lock().unwrap();
            *serial += 1;
            *serial
        };
        let token = format!("cap.{}.{n}", hex(&random_bytes::<16>()));
        self.caps.lock().unwrap().insert(
            token.clone(),
            Cap {
                invoke,
                revoked: Signal::new(),
            },
        );
        token
    }

    fn revoke(&self, tokens: &[String]) {
        let mut caps = self.caps.lock().unwrap();
        for token in tokens {
            if let Some(cap) = caps.remove(token) {
                cap.revoked.abort("capability revoked");
            }
        }
    }

    async fn invoke(&self, token: &str, value: Value, signal: Signal) -> Result<Value, String> {
        let found = self
            .caps
            .lock()
            .unwrap()
            .get(token)
            .map(|c| (c.invoke.clone(), c.revoked.clone()));
        let Some((invoke, revoked)) = found else {
            return Err("capability revoked".into());
        };
        tokio::select! {
            out = invoke(value, signal.clone()) => out,
            reason = signal.aborted() => Err(reason),
            reason = revoked.aborted() => Err(reason),
        }
    }

    // A run's call input: { $cap: name } exports a callback into the run (emit), { $cap: token } it holds is
    // passed on as is. Returns the input the callee sees and the tokens to revoke when the call settles.
    fn export_caps(
        self: &Arc<Self>,
        value: Option<&Value>,
        held: &Held,
        emit: Option<run::Emit>,
    ) -> Result<(Option<Value>, Minted), String> {
        let minted: Minted = Arc::default();
        let Some(value) = value else {
            return Ok((None, minted));
        };
        let out = walk(value, &mut |v| {
            let Some(name) = cap_of(v) else {
                return Ok(None);
            };
            if token_ok(name) {
                if !held.lock().unwrap().contains(name) {
                    return Err(format!("{name} is not a capability this run holds"));
                }
                return Ok(Some(v.clone()));
            }
            let Some(emit) = emit.clone() else {
                return Err("callbacks need a run to call back into".into());
            };
            let (shared, held, minted_for_replies, name) = (
                Arc::downgrade(self),
                held.clone(),
                minted.clone(),
                name.to_string(),
            );
            let token = self.mint(Arc::new(move |data: Value, _signal: Signal| {
                let (shared, held, minted, emit, name) = (
                    shared.clone(),
                    held.clone(),
                    minted_for_replies.clone(),
                    emit.clone(),
                    name.clone(),
                );
                Box::pin(async move {
                    let shared = shared.upgrade().ok_or("capability revoked")?;
                    let (answered, answer) = oneshot::channel::<Value>();
                    let answered = Arc::new(Mutex::new(Some(answered)));
                    let reply_slot: Arc<Mutex<Option<String>>> = Arc::default();
                    let (weak, slot) = (Arc::downgrade(&shared), reply_slot.clone());
                    // one answer per invocation
                    let reply = shared.mint(Arc::new(move |result: Value, _| {
                        let (weak, slot, answered) = (weak.clone(), slot.clone(), answered.clone());
                        Box::pin(async move {
                            if let (Some(shared), Some(token)) =
                                (weak.upgrade(), slot.lock().unwrap().clone())
                            {
                                shared.caps.lock().unwrap().remove(&token);
                            }
                            if let Some(tx) = answered.lock().unwrap().take() {
                                let _ = tx.send(result);
                            }
                            Ok(Value::Null)
                        })
                    }));
                    *reply_slot.lock().unwrap() = Some(reply.clone());
                    held.lock().unwrap().insert(reply.clone());
                    minted.lock().unwrap().push(reply.clone()); // an unanswered reply dies with the call too
                    shared.grant(&held, &data);
                    emit.emit(json!({ "cap": name, "input": data, "reply": reply }));
                    answer.await.map_err(|_| "capability revoked".to_string())
                })
            }));
            minted.lock().unwrap().push(token.clone());
            Ok(Some(json!({ "$cap": token })))
        })?;
        Ok((Some(out), minted))
    }

    /// grants: whether caps in `input` become the run's (not for policy runs: a policy sees the call,
    /// it must not invoke the caller's callbacks).
    fn run_instance(
        self: Arc<Self>,
        name: String,
        entry: String,
        input: Map<String, Value>,
        stack: Vec<String>,
        signal: Option<Signal>,
        grants: bool,
    ) -> BoxFut<'static, Value> {
        Box::pin(async move {
            let inst = self.instances[&name].clone();
            let mut module = self.comp.modules[&name].clone();
            module.owner = Some(self.owner.clone());
            // the root's run (its stack is itself alone) starts with the host's held values
            let initial = if stack.len() == 1 { self.held.clone() } else { None };
            let on_event = if stack.len() == 1 { self.on_event.clone() } else { None };
            let held: Held = Arc::default();
            if grants {
                self.grant(&held, &Value::Object(input.clone()));
            }
            let call: Arc<dyn Call> = Arc::new(ComposeCall {
                shared: self.clone(),
                name: name.clone(),
                held,
                stack,
            });
            // Grants (and hooks) are enforced here, so run-module gets no allowlist or hooks of its own.
            module.calls = None;
            module.hooks = vec![];
            let budget = capped(inst.get("budget"), self.budget.as_ref());
            let mut merged = inst
                .get("input")
                .and_then(Value::as_object)
                .cloned()
                .unwrap_or_default();
            merged.extend(input);
            let ctx = Ctx {
                runner: Arc::new(Bound {
                    wasm: self.wasm.clone(),
                    bytes: module.bytes.clone(),
                    sha256: module.sha256.clone(),
                }),
                ledger: self.ledger.clone(),
                now: None,
            };
            run::run_module(
                &ctx,
                &module,
                &entry,
                &merged,
                call,
                Options {
                    signal,
                    budget: Some(Value::Object(budget)),
                    http: self.http.clone(),
                    held: initial,
                    on_event,
                    // the instance's own session mints http's serviceAuth: the host's, not a call the module makes
                    session: (self.call_for)(&name),
                    ..Default::default()
                },
            )
            .await
        })
    }

    async fn check_policies(
        self: &Arc<Self>,
        list: Vec<Value>,
        request: &Value,
        stack: &[String],
        signal: &Signal,
    ) -> Result<(), String> {
        for policy in list {
            let instance = js_string(policy.get("instance"));
            let entry = policy
                .get("entry")
                .and_then(Value::as_str)
                .unwrap_or("evaluate")
                .to_string();
            let ran: Result<Value, String> = async {
				self.spend()?;
				if stack.len() as f64 >= self.depth {
					return Err(format!("composition exceeded depth {}", js_num(self.depth)));
				}
				if stack.contains(&instance) {
					return Err(format!("cycle: {} -> {instance}", stack.join(" -> ")));
				}
				let input = json!({ "args": policy.get("args").cloned().unwrap_or_else(|| json!({})), "call": redacted(request) });
				let mut nested = stack.to_vec();
				nested.push(instance.clone());
				let ran = self.clone().run_instance(instance.clone(), entry, input.as_object().cloned().unwrap_or_default(), nested, Some(signal.clone()), false).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)),
				}
			}
			.await;
            let verdict = ran.unwrap_or_else(
                |e| json!({ "allow": false, "violations": [{ "msg": e, "policyId": instance }] }),
            );
            let allowed = verdict.get("allow") == Some(&Value::Bool(true));
            let mut entry = json!({ "from": request.get("from"), "nsid": request.get("nsid"), "policy": instance, "ok": allowed });
            if !allowed {
                entry["violations"] = verdict
                    .get("violations")
                    .filter(|v| !v.is_null())
                    .cloned()
                    .unwrap_or_else(|| json!([]));
            }
            self.log(entry);
            if !allowed {
                let msgs: Vec<String> = verdict
                    .get("violations")
                    .and_then(Value::as_array)
                    .into_iter()
                    .flatten()
                    .filter_map(|v| v.get("msg"))
                    .filter(|m| {
                        !matches!(m, Value::Null | Value::Bool(false))
                            && m.as_str() != Some("")
                            && m.as_f64() != Some(0.0)
                    })
                    .map(|m| js_string(Some(m)))
                    .collect();
                return Err(format!(
                    "policy {instance} denied{}",
                    if msgs.is_empty() {
                        String::new()
                    } else {
                        format!(": {}", msgs.join("; "))
                    }
                ));
            }
        }
        Ok(())
    }
}

/// The `call` run-module gets for a run of one instance.
struct ComposeCall {
    shared: Arc<Shared>,
    name: String,
    held: Held,
    stack: Vec<String>,
}

/// What a policy sees of a call: all of it but `input.stdin` (a secret for the guest).
fn redacted(call: &Value) -> Value {
    let mut call = call.clone();
    if let Some(input) = call.get_mut("input").and_then(Value::as_object_mut) {
        input.remove("stdin");
    }
    call
}

fn policies_of(v: &Value) -> Vec<Value> {
    v.get("policies")
        .and_then(Value::as_array)
        .cloned()
        .unwrap_or_default()
}

impl ComposeCall {
    fn what(&self, nsid: &str, request: &Request) -> Value {
        let mut what = json!({ "from": self.name, "nsid": nsid });
        if let Some(p) = &request.params {
            what["params"] = Value::Object(p.clone());
        }
        if let Some(i) = &request.input {
            what["input"] = i.clone();
        }
        if let Some(p) = &request.proxy {
            what["proxy"] = json!(p);
        }
        if let Some(s) = &request.service {
            what["service"] = json!(s);
        }
        what
    }

    // Not routed: an ordinary call, allowed only by module.json calls and instance calls, after policies.
    async fn direct(&self, nsid: &str, request: &Request) -> Result<Option<Arc<dyn Call>>, String> {
        let shared = &self.shared;
        if shared.edges.contains_key(&format!("{} {nsid}", self.name)) || token_ok(nsid) {
            return Ok(None);
        }
        let inst = &shared.instances[&self.name];
        let allowed = |list: Option<&Vec<String>>| list.is_none_or(|l| l.iter().any(|n| n == nsid));
        let inst_calls: Option<Vec<String>> =
            inst.get("calls").and_then(Value::as_array).map(|c| {
                c.iter()
                    .filter_map(|n| n.as_str().map(String::from))
                    .collect()
            });
        if !allowed(shared.comp.modules[&self.name].calls.as_ref()) || !allowed(inst_calls.as_ref())
        {
            return Err(format!(
                "{} may not call {nsid}: no edge, and not in its calls allowlist",
                self.name
            ));
        }
        shared
            .check_policies(
                policies_of(inst),
                &self.what(nsid, request),
                &self.stack,
                &request.signal,
            )
            .await?;
        let call = (shared.call_for)(&self.name)
            .ok_or_else(|| format!("{} has no session for {nsid}", self.name))?;
        Ok(Some(call))
    }

    async fn cap_call(&self, nsid: &str, mut request: Request) -> Result<Option<Value>, String> {
        let shared = &self.shared;
        let record = json!({ "from": self.name, "nsid": "cap", "cap": nsid.split('.').nth(2) });
        if !self.held.lock().unwrap().contains(nsid)
            || !shared.caps.lock().unwrap().contains_key(nsid)
        {
            let mut entry = record;
            entry["ok"] = json!(false);
            entry["error"] = json!("not held");
            shared.log(entry);
            return Err(format!("{nsid} is not a capability this run holds"));
        }
        shared.spend()?;
        let (input, minted) =
            shared.export_caps(request.input.as_ref(), &self.held, request.emit.take())?;
        let out = shared
            .invoke(nsid, input.unwrap_or(Value::Null), request.signal.clone())
            .await;
        let mut entry = record;
        match &out {
            Ok(output) => {
                shared.grant(&self.held, output);
                entry["ok"] = json!(true);
            }
            Err(e) => {
                entry["ok"] = json!(false);
                entry["error"] = json!(e);
            }
        }
        shared.log(entry);
        shared.revoke(&minted.lock().unwrap());
        out.map(Some)
    }

    // One route of a routed call; an all-allow hook's routes in order, the first deny the answer.
    async fn routed(&self, nsid: &str, request: Request) -> Result<Option<Value>, String> {
        let routes = self.shared.edges[&format!("{} {nsid}", self.name)].clone();
        let all = self.shared.comp.modules[&self.name].hooks.iter().any(|h| h.nsid == nsid && h.all_allow);
        if !all {
            return self.route_one(nsid, routes[0].clone(), request).await;
        }
        for edge in routes {
            let id = js_string(edge.get("to").or_else(|| edge.get("handler")));
            let verdict = match self.route_one(nsid, edge, request.clone()).await {
                Ok(out) => run::verdict_of(out.as_ref(), &id, None),
                Err(e) => run::verdict_of(None, &id, Some(&e)),
            };
            if verdict["allow"] != true {
                return Ok(Some(verdict));
            }
        }
        Ok(Some(json!({ "allow": true, "violations": [] })))
    }

    async fn route_one(&self, nsid: &str, edge: Value, mut request: Request) -> Result<Option<Value>, String> {
        let shared = &self.shared;
        let inst = shared.instances[&self.name].clone();
        let reason = js_string(edge.get("reason"));
        let to = edge.get("to").and_then(Value::as_str).map(String::from);
        let mut route = Map::new();
        match &to {
            Some(to) => {
                route.insert("to".into(), json!(to));
                route.insert("entry".into(), json!(entry_of(&shared.comp.modules, &edge)));
            }
            None => {
                route.insert(
                    "handler".into(),
                    edge.get("handler").cloned().unwrap_or(Value::Null),
                );
            }
        }
        let mut record = json!({ "from": self.name, "nsid": nsid });
        record.as_object_mut().unwrap().extend(route.clone());
        record["reason"] = json!(reason);
        let minted: Mutex<Option<Minted>> = Mutex::default();
        let out: Result<Value, String> = async {
            if request.proxy.is_some() || request.service.is_some() || request.subscribe {
                return Err(format!(
                    "{nsid} is routed in this composition: no proxy, service or subscribe"
                ));
            }
            if let Some(to) = &to {
                if self.stack.len() as f64 >= shared.depth {
                    return Err(format!(
                        "composition exceeded depth {}",
                        js_num(shared.depth)
                    ));
                }
                if self.stack.contains(to) {
                    return Err(format!("cycle: {} -> {to}", self.stack.join(" -> ")));
                }
            }
            let mut what = self.what(nsid, &request);
            what.as_object_mut().unwrap().extend(route.clone());
            what["reason"] = json!(reason);
            shared
                .check_policies(
                    [policies_of(&inst), policies_of(&edge)].concat(),
                    &what,
                    &self.stack,
                    &request.signal,
                )
                .await?;
            shared.spend()?;
            let (input, exported) =
                shared.export_caps(request.input.as_ref(), &self.held, request.emit.take())?;
            *minted.lock().unwrap() = Some(exported);
            match &to {
                None => {
                    let handler = js_string(edge.get("handler"));
                    let info = Info {
                        caller: self.name.clone(),
                        nsid: nsid.into(),
                        reason: reason.clone(),
                        signal: request.signal.clone(),
                        shared: shared.clone(),
                    };
                    let f = shared.handlers.get(&handler).cloned().ok_or_else(|| {
                        format!("composition: handler {handler} was not supplied")
                    })?;
                    f(input.unwrap_or(Value::Null), info).await
                }
                Some(to) => {
                    let mut body = request.params.clone().unwrap_or_default();
                    match input {
                        Some(Value::Object(m)) => body.extend(m),
                        Some(other) => drop(body.insert("input".into(), other)),
                        None => {}
                    }
                    body.insert(
                        "caller".into(),
                        json!({ "instance": self.name, "nsid": nsid, "reason": reason }),
                    );
                    let mut stack = self.stack.clone();
                    stack.push(to.clone());
                    let ran = shared
                        .clone()
                        .run_instance(
                            to.clone(),
                            entry_of(&shared.comp.modules, &edge).unwrap_or_default(),
                            body,
                            stack,
                            Some(request.signal.clone()),
                            true,
                        )
                        .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)),
                    }
                }
            }
        }
        .await;
        match &out {
            Ok(output) => {
                shared.grant(&self.held, output);
                record["ok"] = json!(true);
            }
            Err(e) => {
                record["ok"] = json!(false);
                record["error"] = json!(e);
            }
        }
        shared.log(record);
        // caps passed with the call, and replies minted while it ran, die with it
        if let Some(minted) = minted.lock().unwrap().take() {
            shared.revoke(&minted.lock().unwrap());
        }
        out.map(Some)
    }
}

impl Call for ComposeCall {
    fn call(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Option<Value>, String>> {
        Box::pin(async move {
            if token_ok(&nsid) {
                return self.cap_call(&nsid, request).await;
            }
            // An optional hook nothing serves: skip, never the network. One the module
            // also lists in calls (a transport) is the host's: direct, below.
            let module = &self.shared.comp.modules[&self.name];
            if !self.shared.edges.contains_key(&format!("{} {nsid}", self.name))
                && module.hooks.iter().any(|h| h.nsid == nsid && !run::hosts_hook(h, module.calls.as_ref()))
            {
                self.shared.log(json!({ "from": self.name, "nsid": nsid, "hook": "unrouted", "ok": true }));
                return Ok(Some(Value::Null));
            }
            match self.direct(&nsid, &request).await? {
                Some(call) => {
                    call.call(
                        nsid,
                        Request {
                            emit: None,
                            ..request
                        },
                    )
                    .await
                }
                None => self.routed(&nsid, request).await,
            }
        })
    }

    fn subscribe(&self, nsid: String, request: Request) -> BoxFut<'_, Result<Sub, String>> {
        Box::pin(async move {
            if token_ok(&nsid) {
                return Err(format!("{nsid} cannot be subscribed to"));
            }
            match self.direct(&nsid, &request).await? {
                Some(call) => {
                    call.subscribe(
                        nsid,
                        Request {
                            emit: None,
                            ..request
                        },
                    )
                    .await
                }
                // refused before any route runs: an all-allow hook would answer a verdict, not a stream
                None => Err(format!("{nsid} is routed in this composition: no proxy, service or subscribe")),
            }
        })
    }
}

/// runCompose: run `entry` of instance `root` with everything the composition grants it. Module, call
/// and policy errors come back in the result; Err is a bad root or a declared handler not supplied.
#[allow(clippy::too_many_arguments)]
pub async fn run_compose(
    comp: Arc<Composition>,
    wasm: Arc<Wasm>,
    ledger: Option<Arc<dyn Ledger>>,
    root: &str,
    entry: &str,
    input: Map<String, Value>,
    call_for: CallFor,
    handlers: HashMap<String, Handler>,
    limits: Option<&Value>,
    options: Options,
) -> Result<Value, String> {
    crate::params::check_bound(&comp.manifest)?;
    let instances: HashMap<String, Value> = comp.manifest["instances"]
        .as_array()
        .into_iter()
        .flatten()
        .map(|i| (js_string(i.get("name")), i.clone()))
        .collect();
    if !instances.contains_key(root) {
        return Err(format!("composition: no instance {root}"));
    }
    // The root entry's own hooks (load checked the entries edges and policies reach).
    let root_module = &comp.modules[root];
    for h in &root_module.hooks {
        let routed = comp.manifest.get("edges").and_then(Value::as_array).into_iter().flatten().any(|e| js_string(e.get("from")) == root && js_string(e.get("nsid")) == h.nsid);
        if let Some(why) = run::unserved_hook(h, root_module.calls.as_ref(), options.provides.as_deref(), Some(&[entry.to_string()])).filter(|_| !routed) {
            return Err(format!("composition: instance {root} needs hook {} ({}) for {entry}: {why}", h.nsid, h.reason));
        }
    }
    for h in comp
        .manifest
        .get("handlers")
        .and_then(Value::as_array)
        .into_iter()
        .flatten()
    {
        let name = js_string(h.get("name"));
        if !handlers.contains_key(&name) {
            return Err(format!("composition: handler {name} was not supplied"));
        }
    }
    let limit = |key: &str, default: f64| limit(limits, comp.manifest.get("limits"), key, default);
    let mut edges: HashMap<String, Vec<Value>> = HashMap::new();
    for e in comp.manifest.get("edges").and_then(Value::as_array).into_iter().flatten() {
        edges.entry(format!("{} {}", js_string(e.get("from")), js_string(e.get("nsid")))).or_default().push(e.clone());
    }
    let owner = composition_owner(&comp.modules);
    let shared = Arc::new(Shared {
        depth: limit("depth", LIMITS.0),
        max_calls: limit("calls", LIMITS.1),
        comp,
        instances,
        edges,
        wasm,
        ledger,
        call_for,
        handlers,
        graph_max: run::Budget::merged(options.budget.as_ref()).log,
        budget: options.budget,
        http: options.http,
        held: options.held,
        on_event: options.on_event,
        owner,
        caps: Mutex::default(),
        graph: Mutex::default(),
        calls: Mutex::new(0.0),
        serial: Mutex::new(0),
    });
    let mut ran = shared
        .clone()
        .run_instance(
            root.into(),
            entry.into(),
            input,
            vec![root.into()],
            options.signal,
            true,
        )
        .await;
    let (graph, dropped) = shared.graph.lock().unwrap().clone();
    ran["graph"] = Value::Array(graph.into());
    if dropped > 0 {
        ran["graphDropped"] = json!(dropped);
    }
    Ok(ran)
}

/// The host's limits cap the manifest's (JS capped): the smaller; "unlimited" only from the host.
fn limit(host: Option<&Value>, manifest: Option<&Value>, key: &str, default: f64) -> f64 {
    let host = run::budget_value(host.and_then(|l| l.get(key)), Some(key));
    let manifest = run::budget_value(manifest.and_then(|l| l.get(key)), None);
    match (host, manifest) {
        (Some(h), Some(m)) => h.min(m),
        (h, m) => h.or(m).unwrap_or(default),
    }
}

/// JS capped: the host's budget caps the manifest's; a manifest may lower a value, never raise it.
/// Only positive finite numbers count for a manifest, and "unlimited" too for a host's UNLIMITED keys;
/// with no host value the manifest's own applies.
fn capped(manifest: Option<&Value>, host: Option<&Value>) -> Map<String, Value> {
    let mut out = manifest
        .and_then(Value::as_object)
        .cloned()
        .unwrap_or_default();
    for (key, value) in host.and_then(Value::as_object).into_iter().flatten() {
        let Some(h) = run::budget_value(Some(value), Some(key)) else {
            continue;
        };
        let min = match run::budget_value(out.get(key), None) {
            Some(m) if m < h => out[key].clone(),
            _ => value.clone(),
        };
        out.insert(key.clone(), min);
    }
    out
}

/// JS compositionOwner: one ledger owner for all of a composition's instances (they share records:
/// agents deletes the vm its requester made), from its modules' own owners.
pub fn composition_owner(modules: &HashMap<String, Module>) -> String {
    let mut ids: Vec<String> = modules.values().map(Module::owner).collect();
    ids.sort();
    ids.dedup();
    format!("composition:{}", sha256_hex(ids.join("\n").as_bytes()))
}

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

    #[test]
    fn rfp_policy_records_route_by_their_kcp_names() {
        let module = "at://did:plc:a/com.fedfork.exec.wasm.module";
        let raw = json!({ "instances": [{ "name": "requester", "module": format!("{module}/requester") }] });
        // a loaded manifest: the requester's default policy routes the hook, and keeps routing it
        let mut manifest = raw.clone();
        manifest["instances"].as_array_mut().unwrap().push(json!({ "name": "requester-own", "module": format!("{module}/policy-requester-tangled-vouch") }));
        manifest["edges"] = json!([{ "from": "requester", "nsid": POLICY_NSID, "to": "requester-own", "reason": "its default" }]);
        let routed = rfp_policy_routes(&manifest, "requester", &[json!({ "name": " Only-Me " }), json!({ "args": { "n": 1 } }), json!({ "name": "open" })]).unwrap();
        assert_eq!(routed["edges"][0]["to"], "requester-own", "the default stays, first");
        assert_eq!(routed["instances"][2], json!({ "name": "requester-rfp-0", "module": format!("{module}/policy-requester-only-me") }));
        assert_eq!(routed["instances"][3]["input"], json!({ "args": { "n": 1 } }));
        assert_eq!(routed["instances"][3]["module"], format!("{module}/policy-requester-tangled-vouch"));
        assert_eq!(routed["edges"][3], json!({ "from": "requester", "nsid": POLICY_NSID, "to": "requester-rfp-2", "reason": "the RFP attaches policy open" }));
        assert_eq!(rfp_policy_routes(&manifest, "requester", &[json!({ "name": "mine" })]).unwrap_err(), r#"composition: no policy module for the RFP's policy "mine""#);
        // fail closed: a record that is not an object, a hook nothing routes yet (it would drop the default)
        assert_eq!(rfp_policy_routes(&manifest, "requester", &[Value::Null]).unwrap_err(), "composition: no policy module for the RFP's policy null");
        assert_eq!(rfp_policy_routes(&raw, "requester", &[json!({ "name": "open" })]).unwrap_err(), format!("composition: requester's {POLICY_NSID} is not routed: RFP policies add to its routes, so pass the loaded manifest (loading routes its default)"));
        assert_eq!(rfp_policy_routes(&manifest, "nope", &[]).unwrap_err(), "composition: no instance nope");
        assert_eq!(sibling("requester", "policy-open"), "policy-open");
        assert_eq!(sibling("at://did:plc:a/x", "k"), "k");
    }

    #[test]
    fn host_budgets_cap_the_manifests() {
        let out = capped(
            Some(&json!({ "wallMs": 1e9, "calls": 5, "events": 10 })),
            Some(&json!({ "wallMs": 1000, "calls": 50, "queue": 7 })),
        );
        assert_eq!(
            Value::Object(out),
            json!({ "wallMs": 1000, "calls": 5, "events": 10, "queue": 7 })
        );
        assert_eq!(capped(None, None), Map::new());
        // only positive finite numbers count, from either side
        let out = capped(
            Some(&json!({ "calls": 5 })),
            Some(&json!({ "calls": "lots", "wallMs": -1, "events": 0, "queue": 9 })),
        );
        assert_eq!(Value::Object(out), json!({ "calls": 5, "queue": 9 }));
        let merged = run::Budget::merged(Some(&json!({ "calls": "lots", "wallMs": -5, "events": 7 })));
        assert_eq!((merged.calls, merged.wall_ms, merged.events), (100.0, 60_000.0, 7.0));
        // "unlimited" is the host's alone, for calls, callsPerMinute, events and wallMs: a manifest under it
        // still lowers it, and cannot say it
        let out = capped(Some(&json!({ "calls": 5 })), Some(&json!({ "calls": "unlimited", "wallMs": "unlimited", "inFlight": "unlimited" })));
        assert_eq!(Value::Object(out), json!({ "calls": 5, "wallMs": "unlimited" }));
        let merged = run::Budget::merged(Some(&json!({ "calls": "unlimited", "callsPerMinute": 30, "events": "unlimited", "wallMs": "unlimited", "inFlight": "unlimited" })));
        assert_eq!((merged.calls, merged.calls_per_minute, merged.events, merged.wall_ms, merged.in_flight), (f64::INFINITY, 30.0, f64::INFINITY, f64::INFINITY, 6.0));
        assert_eq!(run::BUDGET.calls_per_minute, f64::INFINITY);
        let unlimited = json!({ "calls": "unlimited", "depth": "unlimited" });
        assert_eq!(limit(Some(&unlimited), None, "calls", LIMITS.1), f64::INFINITY);
        assert_eq!(limit(Some(&unlimited), None, "depth", LIMITS.0), 8.0, "depth is never unlimited");
        assert_eq!(limit(Some(&unlimited), Some(&json!({ "calls": 3 })), "calls", LIMITS.1), 3.0);
        assert_eq!(limit(Some(&json!({ "calls": 1 })), Some(&json!({ "calls": 50 })), "calls", LIMITS.1), 1.0);
        assert_eq!(limit(None, Some(&json!({ "calls": "unlimited" })), "calls", LIMITS.1), 100.0);
        let manifest = json!({ "instances": [{ "name": "a", "module": "m" }], "limits": { "calls": "unlimited" } });
        assert!(validate(&manifest).unwrap_err().contains("limits must map"));
        for budget in [json!({ "calls": "1" }), json!({ "calls": 0 }), json!({ "wallMs": "unlimited" }), json!({ "nope": 1 }), json!([1])] {
            let manifest = json!({ "instances": [{ "name": "a", "module": "m", "budget": budget }] });
            assert!(validate(&manifest).unwrap_err().contains("budget must map"), "{budget}");
        }
    }

    #[test]
    fn manifest_limits_are_positive_numbers_and_policies_never_see_stdin() {
        for limits in [json!({ "calls": "x" }), json!({ "calls": null }), json!({ "depth": 0 }), json!({ "nope": 1 }), json!([1])] {
            let manifest = json!({ "instances": [{ "name": "a", "module": "m" }], "limits": limits });
            assert_eq!(validate(&manifest).unwrap_err(), "composition: limits must map depth, calls to positive numbers", "{limits}");
        }
        assert!(validate(&json!({ "instances": [{ "name": "a", "module": "m" }], "limits": { "depth": 2, "calls": 10 } })).is_ok());
        let call = json!({ "from": "a", "nsid": "x.y.z", "input": { "stdin": "tok", "command": "id" } });
        assert_eq!(redacted(&call), json!({ "from": "a", "nsid": "x.y.z", "input": { "command": "id" } }));
    }

    #[test]
    fn composition_owner_matches_the_js_host() {
        let module = |id: Option<&str>, bytes: &[u8]| Module {
            id: id.map(String::from),
            bytes: Arc::new(bytes.to_vec()),
            ..Default::default()
        };
        let uri = Some("at://did:plc:a/com.fedfork.exec.wasm.module/r");
        let modules = HashMap::from([
            ("x".to_string(), module(uri, b"")),
            ("y".to_string(), module(None, &[1])),
            ("z".to_string(), module(uri, b"")),
        ]);
        // javascript/src/compose compositionOwner of the same modules
        assert_eq!(
            composition_owner(&modules),
            "composition:664c3446fd903186ae49de5fa1349a6f8b1b6a56a6d270af67cb796729a8db5a"
        );
    }

    // An instance's input is its module's config (JS checkInput, javascript/src/compose/compose.test.mjs)
    #[tokio::test]
    async fn instance_input_is_the_modules_config() {
        let with = |config: Value| {
            let m = Module::from_manifest(vec![], &json!({ "calls": [], "config": config })).unwrap();
            let loader: ModuleLoader = Arc::new(move |_, _| {
                let m = m.clone();
                Box::pin(async move { Ok(m) })
            });
            loader
        };
        let n = with(json!([{ "key": "n", "label": "n", "type": "integer" }]));
        let one = |input: Value| json!({ "parameters": [{ "key": "n", "type": "integer" }, { "key": "s" }], "instances": [{ "name": "a", "module": "compose", "input": input }] });
        assert!(load(&one(json!({ "n": { "$param": "n" } })), &n).await.is_ok());
        assert!(load(&one(json!({ "n": 3 })), &n).await.is_ok());
        for (input, err) in [
            (json!({ "m": 1 }), "input m is not in compose's config (n)"),
            (json!({ "n": "3" }), "input n must be of type integer"),
            (json!({ "n": { "$param": "s" } }), "input n is integer, but parameter s is string"),
            (json!([1]), "input must be an object"),
        ] {
            assert_eq!(load(&one(input), &n).await.err().unwrap(), format!("composition: instance a: {err}"));
        }
        let none = Module::from_manifest(vec![], &json!({ "calls": [] })).unwrap();
        let none: ModuleLoader = Arc::new(move |_, _| {
            let m = none.clone();
            Box::pin(async move { Ok(m) })
        });
        assert_eq!(load(&one(json!({ "n": 1 })), &none).await.err().unwrap(), "composition: instance a: input n is not in compose's config");
        let required = with(json!([{ "key": "n", "label": "n", "required": true }]));
        let bare = json!({ "instances": [{ "name": "a", "module": "compose" }] });
        assert_eq!(load(&bare, &required).await.err().unwrap(), "composition: instance a: input n is required by compose's config");
        assert!(Module::from_manifest(vec![], &json!({ "config": [{ "key": "n", "type": "date" }] })).err().unwrap().starts_with("module config must be a list of { key, label, type?"));
    }

    // Hooks (JS hookRoutes, javascript/src/compose/hooks.test.mjs)
    #[tokio::test]
    async fn hooks_check_edges_defaults_entries_and_transports() {
        const H: &str = "com.example.hooks.check";
        const IROH: &str = "com.fedfork.exec.iroh.connect";
        let module = |spec: Value| Module::from_manifest(vec![], &spec).unwrap();
        let asker = |hook: Value| module(json!({ "calls": [IROH], "hooks": [hook] }));
        let policy = module(json!({ "calls": [], "implements": [{ "nsid": H, "entry": "even" }] }));
        // modules by ref; every ref asked for is noted
        let asked = Arc::new(Mutex::new(vec![]));
        let run = |manifest: Value, modules: &[(&str, &Module)], provides: &HashSet<String>| {
            let modules: HashMap<String, Module> = modules.iter().map(|(n, m)| (n.to_string(), (*m).clone())).collect();
            let asked = asked.clone();
            let loader: ModuleLoader = Arc::new(move |r: String, _| {
                asked.lock().unwrap().push(r.clone());
                let found = modules.get(&r).cloned().ok_or(format!("no module {r}"));
                Box::pin(async move { found })
            });
            let provides = provides.clone();
            async move {
                let comp = load(&manifest, &loader).await?;
                check(&comp, &provides).map(|_| comp.manifest)
            }
        };
        let none = HashSet::new();
        let only = |name: &str| json!({ "instances": [{ "name": name, "module": "asker" }] });
        let required = asker(json!({ "nsid": H, "required": true, "reason": "r" }));
        assert_eq!(run(only("a"), &[("asker", &required)], &none).await.unwrap_err(), format!("composition: instance a needs hook {H} (r): no edge routes it"));
        // entries: a root's entry is checked when it runs, not at load
        let scoped = asker(json!({ "nsid": H, "required": true, "entries": ["launch"], "reason": "r" }));
        assert!(run(only("a"), &[("asker", &scoped)], &none).await.is_ok());
        let reached = json!({ "instances": [{ "name": "a", "module": "both" }, { "name": "b", "module": "asker" }],
            "edges": [{ "from": "a", "nsid": "x.y.z", "to": "b", "reason": "r" }] });
        let both = asker(json!({ "nsid": "x.y.z", "reason": "r" }));
        let launches = module(json!({ "calls": [IROH], "hooks": [{ "nsid": H, "required": true, "entries": ["launch"], "reason": "r" }], "implements": [{ "nsid": "x.y.z", "entry": "launch" }] }));
        assert!(run(reached.clone(), &[("both", &both), ("asker", &launches)], &none).await.unwrap_err().contains("instance b needs hook"));
        // an edge reaches what its target implements: one that does not is refused
        assert_eq!(run(reached.clone(), &[("both", &both), ("asker", &scoped)], &none).await.unwrap_err(), "composition: edge a x.y.z: b (asker) does not implement x.y.z");
        let entry = json!({ "instances": [{ "name": "a", "module": "both" }, { "name": "b", "module": "asker" }], "edges": [{ "from": "a", "nsid": "x.y.z", "to": "b", "entry": "launch", "reason": "r" }] });
        assert_eq!(run(entry, &[], &none).await.unwrap_err(), "composition: edge a x.y.z: an edge has no entry: it reaches the entry its target's module implements for x.y.z");
        // an edge the module does not declare; twice without all-allow
        let edge = |to: &str| json!({ "from": "a", "nsid": H, "to": to, "reason": "r" });
        let two = json!({ "instances": [{ "name": "a", "module": "asker" }, { "name": "p", "module": "policy" }, { "name": "q", "module": "policy" }], "edges": [edge("p"), edge("q")] });
        let optional = asker(json!({ "nsid": H, "reason": "r" }));
        assert!(run(two.clone(), &[("asker", &optional), ("policy", &policy)], &none).await.unwrap_err().ends_with("routed twice"));
        let all = asker(json!({ "nsid": H, "combine": "all-allow", "reason": "r" }));
        assert!(run(two.clone(), &[("asker", &all), ("policy", &policy)], &none).await.is_ok());
        let undeclared = asker(json!({ "nsid": "x.y.other", "reason": "r" }));
        assert!(run(two, &[("asker", &undeclared), ("policy", &policy)], &none).await.unwrap_err().contains("asker neither declares hook"));
        // a default: an instance of the sibling module, and an edge to its implements entry
        let with_default = asker(json!({ "nsid": H, "default": "policy", "reason": "r" }));
        let manifest = run(only("a"), &[("asker", &with_default), ("policy", &policy)], &none).await.unwrap();
        assert_eq!(manifest["instances"][1], json!({ "name": "a-policy", "module": "policy" }));
        assert_eq!(manifest["edges"][0], json!({ "from": "a", "nsid": H, "to": "a-policy", "reason": "r" }));
        // two hooks defaulting to one module share one instance of it, each at its own implements entry
        let two_hooks = module(json!({ "calls": [], "hooks": [{ "nsid": H, "default": "policy", "reason": "r" }, { "nsid": "com.example.hooks.second", "default": "policy", "reason": "r2" }] }));
        let serves_both = module(json!({ "calls": [], "implements": [{ "nsid": H, "entry": "even" }, { "nsid": "com.example.hooks.second", "entry": "lookup" }] }));
        let manifest = run(only("a"), &[("asker", &two_hooks), ("policy", &serves_both)], &none).await.unwrap();
        assert_eq!(manifest["instances"], json!([{ "name": "a", "module": "asker" }, { "name": "a-policy", "module": "policy" }]));
        let routes: Vec<_> = manifest["edges"].as_array().unwrap().iter().map(|e| (e["nsid"].clone(), e["to"].clone(), entry_of(&HashMap::from([("a-policy".to_string(), serves_both.clone())]), e))).collect();
        assert_eq!(routes, [(json!(H), json!("a-policy"), Some("even".into())), (json!("com.example.hooks.second"), json!("a-policy"), Some("lookup".into()))]);
        // an at:// module's default is its own publisher's record, whoever publishes the composition (review 4b #1)
        let bob = "at://did:plc:bobtest/com.fedfork.exec.wasm.module";
        asked.lock().unwrap().clear();
        let remote = json!({ "instances": [{ "name": "a", "module": format!("{bob}/asker") }] });
        let manifest = run(remote, &[(&format!("{bob}/asker"), &with_default), (&format!("{bob}/policy"), &policy)], &none).await.unwrap();
        assert_eq!(manifest["instances"][1]["module"], format!("{bob}/policy"));
        assert_eq!(*asked.lock().unwrap(), [format!("{bob}/asker"), format!("{bob}/policy")]);
        // an instance the user named like the default is not taken as it (review 4b #2)
        let squat = json!({ "instances": [{ "name": "a", "module": "asker" }, { "name": "a-policy", "module": "other" }] });
        assert_eq!(run(squat, &[("asker", &with_default), ("other", &policy)], &none).await.unwrap_err(), format!("composition: instance a-policy (the default for a's hook {H}) named twice"));
        // a default's own required hooks, for the entry it is reached at, are checked (review 4b #3)
        let pol = module(json!({ "calls": [], "implements": [{ "nsid": H, "entry": "even" }], "hooks": [{ "nsid": "com.example.hooks.two", "required": true, "entries": ["even"], "reason": "r2" }] }));
        assert_eq!(run(only("a"), &[("asker", &with_default), ("policy", &pol)], &none).await.unwrap_err(), "composition: instance a-policy needs hook com.example.hooks.two (r2): no edge routes it");
        // a default that lists no calls; a pinned instance's default (review 4a H1); pins
        let unlisted = module(json!({ "implements": [{ "nsid": H, "entry": "even" }] }));
        assert_eq!(run(only("a"), &[("asker", &with_default), ("policy", &unlisted)], &none).await.unwrap_err(), format!("composition: instance a hook {H}: its default policy lists no calls (a hook default must)"));
        let pinned = json!({ "instances": [{ "name": "a", "module": "asker", "pin": with_default.pin }] });
        assert_eq!(run(pinned, &[("asker", &with_default), ("policy", &policy)], &none).await.unwrap_err(), format!("composition: instance a is pinned, but its hook {H} default policy is not: route the hook to a pinned instance"));
        let wrong = json!({ "instances": [{ "name": "a", "module": "asker", "pin": policy.pin }] });
        assert_eq!(run(wrong, &[("asker", &optional)], &none).await.unwrap_err(), "composition: instance a module does not match its pin");
        assert_eq!(validate(&json!({ "instances": [{ "name": "a", "module": "asker", "pin": "deadbeef" }] })).unwrap_err(), "composition: instance a pin must be 64 lowercase hex characters");
        // a transport: in calls, required, served when provides has it (no provides: none)
        let transport = asker(json!({ "nsid": IROH, "required": true, "reason": "dials" }));
        assert_eq!(run(only("a"), &[("asker", &transport)], &none).await.unwrap_err(), format!("composition: instance a needs hook {IROH} (dials): this host was built without iroh"));
        assert!(run(only("a"), &[("asker", &transport)], &HashSet::from([IROH.to_string()])).await.is_ok());
        // an XRPC method in calls is not a transport: a required hook on it needs a route
        let xrpc = module(json!({ "calls": ["com.example.foo"], "hooks": [{ "nsid": "com.example.foo", "required": true, "reason": "r" }] }));
        assert!(run(only("a"), &[("asker", &xrpc)], &HashSet::from(["com.example.foo".to_string()])).await.unwrap_err().ends_with("no edge routes it"));
    }

    #[test]
    fn verdicts_fail_closed() {
        assert_eq!(run::verdict_of(Some(&json!({ "allow": true, "violations": [{ "msg": "x" }] })), "p", None), json!({ "allow": true, "violations": [] }));
        assert_eq!(run::verdict_of(Some(&json!("yes")), "p", None), json!({ "allow": false, "violations": [{ "msg": "no verdict", "policyId": "p" }] }));
        assert_eq!(run::verdict_of(None, "p", Some("boom")), json!({ "allow": false, "violations": [{ "msg": "boom", "policyId": "p" }] }));
        let own = json!({ "allow": false, "violations": [{ "msg": "no", "policyId": "vouch" }] });
        assert_eq!(run::verdict_of(Some(&own), "p", None), own);
    }
}