// Module composition: run wasm module instances that call each
// other, caller-supplied handlers and callbacks, over the edges a com.fedfork.exec.composition manifest
// grants, with policy modules checked before each call. Every run is runModule's.
// No dependencies; browser, Node and Workers.

import { checkBound, checkInput, validateParameters } from "./params.js";
import { BUDGET, budgetValue, checkLists, hostsHook, loadModule, localModule, moduleOwner, modulePin, runModule, unservedHook, verdictOf } from "../run-module/index.js";

const LIMITS = Object.freeze({ depth: 8, calls: 100 });
const NAME = /^[A-Za-z0-9_-]{1,256}$/;
const PIN = /^[0-9a-f]{64}$/;
const ENTRY = /^[A-Za-z_][A-Za-z0-9_]{0,63}$/;
const NSID = /^[a-zA-Z][a-zA-Z0-9-]*(\.[a-zA-Z0-9-]+){2,}$/;
const TOKEN = /^cap\.[0-9a-f]{32}\.\d+$/;
const messageOf = (error) => (error instanceof Error ? error.message : String(error));
const hex = (bytes) => [...new Uint8Array(bytes)].map((b) => b.toString(16).padStart(2, "0")).join("");
const random = () => hex(crypto.getRandomValues(new Uint8Array(16)));
/** The entry an edge reaches: what its target's module implements for the edge's NSID (none for a handler). */
const entryOf = (modules, edge) => modules[edge.to]?.implements?.find((i) => i.nsid === edge.nsid)?.entry;
const isCap = (value) => value !== null && typeof value === "object" && !Array.isArray(value) && Object.keys(value).length === 1 && typeof value.$cap === "string";

const withoutStdin = (input) => {
	if (input === null || typeof input !== "object" || Array.isArray(input) || !("stdin" in input)) return input;
	const { stdin, ...rest } = input;
	return rest;
};

// Rebuild a JSON value, replacing what `fn` returns non-undefined for (caps, functions).
function walk(value, fn) {
	const swapped = fn(value);
	if (swapped !== undefined) return swapped;
	if (Array.isArray(value)) return value.map((item) => walk(item, fn));
	if (value !== null && typeof value === "object") return Object.fromEntries(Object.entries(value).map(([key, item]) => [key, walk(item, fn)]));
	return value;
}

// The host's budget or limits cap the manifest's: a manifest may lower a value, never raise it. Only
// positive finite numbers count for a manifest (its other values are dropped), and Infinity too for a
// host's UNLIMITED keys (limits.calls is one); with no host value the manifest's own applies.
const capped = (manifest, host) => {
	const out = Object.fromEntries(Object.entries(manifest ?? {}).filter(([, value]) => budgetValue(value)));
	for (const [key, value] of Object.entries(host ?? {})) if (budgetValue(value, key)) out[key] = budgetValue(out[key]) && out[key] < value ? out[key] : value;
	return out;
};

/**
 * The ledger owner all of a composition's instances share (they share records: agents deletes the vm
 * its requester made): from the modules' own owners (moduleOwner), so another composition of other
 * modules cannot claim them.
 */
async function compositionOwner(modules) {
	const ids = [...new Set(await Promise.all(Object.values(modules).map(moduleOwner)))].sort();
	return `composition:${hex(await crypto.subtle.digest("SHA-256", new TextEncoder().encode(ids.join("\n"))))}`;
}

/** Check a manifest's shape and references; throws on the first problem. */
export function validateComposition(manifest) {
	const fail = (message) => {
		throw new Error(`composition: ${message}`);
	};
	const instances = new Set();
	const handlers = new Set();
	if (!Array.isArray(manifest?.instances) || !manifest.instances.length) fail("instances must be a non-empty list");
	for (const inst of manifest.instances) {
		if (!NAME.test(inst?.name ?? "")) fail(`instance name ${inst?.name} must match ${NAME}`);
		if (instances.has(inst.name)) fail(`instance ${inst.name} named twice`);
		if (typeof inst.module !== "string" || !inst.module) fail(`instance ${inst.name} needs a module`);
		if (inst.pin !== undefined && !PIN.test(inst.pin)) fail(`instance ${inst.name} pin must be 64 lowercase hex characters`);
		if (inst.calls !== undefined && !(Array.isArray(inst.calls) && inst.calls.every((n) => NSID.test(n)))) fail(`instance ${inst.name} calls must be NSIDs`);
		const budget = inst.budget ?? {};
		if (typeof budget !== "object" || Array.isArray(budget) || !Object.entries(budget).every(([key, value]) => Object.hasOwn(BUDGET, key) && budgetValue(value))) {
			fail(`instance ${inst.name} budget must map ${Object.keys(BUDGET).join(", ")} to positive numbers`);
		}
		instances.add(inst.name);
	}
	const limits = manifest.limits ?? {};
	if (typeof limits !== "object" || Array.isArray(limits) || !Object.entries(limits).every(([key, value]) => Object.hasOwn(LIMITS, key) && budgetValue(value))) {
		fail(`limits must map ${Object.keys(LIMITS).join(", ")} to positive numbers`);
	}
	for (const handler of manifest.handlers ?? []) {
		if (!NAME.test(handler?.name ?? "") || handlers.has(handler.name)) fail(`handler name ${handler?.name} must be unique and match ${NAME}`);
		handlers.add(handler.name);
	}
	const policies = (list, where) => {
		for (const p of list ?? []) {
			if (!instances.has(p?.instance)) fail(`${where}: policy instance ${p?.instance} is not an instance`);
			if (p.entry !== undefined && !ENTRY.test(p.entry)) fail(`${where}: policy entry ${p.entry} is not an export name`);
		}
	};
	for (const inst of manifest.instances) policies(inst.policies, `instance ${inst.name}`);
	for (const edge of manifest.edges ?? []) {
		const where = `edge ${edge?.from} ${edge?.nsid}`;
		if (!instances.has(edge?.from)) fail(`${where}: from is not an instance`);
		if (!NSID.test(edge.nsid ?? "") || edge.nsid.startsWith("cap.")) fail(`${where}: nsid must be an NSID (not cap.*)`);
		if (typeof edge.reason !== "string" || !edge.reason.trim()) fail(`${where}: reason is required`);
		if ((edge.to === undefined) === (edge.handler === undefined)) fail(`${where}: needs exactly one of to and handler`);
		if ("entry" in edge) fail(`${where}: an edge has no entry: it reaches the entry its target's module implements for ${edge.nsid}`);
		if (edge.to !== undefined && !instances.has(edge.to)) fail(`${where}: to is not an instance`);
		if (edge.handler !== undefined && !handlers.has(edge.handler)) fail(`${where}: handler ${edge.handler} is not declared in handlers`);
		policies(edge.policies, where);
	}
	validateParameters(manifest, fail);
	return manifest;
}

export { bindParameters, exportOf } from "./params.js";

/**
 * Validate a manifest and load each instance's module (once per distinct module).
 * @param {{ http: { fetch: typeof fetch } }} ctx  used by the default loader (loadModule)
 * @param {object} manifest  a com.fedfork.exec.composition record or local file
 * @param {{ load?: (module: string, instance: object) => Promise<{ module: Uint8Array, calls?: string[], anyRecords?: string[] }>, provides?: Iterable<string> }} [options]
 *   load: resolves an instance's `module` (staged() for a staged directory, also a hook default's: the sibling of its
 *   instance's ref); default: loadModule (<handle>/<name>, <did>/<name>, at://; a pinned instance's at its pin's version). An instance's `pin` is checked against modulePin;
 *   provides: the host pseudo-call NSIDs (transports) this host serves (absent: none): a required transport hook it lacks fails the load
 */
export async function loadComposition(ctx, manifest, { load = (ref, inst) => loadModule(ctx, ref, inst.pin), provides } = {}) {
	validateComposition(manifest);
	const cache = new Map();
	const modules = {};
	const loadOne = async (inst) => {
		const key = `${inst.module}@${inst.pin ?? ""}`;
		if (!cache.has(key)) cache.set(key, Promise.resolve(load(inst.module, inst)).then(checkLists));
		const mod = await cache.get(key);
		if (inst.pin !== undefined && (await modulePin(mod)) !== inst.pin) throw new Error(`composition: instance ${inst.name} module does not match its pin`);
		return mod;
	};
	for (const inst of manifest.instances) modules[inst.name] = await loadOne(inst);
	manifest = validateComposition(await hookRoutes(manifest, modules, loadOne, provides));
	return { manifest, modules, owner: await compositionOwner(modules), provides };
}

/**
 * loadComposition's checks with no wasm (Rust compose::check_specs; `atlibs compose check`): `specOf(ref, instance)`
 * gives a module's record fields (generated bindings, a vendored modules/<key>.json), with `pin` when it is known
 * (a lockfile's; else pins are the running host's to check). Transports count as provided.
 * @returns {Promise<object>}  the manifest as it would run (hook defaults added)
 */
export async function checkComposition(manifest, specOf) {
	validateComposition(manifest);
	const provides = []; // every hook of a module loaded so far: transports are the running host's to check
	const modules = {};
	const loadOne = async (inst) => {
		const { pin, ...spec } = checkLists(await specOf(inst.module, inst));
		if (inst.pin !== undefined && pin !== undefined && pin !== inst.pin) throw new Error(`composition: instance ${inst.name} module does not match its pin`);
		provides.push(...(spec.hooks ?? []).map((h) => h.nsid));
		return spec;
	};
	for (const inst of manifest.instances) modules[inst.name] = await loadOne(inst);
	return validateComposition(await hookRoutes(manifest, modules, loadOne, provides));
}

// The default's ref: the same publisher's record for an AT-URI or <handle|did>/<name> module, else the same loader's key.
const sibling = (ref, key) => {
	const at = /^(at:\/\/[^/]+\/[^/]+\/)[^/]+$/.exec(ref) ?? /^((?:did:|[^./@\s][^/@\s]*\.)[^/@\s]*\/)[^/@\s]+(?:@[A-Za-z0-9]+)?$/.exec(ref);
	return at ? at[1] + key : key;
};

// Hooks, checked once the modules are known: 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 of it (the sibling of the instance's own ref; it must list its
// calls, and a pinned instance's default is refused: nothing pins it; instances added this way are checked
// too). An instance's hooks that default to one module share one instance of it, <from>-<default>, 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. An unrouted required hook fails the load unless it is a transport this host provides
// (unservedHook). Returns the manifest with the default instances and edges added.
async function hookRoutes(manifest, modules, loadOne, provides) {
	const fail = (message) => {
		throw new Error(`composition: ${message}`);
	};
	const instances = [...manifest.instances];
	const edges = [...(manifest.edges ?? [])];
	const defaults = new Set(); // the instances added here
	// The entries each instance is reached at (edges, policies); one reached by nothing is a root, whose
	// entry runCompose checks (a hook naming entries binds only those).
	const reached = new Map();
	const reach = (name, entry) => reached.set(name, [...(reached.get(name) ?? []), entry]);
	for (const edge of edges) if (edge.to) reach(edge.to, entryOf(modules, edge));
	for (const p of [...instances.flatMap((x) => x.policies ?? []), ...edges.flatMap((e) => e.policies ?? [])]) reach(p.instance, p.entry ?? "evaluate");
	for (let i = 0; i < instances.length; i++) {
		const inst = instances[i];
		const mod = modules[inst.name];
		checkInput(inst, mod, manifest.parameters, fail);
		const hooks = mod.hooks ?? [];
		const mine = edges.filter((e) => e.from === inst.name);
		for (const edge of mine) {
			const hook = hooks.find((h) => h.nsid === edge.nsid);
			if (!hook && !mod.anyHooks && mod.calls && !mod.calls.includes(edge.nsid)) fail(`edge ${inst.name} ${edge.nsid}: ${inst.module} neither declares hook ${edge.nsid} nor lists it in calls`);
			if (hook?.combine !== "all-allow" && mine.filter((e) => e.nsid === edge.nsid).length > 1) fail(`edge ${inst.name} ${edge.nsid}: routed twice`);
			if (edge.to && !entryOf(modules, edge)) fail(`edge ${inst.name} ${edge.nsid}: ${edge.to} (${instances.find((x) => x.name === edge.to).module}) does not implement ${edge.nsid}`);
		}
		for (const hook of hooks) {
			if (mine.some((e) => e.nsid === hook.nsid)) continue;
			if (!hook.default) {
				const why = unservedHook(hook, mod.calls, provides, reached.get(inst.name));
				if (why) fail(`instance ${inst.name} needs hook ${hook.nsid} (${hook.reason}): ${why}`);
				continue;
			}
			const def = { name: `${inst.name}-${hook.default}`.slice(0, 256), module: sibling(inst.module, hook.default) };
			const named = instances.find((x) => x.name === def.name);
			const shared = defaults.has(named) && named.module === def.module;
			if (named && !shared) fail(`instance ${def.name} (the default for ${inst.name}'s hook ${hook.nsid}) named twice`);
			if (inst.pin !== undefined) fail(`instance ${inst.name} is pinned, but its hook ${hook.nsid} default ${hook.default} is not: route the hook to a pinned instance`);
			const target = shared ? modules[def.name] : await loadOne(def);
			if (target.calls === undefined) fail(`instance ${inst.name} hook ${hook.nsid}: its default ${hook.default} lists no calls (a hook default must)`);
			const entry = target.implements?.find((x) => x.nsid === hook.nsid)?.entry;
			if (!entry) fail(`instance ${inst.name} hook ${hook.nsid}: its default ${hook.default} does not implement it`);
			reach(def.name, entry);
			if (!shared) {
				modules[def.name] = target;
				defaults.add(def);
				instances.push(def);
			}
			edges.push({ from: inst.name, nsid: hook.nsid, to: def.name, reason: hook.reason || `the default for ${hook.nsid}` });
		}
	}
	return { ...manifest, instances, edges };
}

// kcp market/requester/policy.go PolicyWorkflowName: a computer.socialweb.temp.policy.ghalite record's name.
const POLICY_ALIASES = { "": "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 = ["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 = "com.publicdomainrelay.temp.market.evaluatePolicy";

/**
 * Add the policies an RFP attaches to `from`'s policy hook, after the routes it has: pass a
 * loaded manifest (loadComposition(...).manifest: its default routes the hook), so the requester's or the
 * bidder's own policy still runs, first, and the RFP's are ANDed after it (combine all-allow). Each
 * computer.socialweb.temp.policy.ghalite record's `name` (kcp-canonicalised) is the module policy-<name> (the
 * same publisher's record for an AT-URI instance), its `args` the instance's input.args. Throws (fail closed)
 * for a hook not routed yet, a record that is not an object, or a name no policy module implements.
 * @param {object} manifest  a loaded composition's manifest
 * @param {string} from  the instance whose hook (com.publicdomainrelay.temp.market.evaluatePolicy, combine all-allow) they serve
 * @param {object[]} records  the attached policy records' values
 * @returns {object}  the manifest with one instance and edge per record added after its own; load it again to run it
 */
export function rfpPolicyRoutes(manifest, from, records) {
	const inst = manifest.instances.find((x) => x.name === from);
	if (!inst) throw new Error(`composition: no instance ${from}`);
	if (!(manifest.edges ?? []).some((e) => e.from === from && e.nsid === POLICY_NSID)) {
		throw new Error(`composition: ${from}'s ${POLICY_NSID} is not routed: RFP policies add to its routes, so pass the loaded manifest (loading routes its default)`);
	}
	const added = records.map((record, i) => {
		if (record === null || typeof record !== "object" || Array.isArray(record)) throw new Error(`composition: no policy module for the RFP's policy ${JSON.stringify(record)}`);
		const raw = typeof record.name === "string" ? record.name.trim().toLowerCase() : "";
		const name = POLICY_ALIASES[raw] ?? raw;
		if (!POLICIES.includes(name)) throw new Error(`composition: no policy module for the RFP's policy ${JSON.stringify(record.name)}`);
		const instance = { name: `${from}-rfp-${i}`, module: sibling(inst.module, `policy-${name}`), ...(record.args !== undefined && { input: { args: record.args } }) };
		return [instance, { from, nsid: POLICY_NSID, to: instance.name, reason: `the RFP attaches policy ${name}` }];
	});
	return { ...manifest, instances: [...manifest.instances, ...added.map(([x]) => x)], edges: [...(manifest.edges ?? []), ...added.map(([, e]) => e)] };
}

/**
 * A loader for `atlibs stage` output: modules/<key>.wasm and modules/<key>.json under base. Only the
 * module record's own fields are taken from the json; the id (and so the ledger owner) is where the
 * wasm was loaded from, never something the json names.
 * @param {string | URL} base  where the staged directory is served
 */
export const staged = (base, fetchImpl = fetch) => async (key) => {
	if (!/^[A-Za-z0-9_-]{1,200}$/.test(key)) throw new Error(`composition: a staged module is a key, not ${key}`);
	return localModule(new URL(`modules/${key}.wasm`, base), new URL(`modules/${key}.json`, base), fetchImpl);
};

/**
 * Run `entry` of instance `root` with everything the composition grants it. Never throws for module, call or
 * policy errors (they come back as the run's `error` or as refused calls); throws for a bad manifest or a
 * declared handler that was not supplied.
 * @param {object} ctx  as for runModule (ctx.wasm runs modules, ctx.storage.managed_records the rkey ledger)
 * @param {{ manifest: object, modules: Record<string, object> }} composition  loadComposition()
 * @param {string} root  instance name
 * @param {string} entry
 * @param {Record<string, unknown>} input  JS functions anywhere in it become capabilities the root may call
 * @param {{ callFor?: (instance: object) => Function, handlers?: Record<string, (input: unknown, info: { caller: string, nsid: string, reason: string, signal: AbortSignal }) => unknown>,
 *   limits?: Partial<typeof LIMITS>, signal?: AbortSignal, budget?: object, onEvent?: (event: object, output: object, instance: string) => void,
 *   http?: object, fetchers?: object, held?: object }} [options]  http, fetchers: runModule's, for every instance; held: runModule's, for the root's run
 *   callFor: the XRPC `call` for an instance's own calls (its session; policyCall(stage, instance.module, calls) in a page);
 *   handlers: the caller's handlers, named in the manifest; caps in their input arrive as async functions
 * @returns {Promise<object>}  the root run's result (runModule's shape) plus `graph`: one entry per routed,
 *   handler, capability call and policy run
 */
export async function runCompose(ctx, { manifest, modules, owner, provides }, root, entry, input, { callFor, handlers = {}, limits: overrides, signal, budget, onEvent, http, fetchers, held: initial } = {}) {
	checkBound(manifest);
	const instances = Object.fromEntries(manifest.instances.map((inst) => [inst.name, inst]));
	if (!instances[root]) throw new Error(`composition: no instance ${root}`);
	// The root entry's own hooks (load checked the entries edges and policies reach).
	for (const hook of modules[root].hooks ?? []) {
		const why = !(manifest.edges ?? []).some((e) => e.from === root && e.nsid === hook.nsid) && unservedHook(hook, modules[root].calls, provides, [entry]);
		if (why) throw new Error(`composition: instance ${root} needs hook ${hook.nsid} (${hook.reason}) for ${entry}: ${why}`);
	}
	for (const { name } of manifest.handlers ?? []) if (typeof handlers[name] !== "function") throw new Error(`composition: handler ${name} was not supplied`);
	const limits = { ...LIMITS, ...capped(manifest.limits, overrides) };
	owner ??= await compositionOwner(modules);
	const edges = new Map(); // "from nsid" -> its routes, in manifest order (several only for an all-allow hook)
	for (const edge of manifest.edges ?? []) edges.set(`${edge.from} ${edge.nsid}`, [...(edges.get(`${edge.from} ${edge.nsid}`) ?? []), edge]);
	const caps = new Map(); // token -> { invoke(input, signal), pending: Set<reject> }
	const graph = [];
	let graphDropped = 0;
	const graphMax = budgetValue(budget?.log) ? budget.log : BUDGET.log;
	// The last budget.log entries (a long serve run stays bounded); the result counts the rest.
	const note = (entry) => {
		graph.push(entry);
		if (graph.length > graphMax) {
			graph.shift();
			graphDropped++;
		}
	};
	let calls = 0;
	let serial = 0;
	const spend = () => {
		if (++calls > limits.calls) throw new Error(`composition exceeded ${limits.calls} calls`);
	};

	// Tokens in a value delivered to a run become callable by that run.
	const grant = (held, value) => void walk(value, (v) => (isCap(v) && caps.has(v.$cap) ? (held.add(v.$cap), v) : undefined));
	const mint = (invoke) => {
		const token = `cap.${random()}.${++serial}`;
		caps.set(token, { invoke, pending: new Set() });
		return token;
	};
	const revoke = (tokens) => {
		for (const token of tokens) {
			for (const reject of caps.get(token)?.pending ?? []) reject(new Error("capability revoked"));
			caps.delete(token);
		}
	};
	const invoke = (token, value, signal) => {
		const cap = caps.get(token);
		if (!cap) return Promise.reject(new Error("capability revoked"));
		return new Promise((resolve, reject) => {
			cap.pending.add(reject);
			signal?.addEventListener("abort", () => reject(signal.reason), { once: true });
			Promise.resolve()
				.then(() => cap.invoke(value, signal))
				.then(resolve, reject)
				.finally(() => cap.pending.delete(reject));
		});
	};

	// 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.
	const exportCaps = (value, held, emit) => {
		const minted = [];
		const out = walk(value, (v) => {
			if (!isCap(v)) return undefined;
			if (TOKEN.test(v.$cap)) {
				if (!held.has(v.$cap)) throw new Error(`${v.$cap} is not a capability this run holds`);
				return v;
			}
			const name = v.$cap;
			const token = mint((data) => {
				let answered;
				const answer = new Promise((resolve) => (answered = resolve));
				const reply = mint((result) => (caps.delete(reply), answered(result), null)); // one answer per invocation
				held.add(reply);
				minted.push(reply); // an unanswered reply dies with the call too
				grant(held, data);
				emit({ cap: name, input: data, reply });
				return answer;
			});
			minted.push(token);
			return { $cap: token };
		});
		return { input: out, minted };
	};
	// Caps in a handler's input become async functions.
	const asFunctions = (value, signal) => walk(value, (v) => (isCap(v) && TOKEN.test(v.$cap) ? (data) => invoke(v.$cap, data, signal) : undefined));

	// 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).
	async function runInstance(name, entry, input, stack, signal, grants = true) {
		const inst = instances[name];
		const mod = modules[name];
		const held = new Set();
		if (grants) grant(held, input);
		const call = callOf(name, held, stack);
		// Grants are enforced here, so run-module gets no allowlist of its own.
		const unchecked = { ...mod, calls: undefined, hooks: undefined, owner };
		// http, fetchers: the host's allowlist and scheme fetchers; each instance reaches only what its own module.json http declares.
		// session: the instance's own (callFor, asked only when a token is minted; none: its call), which mints http's
		// serviceAuth tokens: the host's, not a call the module makes
		// held: the root's run (its stack is itself alone) starts with the host's held values
		const options = { signal, budget: capped(inst.budget, budget), onEvent: onEvent && ((event, output) => onEvent(event, output, name)), http, fetchers, session: callFor && ((nsid, request) => (callFor(inst) ?? call)(nsid, request)), ...(stack.length === 1 && initial && { held: initial }) };
		const merged = { ...inst.input, ...input };
		return runModule(ctx, unchecked, entry, merged, call, options);
	}

	async function checkPolicies(list, request, stack, signal) {
		for (const policy of list ?? []) {
			const entry = policy.entry ?? "evaluate";
			let verdict;
			try {
				spend();
				if (stack.length >= limits.depth) throw new Error(`composition exceeded depth ${limits.depth}`);
				if (stack.includes(policy.instance)) throw new Error(`cycle: ${[...stack, policy.instance].join(" -> ")}`);
				const ran = await runInstance(policy.instance, entry, { args: policy.args ?? {}, call: request }, [...stack, policy.instance], signal, false);
				if (ran.error !== undefined) throw new Error(ran.error);
				verdict = ran.result;
			} catch (error) {
				verdict = { allow: false, violations: [{ msg: messageOf(error), policyId: policy.instance }] };
			}
			const allowed = verdict?.allow === true;
			note({ from: request.from, nsid: request.nsid, policy: policy.instance, ok: allowed, ...(!allowed && { violations: verdict?.violations ?? [] }) });
			if (!allowed) {
				const msgs = (Array.isArray(verdict?.violations) ? verdict.violations : []).map((v) => v?.msg).filter(Boolean);
				throw new Error(`policy ${policy.instance} denied${msgs.length ? `: ${msgs.join("; ")}` : ""}`);
			}
		}
	}

	// The `call` run-module gets for a run of instance `name`.
	const callOf = (name, held, stack) => async (nsid, { emit, ...request } = {}) => {
		const inst = instances[name];
		const signal = request.signal;
		if (TOKEN.test(nsid)) {
			// Refused before anything runs (the Rust host's too): a capability is one call, not a stream.
			if (request.subscribe) throw new Error(`${nsid} cannot be subscribed to`);
			const record = { from: name, nsid: "cap", cap: nsid.split(".")[2] };
			if (!held.has(nsid) || !caps.has(nsid)) {
				note({ ...record, ok: false, error: "not held" });
				throw new Error(`${nsid} is not a capability this run holds`);
			}
			spend();
			const { input, minted } = exportCaps(request.input, held, emit);
			try {
				const output = await invoke(nsid, input, signal);
				grant(held, output);
				note({ ...record, ok: true });
				return output;
			} catch (error) {
				note({ ...record, ok: false, error: messageOf(error) });
				throw error;
			} finally {
				revoke(minted);
			}
		}
		const routes = edges.get(`${name} ${nsid}`);
		const hook = modules[name].hooks?.find((h) => h.nsid === nsid);
		// Policies see the call, but not its stdin (secrets for the guest).
		const what = { from: name, nsid, params: request.params, input: withoutStdin(request.input), ...(request.proxy && { proxy: request.proxy }), ...(request.service && { service: request.service }) };
		if (!routes && hook && !hostsHook(hook, modules[name].calls)) {
			// An optional hook nothing serves: skip, never the network. A transport the
			// module lists in calls is the host's, below.
			note({ from: name, nsid, hook: "unrouted", ok: true });
			return null;
		}
		if (!routes) {
			const allowed = (list) => list === undefined || list.includes(nsid);
			if (!allowed(modules[name].calls) || !allowed(inst.calls)) throw new Error(`${name} may not call ${nsid}: no edge, and not in its calls allowlist`);
			await checkPolicies(inst.policies, what, stack, signal);
			const call = callFor?.(inst);
			if (typeof call !== "function") throw new Error(`${name} has no session for ${nsid}`);
			return call(nsid, request);
		}
		// refused before any route runs: an all-allow hook would answer a verdict, not a stream
		if (request.subscribe) throw new Error(`${nsid} is routed in this composition: no proxy, service or subscribe`);
		const routed = routedCall(name, nsid, request, emit, held, stack, what);
		if (hook?.combine === "all-allow") {
			// Every route must allow, in edge order; the first that does not is the answer (fail closed).
			for (const edge of routes) {
				const verdict = await routed(edge).then((output) => verdictOf(output, edge.to ?? edge.handler), (error) => verdictOf(undefined, edge.to ?? edge.handler, messageOf(error)));
				if (!verdict.allow) return verdict;
			}
			return { allow: true, violations: [] };
		}
		return routed(routes[0]);
	};

	// One routed call over `edge` (to an instance or a handler), its policies first.
	const routedCall = (name, nsid, request, emit, held, stack, what) => async (edge) => {
		const inst = instances[name];
		const signal = request.signal;
		const route = edge.to ? { to: edge.to, entry: entryOf(modules, edge) } : { handler: edge.handler };
		const record = { from: name, nsid, ...route, reason: edge.reason };
		let minted = [];
		try {
			if (request.proxy || request.service || request.subscribe) throw new Error(`${nsid} is routed in this composition: no proxy, service or subscribe`);
			if (edge.to && stack.length >= limits.depth) throw new Error(`composition exceeded depth ${limits.depth}`);
			if (edge.to && stack.includes(edge.to)) throw new Error(`cycle: ${[...stack, edge.to].join(" -> ")}`);
			await checkPolicies([...(inst.policies ?? []), ...(edge.policies ?? [])], { ...what, ...route, reason: edge.reason }, stack, signal);
			spend();
			const exported = exportCaps(request.input, held, emit);
			minted = exported.minted;
			const caller = { instance: name, nsid, reason: edge.reason };
			let output;
			if (edge.handler) {
				output = await handlers[edge.handler](asFunctions(exported.input, signal), { caller: name, nsid, reason: edge.reason, signal });
			} else {
				const body = exported.input !== null && typeof exported.input === "object" && !Array.isArray(exported.input) ? exported.input : { input: exported.input };
				const ran = await runInstance(edge.to, route.entry, { ...request.params, ...body, caller }, [...stack, edge.to], signal);
				if (ran.error !== undefined) throw new Error(ran.error);
				output = ran.result;
			}
			grant(held, output);
			note({ ...record, ok: true });
			return output;
		} catch (error) {
			note({ ...record, ok: false, error: messageOf(error) });
			throw error;
		} finally {
			revoke(minted);
		}
	};

	// JS functions in the root input are capabilities the root holds for the whole run.
	const hostCaps = [];
	const rootInput = walk(input, (v) => (typeof v === "function" ? (hostCaps.push(mint((data) => v(data))), { $cap: hostCaps.at(-1) }) : undefined));
	try {
		const ran = await runInstance(root, entry, rootInput, [root], signal);
		return { ...ran, graph, ...(graphDropped && { graphDropped }) };
	} finally {
		revoke(hostCaps);
	}
}