// node --test rust/host/conformance.test.mjs   (needs `cargo wasm` and `cargo host`; no network)
// The same modules, inputs and call stubs through the JS host (runModule / runCompose, wasm on V8)
// and the Rust host (atlibs: wasm on wasmtime, its own reactor, did:key and composition), asserting identical
// results, call logs, frames sent and ledgers. `node --import ./rust/host/swap.mjs --test <suites>` runs
// the existing suites' own assertions on the Rust engine as well.
import assert from "node:assert/strict";
import { execFile } from "node:child_process";
import { existsSync, readFileSync } from "node:fs";
import { createServer } from "node:http";
import { copyFile, mkdtemp, readFile, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { test } from "node:test";
import { promisify } from "node:util";

import { bindParameters, checkComposition, loadComposition, rfpPolicyRoutes, runCompose, staged } from "../../javascript/src/compose/index.js";
import { moduleOwner, runModule } from "../../javascript/src/run-module/index.js";
import { main as nodeCli } from "../../javascript/src/server/cli.mjs";
import { contextOptions, createNodeContext, hostCall, login, modulePolicy, sessionCall } from "../../javascript/src/server/index.js";
import { did, load, now } from "../src/harness.mjs";
import { atlibs, hostBin, rfpPolicyRoutesRust, runModuleRust, scopeFixture } from "./conformance.mjs";
import { loopbackBin } from "./loopback.mjs";

const ROOT = new URL("../../", import.meta.url);
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
const json = (value) => JSON.parse(JSON.stringify(value));

async function example(name) {
	const key = name.replaceAll("-", "_");
	const path = new URL(`target/wasm32-unknown-unknown/release/${key}.wasm`, ROOT).pathname;
	const spec = JSON.parse(await readFile(new URL(`examples/rust/${name}/module.json`, ROOT), "utf8"));
	const { module, run } = await load(new URL(`file://${path}`));
	return { path, run, mod: { module: new Uint8Array(await readFile(path)), calls: spec.calls, anyRecords: spec.anyRecords }, spec };
}

// seed: at:// URIs and the module (owner) each entry was made by
function memoryLedger(seed = []) {
	const rows = new Map(seed.map(([uri, module]) => [uri, { createdAt: now, ...(module && { module }) }]));
	return { rows, get: async (k) => rows.get(k) ?? null, put: async (k, v) => void rows.set(k, v), delete: async (k) => void rows.delete(k) };
}

// harness.mjs handle()'s call: respond({ nsid, ...request }) answers { ok, output } or, for a subscribe, the messages.
// Answers come back 5 ms apart in the order the calls were made, so which concurrent call settles first is
// decided by the scenario, not by either host's scheduler.
const callOf = (respond, sends) => {
	let made = 0;
	return async (nsid, request) => {
	const answer = await respond({ nsid, ...request });
	if (!request.subscribe) await sleep(5 * ++made);
	if (!request.subscribe) {
		if (answer?.ok === false) throw new Error(answer.error ?? "call failed");
		return answer?.output;
	}
	return { [Symbol.asyncIterator]: async function* () { yield* answer; }, send: async (data) => void sends.push({ nsid, data }) };
	};
};

// A ctx whose wasm runs on V8 with no imports (what composition modules need), one instance per module.
const rawWasm = () => {
	const instances = new Map();
	return {
		wasm: {
			run: async ({ module, entry, input }) => {
				if (!instances.has(module)) instances.set(module, new WebAssembly.Instance(new WebAssembly.Module(module), {}).exports);
				const x = instances.get(module);
				const data = new TextEncoder().encode(JSON.stringify(input));
				const ptr = x.alloc(data.length);
				new Uint8Array(x.memory.buffer, ptr, data.length).set(data);
				const packed = BigInt.asUintN(64, x[entry](ptr, data.length));
				return JSON.parse(new TextDecoder().decode(new Uint8Array(x.memory.buffer, Number(packed >> 32n), Number(packed & 0xffffffffn))));
			},
		},
	};
};
const settle = (promise) => promise.then((ran) => ({ ran }), (error) => ({ error: error.message }));

/**
 * One scenario on both hosts: fresh stubs (makeRespond()) and a fresh ledger each. Returns the JS outcome.
 * @param {{ ledger?: string[], foreign?: string[], inputs?: () => AsyncIterable<unknown>, budget?: object, call?: (respond, sends) => Function }} [options]
 *   ledger: entries this module made; foreign: the first has no owner (made before owners were recorded), the rest another module's
 */
async function both(ex, entry, input, makeRespond, { ledger = [], foreign = [], inputs, budget, call = callOf } = {}) {
	const owner = await moduleOwner(ex.mod);
	const outcome = async (rust) => {
		const sends = [];
		const managed = memoryLedger([...ledger.map((uri) => [uri, owner]), ...foreign.map((uri, i) => [uri, i ? "another-module" : undefined])]);
		const ctx = { storage: { managed_records: managed }, wasm: { run: async ({ entry, input }) => ex.run(entry, { ...input, now }) } };
		const options = { budget, ...(inputs && { inputs: inputs() }) };
		const ran = rust
			? await settle(runModuleRust(ctx, ex.mod, entry, input, call(makeRespond(), sends), { ...options, wasm: ex.path, now }))
			: await settle(runModule(ctx, ex.mod, entry, input, call(makeRespond(), sends), options));
		const owned = [...managed.rows].filter(([, v]) => v.module === owner).map(([k]) => k);
		return json({ ...ran, sends, ledger: [...managed.rows.keys()].sort(), owned: owned.sort() });
	};
	const js = await outcome(false);
	const rust = await outcome(true);
	assert.deepEqual(rust, js);
	return js;
}

const ok = (output) => ({ ok: true, output });

test("statusphere: anyRecords writes and the ledger match", async () => {
	const ex = await example("statusphere");
	const uri = (rkey) => `at://${did}/xyz.statusphere.status/${rkey}`;
	const respond = () => (call) => (call.nsid === "com.atproto.repo.listRecords" ? ok({ records: [{ uri: uri("old"), value: { status: "🙂" } }] }) : ok({ uri: uri("new"), cid: "bafy" }));
	const set = await both(ex, "set_status", { did, status: "🦋" }, respond);
	assert.equal(set.ran.result.uri, uri("new"));
	assert.deepEqual(set.ledger, [uri("new")]);
	await both(ex, "clear_status", { did }, respond);
	await both(ex, "set_status", { did, status: "🦋" }, () => () => ({ ok: false, error: "listRecords failed with 500" }));
	// without anyRecords the rkey guard refuses the delete of a record it did not create, and allows a managed one
	const guarded = { ...ex, mod: { ...ex.mod, anyRecords: undefined } };
	const refused = await both(guarded, "clear_status", { did }, respond);
	assert.ok(JSON.stringify(refused.ran.calls).includes("was not created by this module"));
	const managed = await both(guarded, "clear_status", { did }, respond, { ledger: [uri("old")] });
	assert.ok(!JSON.stringify(managed.ran.calls).includes("was not created by this module"));
	assert.deepEqual(set.owned, [uri("new")], "entries record the module that made them");
	// an entry another module made (or one with no owner, from before owners were recorded) is not this module's
	for (const foreign of [[uri("old")], ["at://x/y/z", uri("old")]]) {
		const other = await both(guarded, "clear_status", { did }, respond, { foreign });
		assert.ok(JSON.stringify(other.ran.calls).includes("was not created by this module"));
	}
});

test("hello: a public read at the account's DID service, a failed one and a missing field match", async () => {
	const ex = await example("hello");
	const listed = () => (call) => ok({ records: call.service === did ? [{ uri: `at://${did}/x.y.z/1`, value: {} }] : [] });
	const counted = await both(ex, "count", { did, collection: "x.y.z" }, listed);
	assert.deepEqual(counted.ran.result, { collection: "x.y.z", records: 1 });
	await both(ex, "count", { did, collection: "x.y.z" }, () => () => ({ ok: false, error: "listRecords failed with 400" }));
	await both(ex, "count", { did }, listed);
});

test("echo: a paced stream to its limit, a failing stream, inputs echoed until the timer", async () => {
	const ex = await example("echo");
	const service = "https://relay.example";
	const SUB = "com.atproto.sync.subscribeRepos";
	async function* paced(messages, ms = 20) {
		for (const m of messages) {
			await sleep(ms);
			yield m;
		}
	}
	const counted = await both(ex, "watch", { did, service, limit: 3 }, () => (call) => (call.subscribe ? paced([1, 2, 3, 4].map((seq) => ({ $type: `${SUB}#commit`, seq }))) : ok({ handle: "alice.test", did })));
	assert.equal(counted.ran.result.stopped, "limit");
	assert.ok(counted.ran.calls.some((c) => c.id === "feed" && c.cancelled));
	await both(ex, "watch", { did, service }, () => (call) => {
		if (!call.subscribe) return { ok: false, error: "RepoNotFound" };
		return (async function* () {
			await sleep(10);
			yield { seq: 9 };
			throw new Error("FutureCursor");
		})();
	});
	async function* quiet() {
		await sleep(1200);
	}
	const echoed = await both(ex, "watch", { did, service, seconds: 1 }, () => (call) => (call.subscribe ? quiet() : ok({ handle: "a", did })), {
		inputs: async function* () {
			await sleep(50);
			yield { $type: "#ping", n: 1 };
		},
	});
	assert.deepEqual(echoed.sends, [{ nsid: SUB, data: { $type: "#ping", n: 1 } }]);
	await both(ex, "watch", { did, service: "http://x" }, () => () => ok({}));
});

test("composition: the Rust CLI runs a staged composition (routed call, callback, policy) as runCompose does", async () => {
	const dir = await mkdtemp(join(tmpdir(), "atlibs-rust-compose-"));
	const manifestPath = join(dir, "manifest.json");
	await writeFile(manifestPath, JSON.stringify({
		$type: "com.fedfork.exec.composition",
		name: "conformance",
		instances: [{ name: "ui", module: "compose" }, { name: "lookup", module: "compose", input: { n: 0 } }, { name: "even", module: "compose" }],
		edges: [{ from: "ui", nsid: "com.example.compose.lookup", to: "lookup", reason: "ui needs n doubled", policies: [{ instance: "even", entry: "even" }] }],
		limits: { depth: 4, calls: 20 },
	}));
	const out = join(dir, "staged");
	await atlibs(["stage", "--target", "server", "--out", out, manifestPath]);
	const written = JSON.parse(await readFile(join(out, "composition.json"), "utf8"));
	const wasmCtx = rawWasm();
	const fileFetch = async (url) => new Response(await readFile(new URL(url)));
	const comp = await loadComposition(wasmCtx, written, { load: staged(new URL(`file://${out}/`), fileFetch) });
	// budget.log 1: the graph and each run's call log keep only their last entry and count the rest
	for (const [n, budget] of [[6], [5], [6, { log: 1 }]]) {
		const input = { nsid: "com.example.compose.lookup", input: { n } };
		const tokens = (value) => JSON.parse(JSON.stringify(value).replace(/cap\.[0-9a-f]{32}\./g, "cap.<random>."));
		const js = tokens(await runCompose(wasmCtx, comp, "ui", "caller", input, { budget }));
		const { stdout } = await promisify(execFile)(hostBin(), ["run", join(out, "composition.json"), "ui", "caller", JSON.stringify(input)], {
			env: { PATH: process.env.PATH, HOME: process.env.HOME, ATLIBS_SERVER_STATE: join(dir, "state"), ...(budget && { ATLIBS_BUDGET: JSON.stringify(budget) }) },
		});
		assert.deepEqual(tokens(JSON.parse(stdout)), js);
		assert.equal(js.result.q.ok, n % 2 === 0, JSON.stringify(js.result.q));
		if (budget) assert.ok(js.graph.length === 1 && js.graphDropped > 0, JSON.stringify(js.graph));
	}
});

test("composition parameters: bindParameters + runCompose and atlibs run --params give the same run, and the same refusals", async () => {
	const dir = await mkdtemp(join(tmpdir(), "atlibs-params-"));
	const manifestPath = join(dir, "manifest.json");
	await writeFile(manifestPath, JSON.stringify({
		$type: "com.fedfork.exec.composition",
		name: "parameters",
		parameters: [{ key: "n", type: "integer", minimum: 0, maximum: 100, values: { default: 4 } }],
		instances: [{ name: "lookup", module: "compose", input: { n: { $param: "n" } } }],
		exports: [{ nsid: "com.example.compose.lookup", instance: "lookup", entry: "lookup" }],
	}));
	const out = join(dir, "staged");
	await atlibs(["stage", "--target", "server", "--out", out, manifestPath]);
	const written = JSON.parse(await readFile(join(out, "composition.json"), "utf8"));
	const fileFetch = async (url) => new Response(await readFile(new URL(url)));
	const comp = await loadComposition(rawWasm(), written, { load: staged(new URL(`file://${out}/`), fileFetch) });
	const env = { PATH: process.env.PATH, HOME: process.env.HOME, ATLIBS_SERVER_STATE: join(dir, "state") };
	for (const params of [{}, { inputs: { n: 21 } }, { inputs: { n: 101 } }, { inputs: { n: "x" } }, { inputs: { m: 1 } }]) {
		const js = await settle(Promise.resolve().then(() => runCompose(rawWasm(), { ...comp, manifest: bindParameters(comp.manifest, params) }, "lookup", "lookup", {})));
		const rust = await settle(promisify(execFile)(hostBin(), ["run", join(out, "composition.json"), "lookup", "lookup", "{}", "--params", JSON.stringify(params)], { env }));
		if (js.error) assert.ok(rust.error?.includes(js.error), `${JSON.stringify(params)}: ${rust.error}`);
		else assert.deepEqual(JSON.parse(rust.ran.stdout), js.ran, JSON.stringify(params));
	}
	assert.equal((await settle(Promise.resolve().then(() => runCompose(rawWasm(), comp, "lookup", "lookup", {})))).error, "composition: parameter n is not bound");
});

test("atlibs stage --target server: modules and stage.json, no client files; both CLIs print the same bytes for a staged module", async () => {
	const dir = await mkdtemp(join(tmpdir(), "atlibs-rust-stage-"));
	const out = join(dir, "staged");
	const manifest = JSON.parse((await atlibs(["stage", "--target", "server", "--out", out, "compose"])).stdout);
	assert.deepEqual(Object.keys(manifest), ["target", "oauthPath", "sessions", "modules"]);
	assert.deepEqual(Object.keys(manifest.modules.compose), ["session", "wasm", "spec", "sha256", "pin"]);
	await assert.rejects(readFile(join(out, "oauth/compose/client-metadata.json")), /ENOENT/);
	const env = { PATH: process.env.PATH, HOME: process.env.HOME, ATLIBS_SERVER_STATE: join(dir, "state") };
	const input = JSON.stringify({ call: { input: { n: 3 } } });
	const exec = promisify(execFile);
	const rust = await exec(hostBin(), ["run", join(out, "stage.json"), "compose", "even", input], { env });
	const js = await exec(process.execPath, [new URL("../../javascript/src/server/cli.mjs", import.meta.url).pathname, "run", join(out, "modules/compose.wasm"), join(out, "modules/compose.json"), "even", input], { env });
	assert.equal(rust.stdout, js.stdout);
	assert.deepEqual(JSON.parse(rust.stdout).result, { allow: false, violations: [{ msg: "n 3 is not even", policyId: "even" }] });
	await assert.rejects(exec(hostBin(), ["run", join(out, "stage.json"), "nope", "even", input], { env }), /no module nope/);
});

// Review 3b #1/#2: a scripted handler (ctx.wasm.run, over the bridge on the Rust side) on both engines.
// Events already queued for an id a module cancels (or a timer it re-arms) never arrive; a subscription
// still opening when the run ends, or is cancelled, is stopped then, not opened.
async function scripted(handler, makeCall, calls) {
	const outcome = async (rust) => {
		const seen = [];
		const label = (e) => [e.type, e.id, e.data?.n].filter((v) => v !== undefined).join(":");
		const ctx = { wasm: { run: async ({ input }) => (seen.push(label(input.event)), handler(input.event)) } };
		const mod = { module: new Uint8Array(), calls };
		const log = [];
		const started = performance.now();
		const ran = await (rust ? runModuleRust : runModule)(ctx, mod, "go", {}, makeCall(log), {});
		const ms = performance.now() - started;
		await sleep(500); // anything a stopped open would still do happens by now
		return { seen, calls: json(ran.calls), error: ran.error ?? null, log, ms };
	};
	const js = await outcome(false);
	const rust = await outcome(true);
	const { ms: jsMs, ...jsOut } = js;
	const { ms: rustMs, ...rustOut } = rust;
	assert.deepEqual(rustOut, jsOut);
	return { ...jsOut, jsMs, rustMs };
}

test("queued events: cancel drops a queued result and messages, a re-armed timer fires once, a connecting subscription is stopped", async () => {
	const READ = "com.atproto.repo.getRecord";
	const SUB = "com.atproto.sync.subscribeRepos";
	const timed = () => async (nsid, { params }) => (await sleep(Number(params.ms)), {});
	// b's result is queued while the handler of a's runs; a then cancels b
	const cancelled = await scripted(async (event) => {
		if (event.type === "start") return { calls: [{ id: "a", nsid: READ, params: { ms: 5 } }, { id: "b", nsid: READ, params: { ms: 30 } }] };
		if (event.type === "result" && event.id === "a") return (await sleep(80), { cancel: ["b"] });
		return {};
	}, timed, [READ]);
	assert.deepEqual(cancelled.seen, ["start", "result:a"]);
	// the timer fired while a's handler ran; re-armed, it fires once (after 200 ms)
	const rearmed = await scripted(async (event) => {
		if (event.type === "start") return { calls: [{ id: "a", nsid: READ, params: { ms: 5 } }], timers: [{ id: "t", ms: 30 }] };
		if (event.type === "result" && event.id === "a") return (await sleep(80), { timers: [{ id: "t", ms: 200 }] });
		return {};
	}, timed, [READ]);
	assert.deepEqual(rearmed.seen, ["start", "result:a", "timer:t"]);
	// messages 1 and 2 queue while message 0's handler runs, which cancels the subscription
	const stream = () => async () => ({ async *[Symbol.asyncIterator]() { for (let n = 0; n < 3; n++) { yield { n }; await sleep(10); } await sleep(1e4); } });
	const sub = await scripted(async (event) => {
		if (event.type === "start") return { calls: [{ id: "s", nsid: SUB, subscribe: true }] };
		if (event.type === "message" && event.data.n === 0) return (await sleep(80), { cancel: ["s"] });
		return {};
	}, stream, [SUB]);
	assert.deepEqual(sub.seen, ["start", "message:s:0", "closed:s"]);
	// a subscription that takes 400 ms to open: done (or a cancel) stops the open; neither host waits for it
	const slow = (log) => async (nsid, { signal }) => {
		await sleep(400);
		if (signal?.aborted) throw new Error("aborted");
		log.push("opened");
		return { async *[Symbol.asyncIterator]() {}, close: () => log.push("close()") };
	};
	for (const end of [{ done: true }, { cancel: ["s"] }]) {
		const ran = await scripted(async (event) => (event.type === "start" ? { calls: [{ id: "s", nsid: SUB, subscribe: true }], timers: [{ id: "t", ms: 1 }] } : event.type === "timer" ? end : {}), slow, [SUB]);
		assert.deepEqual(ran.log, [], JSON.stringify(end));
		assert.ok(ran.rustMs < 380, `Rust waited ${ran.rustMs} ms for an open it had stopped`);
	}
});

// Manifest limits are positive numbers on both hosts (loadComposition and the CLI).
test("composition limits that are not positive numbers are refused the same by both hosts", async () => {
	const dir = await mkdtemp(join(tmpdir(), "atlibs-limits-"));
	for (const limits of [{ calls: "x" }, { calls: null }, { depth: 0 }, { other: 1 }]) {
		const manifest = { $type: "com.fedfork.exec.composition", name: "limits", instances: [{ name: "ui", module: "compose" }], limits };
		const file = join(dir, "composition.json");
		await writeFile(file, JSON.stringify(manifest));
		const js = await settle(loadComposition({}, manifest, { load: async () => { throw new Error("not reached"); } }));
		const rust = await settle(atlibs(["run", file, "ui", "even", "{}"]));
		assert.equal(js.error, "composition: limits must map depth, calls to positive numbers", JSON.stringify(limits));
		assert.ok(rust.error?.includes(js.error), `${JSON.stringify(limits)}: ${rust.error}`);
	}
});

// A host built without a transport refuses its calls with the same words: `cargo loopback` is atlibs
// without iroh (--no-default-features), the Node CLI's main() is given no transport.
test("a host built without iroh refuses iroh.connect the same through both CLIs", async (t) => {
	const bin = loopbackBin(hostBin(), "atlibs");
	if (!existsSync(bin)) return t.skip(`no ${bin}: cargo loopback`);
	const dir = await mkdtemp(join(tmpdir(), "atlibs-no-iroh-"));
	const env = { PATH: process.env.PATH, HOME: process.env.HOME, ATLIBS_SERVER_STATE: join(dir, "state") };
	const { path } = await example("echo");
	const manifest = new URL("examples/rust/echo/module.json", ROOT).pathname;
	const T = "endpointabex5k47heqqkdykv2b27ydg75jjaxeucm2ej74nt34a7up6wqlkyaiaenuhi5dqom5c6l3von3tcljrfzzgk3dbpexg4mbonfzg62bonruw42zof4";
	const args = ["run", path, manifest, "connect", JSON.stringify({ ticket: T, alpn: "echo/1", send: "hi\n" })];
	const failed = await promisify(execFile)(bin, args, { env }).then(() => assert.fail("a refused run exits 1"), (e) => e);
	const rust = failed.stdout ? JSON.parse(failed.stdout).error : failed.stderr.trim();
	const js = await nodeCli(args, env).then((ran) => ran.error, (e) => e.message);
	assert.match(js, /com\.fedfork\.exec\.iroh\.connect.*built without iroh/);
	assert.equal(rust, js);
});

// Hooks: one staged composition, its module's hooks edited in modules/compose.json (both hosts
// read that file). A required hook nothing routes, or a transport the host lacks, fails the load the
// same way in runCompose's loader and the Rust CLI; a hook naming entries binds only the entries reached.
test("hooks: both hosts refuse at load a required hook nothing serves (no route; a transport the host lacks)", async () => {
	const dir = await mkdtemp(join(tmpdir(), "atlibs-hooks-"));
	const manifestPath = join(dir, "manifest.json");
	await writeFile(manifestPath, JSON.stringify({
		$type: "com.fedfork.exec.composition",
		name: "hooks",
		instances: [{ name: "ui", module: "compose" }, { name: "helper", module: "compose" }],
		edges: [{ from: "ui", nsid: "com.example.compose.lookup", to: "helper", reason: "ui needs n doubled" }],
	}));
	const out = join(dir, "staged");
	await atlibs(["stage", "--target", "server", "--out", out, manifestPath]);
	const specPath = join(out, "modules/compose.json");
	const spec = JSON.parse(await readFile(specPath, "utf8"));
	const exportsOf = new Map();
	const wasmCtx = {
		wasm: {
			run: async ({ module, entry, input }) => {
				if (!exportsOf.has(module)) exportsOf.set(module, new WebAssembly.Instance(new WebAssembly.Module(module), {}).exports);
				const x = exportsOf.get(module);
				const data = new TextEncoder().encode(JSON.stringify(input));
				const ptr = x.alloc(data.length);
				new Uint8Array(x.memory.buffer, ptr, data.length).set(data);
				const packed = BigInt.asUintN(64, x[entry](ptr, data.length));
				return JSON.parse(new TextDecoder().decode(new Uint8Array(x.memory.buffer, Number(packed >> 32n), Number(packed & 0xffffffffn))));
			},
		},
	};
	const fileFetch = async (url) => new Response(await readFile(new URL(url)));
	const input = { nsid: "com.example.compose.lookup", input: { n: 2 } };
	const env = { PATH: process.env.PATH, HOME: process.env.HOME, ATLIBS_SERVER_STATE: join(dir, "state") };
	const rust = () => promisify(execFile)(hostBin(), ["run", join(out, "composition.json"), "ui", "caller", JSON.stringify(input)], { env })
		.then(({ stdout }) => JSON.parse(stdout).result.q.output.doubled, (e) => e.stderr.trim().replace(/^error: /, ""));
	const js = (provides) => loadComposition(wasmCtx, JSON.parse(readFileSync(join(out, "composition.json"), "utf8")), { load: staged(new URL(`file://${out}/`), fileFetch), provides })
		.then(async (comp) => (await runCompose(wasmCtx, comp, "ui", "caller", input)).result.q.output.doubled, (e) => e.message);
	const IROH = "com.fedfork.exec.iroh.connect";
	const WEBRTC = "com.fedfork.exec.webrtc.connect";
	const withHooks = (calls, hooks) => writeFile(specPath, JSON.stringify({ ...spec, calls, hooks: [...spec.hooks, ...hooks] }));

	assert.deepEqual([await rust(), await js([IROH])], [4, 4], "as staged: the Rust stage's pins are the JS modulePin");
	// a pin covers the record's grants: an edited module.json no longer matches, on either host
	await withHooks(spec.calls, [{ nsid: "com.example.hooks.ask", reason: "a new hook" }]);
	assert.deepEqual([await rust(), await js([IROH])], Array(2).fill("composition: instance ui module does not match its pin"));
	const composition = JSON.parse(await readFile(join(out, "composition.json"), "utf8"));
	await writeFile(join(out, "composition.json"), JSON.stringify({ ...composition, instances: composition.instances.map(({ pin, ...unpinned }) => unpinned) }));
	// helper is reached at lookup, which needs a hook no edge routes; ui runs caller, which does not
	await withHooks([], [{ nsid: "com.example.hooks.ask", required: true, entries: ["lookup"], reason: "lookup asks first" }]);
	const noRoute = "composition: instance helper needs hook com.example.hooks.ask (lookup asks first): no edge routes it";
	assert.deepEqual([await rust(), await js([IROH])], [noRoute, noRoute]);
	await withHooks([], [{ nsid: "com.example.hooks.ask", required: true, entries: ["even"], reason: "even asks first" }]);
	assert.deepEqual([await rust(), await js([IROH])], [4, 4], "an entry nothing reaches binds nothing");
	// a transport: the Rust CLI has iroh and no webrtc; the JS loader has what provides says
	await withHooks([IROH], [{ nsid: IROH, required: true, reason: "dials the guest" }]);
	assert.deepEqual([await rust(), await js([IROH])], [4, 4], "a host with iroh loads it");
	const noIroh = `composition: instance ui needs hook ${IROH} (dials the guest): this host was built without iroh`;
	assert.equal(await js([]), noIroh);
	await withHooks([WEBRTC], [{ nsid: WEBRTC, required: true, reason: "talks to a page" }]);
	const noWebrtc = `composition: instance ui needs hook ${WEBRTC} (talks to a page): this host was built without webrtc`;
	assert.deepEqual([await rust(), await js([IROH])], [noWebrtc, noWebrtc]);
});

// No wasm: a lockfile's specs and pins only. `atlibs compose check` and checkComposition agree.
test("compose check: both hosts check a composition from module.json fields alone, the same way", async () => {
	const dir = await mkdtemp(join(tmpdir(), "atlibs-check-"));
	const H = "com.example.check.ask";
	const specs = {
		asker: { name: "asker", calls: [], hooks: [{ nsid: H, default: "judge", reason: "asks" }], config: [{ key: "n", label: "n", type: "integer" }] },
		judge: { name: "judge", calls: [], implements: [{ nsid: H, entry: "evaluate" }] },
		mute: { name: "mute", calls: [] },
	};
	const pins = { asker: "a".repeat(64), judge: "b".repeat(64), mute: "c".repeat(64) };
	for (const [key, spec] of Object.entries(specs)) await writeFile(join(dir, `${key}.json`), JSON.stringify(spec));
	await writeFile(join(dir, "atlibs.lock.json"), JSON.stringify({ modules: Object.fromEntries(Object.keys(specs).map((k) => [k, { pin: pins[k], spec: `${k}.json` }])) }));
	const env = { PATH: process.env.PATH, HOME: process.env.HOME, ATLIBS_SERVER_STATE: join(dir, "state") };
	const both = async (manifest) => {
		await writeFile(join(dir, "c.json"), JSON.stringify({ name: "c", ...manifest }));
		const rust = await promisify(execFile)(hostBin(), ["compose", "check", join(dir, "c.json")], { env }).then(({ stdout }) => JSON.parse(stdout).instances.map((i) => i.name), (e) => `${e.code} ${e.stderr.trim()}`);
		const js = await checkComposition({ name: "c", ...manifest }, (key) => ({ ...specs[key], pin: pins[key] })).then((m) => m.instances.map((i) => i.name), (e) => `1 ${e.message}`);
		return [rust, js];
	};
	const a = { name: "a", module: "asker", input: { n: 2 } };
	const same = async (manifest, want) => assert.deepEqual(await both(manifest), [want, want], JSON.stringify(manifest));
	await same({ instances: [a] }, ["a", "a-judge"]);
	await same({ instances: [{ ...a, pin: pins.asker }, { name: "j", module: "judge", pin: pins.judge }], edges: [{ from: "a", nsid: H, to: "j", reason: "r" }] }, ["a", "j"]);
	await same({ instances: [{ ...a, pin: pins.judge }] }, "1 composition: instance a module does not match its pin");
	await same({ instances: [{ ...a, pin: pins.asker }] }, `1 composition: instance a is pinned, but its hook ${H} default judge is not: route the hook to a pinned instance`);
	await same({ instances: [a, { name: "m", module: "mute" }], edges: [{ from: "a", nsid: H, to: "m", reason: "r" }] }, `1 composition: edge a ${H}: m (mute) does not implement ${H}`);
	await same({ instances: [{ ...a, input: { n: "2" } }] }, "1 composition: instance a: input n must be of type integer");
	await same({ instances: [{ name: "m", module: "mute", input: { n: 1 } }] }, "1 composition: instance m: input n is not in mute's config");
});

// vm-mcp's provision and release both default to the requester: one shared instance, as compositions route them.
test("hooks: two hooks defaulting to one module load on both hosts, one shared instance serving each at its entry", async () => {
	const dir = await mkdtemp(join(tmpdir(), "atlibs-hooks-shared-"));
	const out = join(dir, "staged");
	await atlibs(["stage", "--target", "server", "--out", out, "compose"]);
	const spec = JSON.parse(await readFile(join(out, "modules/compose.json"), "utf8"));
	const [LOOKUP, CONFIRM] = ["com.example.compose.lookup", "com.example.compose.confirm"];
	await writeFile(join(out, "modules/compose.json"), JSON.stringify({ ...spec, hooks: [LOOKUP, CONFIRM].map((nsid) => ({ nsid, default: "two", reason: nsid })) }));
	const { hooks, ...plain } = spec;
	await writeFile(join(out, "modules/two.json"), JSON.stringify({ ...plain, implements: [{ nsid: LOOKUP, entry: "lookup" }, { nsid: CONFIRM, entry: "lookup" }] }));
	await copyFile(join(out, "modules/compose.wasm"), join(out, "modules/two.wasm"));
	const manifest = { $type: "com.fedfork.exec.composition", name: "shared", instances: [{ name: "ui", module: "compose" }] };
	await writeFile(join(out, "composition.json"), JSON.stringify(manifest));
	const env = { PATH: process.env.PATH, HOME: process.env.HOME, ATLIBS_SERVER_STATE: join(dir, "state") };
	const rust = (input) => promisify(execFile)(hostBin(), ["run", join(out, "composition.json"), "ui", "caller", JSON.stringify(input)], { env })
		.then(({ stdout }) => JSON.parse(stdout).result.q.output.doubled, (e) => e.stderr.trim());
	const ctx = createNodeContext({ id: "hooks-shared", stateDir: join(dir, "js-state") });
	const fileFetch = async (url) => new Response(await readFile(new URL(url)));
	const comp = await loadComposition(ctx, manifest, { load: staged(new URL(`file://${out}/`), fileFetch) });
	assert.deepEqual(comp.manifest.instances, [{ name: "ui", module: "compose" }, { name: "ui-two", module: "two" }]);
	const js = async (input) => (await runCompose(ctx, comp, "ui", "caller", input)).result.q.output.doubled;
	for (const [nsid, n] of [[LOOKUP, 2], [CONFIRM, 3]]) assert.deepEqual([await rust({ nsid, input: { n } }), await js({ nsid, input: { n } })], [n * 2, n * 2], nsid);
});

test("hooks: the Node CLI and the Rust CLI refuse a run whose entry needs a transport they lack, and run its other entries", async () => {
	const dir = await mkdtemp(join(tmpdir(), "atlibs-hooks-cli-"));
	const out = join(dir, "staged");
	await atlibs(["stage", "--target", "server", "--out", out, "compose"]);
	const specPath = join(out, "modules/compose.json");
	const spec = JSON.parse(await readFile(specPath, "utf8"));
	const WEBRTC = "com.fedfork.exec.webrtc.connect";
	await writeFile(specPath, JSON.stringify({ ...spec, calls: [WEBRTC], hooks: [{ nsid: WEBRTC, required: true, entries: ["caller"], reason: "talks to a page" }] }));
	const env = { PATH: process.env.PATH, HOME: process.env.HOME, ATLIBS_SERVER_STATE: join(dir, "state") };
	const cli = async (entry, input) => {
		const args = ["run", join(out, "modules/compose.wasm"), specPath, entry, JSON.stringify(input)];
		const rust = await promisify(execFile)(hostBin(), args, { env }).then(({ stdout }) => JSON.parse(stdout), (e) => (e.stdout ? JSON.parse(e.stdout) : { error: e.stderr.trim() }));
		const js = await nodeCli(args, env);
		return [rust, js].map((ran) => ran.error ?? ran.result);
	};
	const refused = `${spec.name} needs hook ${WEBRTC} (talks to a page): this host was built without webrtc`;
	assert.deepEqual(await cli("caller", { nsid: WEBRTC, input: {} }), [refused, refused]);
	const even = { allow: true, violations: [] };
	assert.deepEqual(await cli("even", { call: { input: { n: 2 } } }), [even, even], "an entry that needs no transport runs");
});

// Review 4a L2: a module subscribing to a routed hook (all-allow here) is refused before any route runs, on
// both hosts; the Rust host used to panic in the subscription task and hang until wallMs.
test("hooks: a subscribe to a routed all-allow hook is refused up front on both hosts", async () => {
	const dir = await mkdtemp(join(tmpdir(), "atlibs-hooks-sub-"));
	const IROH = "com.fedfork.exec.iroh.connect";
	await writeFile(join(dir, "manifest.json"), JSON.stringify({
		instances: [{ name: "e", module: new URL("examples/rust/echo", ROOT).pathname }, { name: "p", module: new URL("examples/rust/compose", ROOT).pathname }],
	}));
	const out = join(dir, "staged");
	await atlibs(["stage", "--target", "server", "--out", out, join(dir, "manifest.json")]);
	const specPath = join(out, "modules/echo.json");
	const spec = JSON.parse(await readFile(specPath, "utf8"));
	await writeFile(specPath, JSON.stringify({ ...spec, hooks: spec.hooks.map((h) => (h.nsid === IROH ? { ...h, combine: "all-allow" } : h)) }));
	const policy = JSON.parse(await readFile(join(out, "modules/compose.json"), "utf8"));
	await writeFile(join(out, "modules/compose.json"), JSON.stringify({ ...policy, implements: [...policy.implements, { nsid: IROH, entry: "even" }] }));
	const composition = JSON.parse(await readFile(join(out, "composition.json"), "utf8"));
	const edges = [{ from: "e", nsid: IROH, to: "p", reason: "p allows the dial" }];
	await writeFile(join(out, "composition.json"), JSON.stringify({ ...composition, edges, instances: composition.instances.map(({ pin, ...unpinned }) => unpinned) }));
	const input = JSON.stringify({ ticket: "x" });
	const env = { PATH: process.env.PATH, HOME: process.env.HOME, ATLIBS_SERVER_STATE: join(dir, "state"), ATLIBS_BUDGET: JSON.stringify({ wallMs: 10000 }) };
	const rust = JSON.parse((await promisify(execFile)(hostBin(), ["run", join(out, "composition.json"), "e", "session", input], { env })).stdout).result.error;
	const wasmCtx = {
		wasm: {
			run: async ({ module, entry, input }) => {
				const x = new WebAssembly.Instance(new WebAssembly.Module(module), {}).exports;
				const data = new TextEncoder().encode(JSON.stringify(input));
				const ptr = x.alloc(data.length);
				new Uint8Array(x.memory.buffer, ptr, data.length).set(data);
				const packed = BigInt.asUintN(64, x[entry](ptr, data.length));
				return JSON.parse(new TextDecoder().decode(new Uint8Array(x.memory.buffer, Number(packed >> 32n), Number(packed & 0xffffffffn))));
			},
		},
	};
	const fileFetch = async (url) => new Response(await readFile(new URL(url)));
	const comp = await loadComposition(wasmCtx, JSON.parse(await readFile(join(out, "composition.json"), "utf8")), { load: staged(new URL(`file://${out}/`), fileFetch), provides: [IROH] });
	const js = (await runCompose(wasmCtx, comp, "e", "session", JSON.parse(input))).result.error;
	assert.deepEqual([rust, js], Array(2).fill(`${IROH} is routed in this composition: no proxy, service or subscribe`));
});

// RFP-attached policies: the ghalite records an RFP attaches route the same on both hosts
// (kcp names canonicalised, args as input.args, siblings of an at:// instance, added after the hook's own
// routes; unknown names, records that are not objects and a hook not routed yet fail closed).
test("rfpPolicyRoutes: RFP-attached policy records route to the same instances and edges on both hosts", async () => {
	const at = { instances: [{ name: "requester", module: "at://did:plc:a/com.fedfork.exec.wasm.module/requester" }], edges: [{ from: "requester", nsid: "com.publicdomainrelay.temp.market.evaluatePolicy", to: "requester" }] };
	const local = {
		instances: [{ name: "requester", module: "requester", session: "requester" }, { name: "vouch", module: "policy-requester-tangled-vouch" }, { name: "payload", module: "policy-bid-payload" }],
		edges: ["vouch", "payload"].map((to) => ({ from: "requester", nsid: at.edges[0].nsid, to })),
	};
	const cases = [
		[at, "requester", [{ name: "only-me" }, { name: " Tangled_Vouch ", args: { min: 2 } }, {}, { name: "open", args: null }]],
		[local, "requester", [{ name: "mutuals" }, { name: "bidder-only-me" }, { name: 7 }]],
		[{ instances: [{ name: "requester", module: "requester" }] }, "requester", []],
		[at, "requester", [{ name: "my-own-workflow" }]],
		[at, "nope", [{ name: "open" }]],
		[at, "requester", [null]],
		[at, "requester", ["open"]],
		[{ instances: at.instances }, "requester", [{ name: "open" }]],
	];
	for (const [manifest, from, records] of cases) {
		const js = await settle(Promise.resolve().then(() => rfpPolicyRoutes(manifest, from, records)));
		const rust = await settle(rfpPolicyRoutesRust(manifest, from, records));
		assert.deepEqual(rust, js, JSON.stringify(records));
	}
	assert.deepEqual(await settle(rfpPolicyRoutesRust(at, "requester", [null])), await settle(Promise.reject(new Error("composition: no policy module for the RFP's policy null"))));
});

// Server sessions: a signed-in run (here an app password, which holds the whole account) may
// make only the calls its own module's session record allows, on both CLIs and in compositions (each instance
// its own module's), with policyCall's refusal text; a refused call never reaches the PDS.
test("per-module scope: the same module, session and calls are allowed or refused alike by both hosts", async () => {
	const dir = await scopeFixture();
	const reached = [];
	const server = createServer((req, res) => {
		let body = "";
		req.on("data", (c) => (body += c));
		req.on("end", () => {
			const nsid = new URL(req.url, "http://x").pathname.slice("/xrpc/".length);
			const input = body ? JSON.parse(body) : {};
			const reply = (out) => res.writeHead(200, { "content-type": "application/json" }).end(JSON.stringify(out));
			if (nsid === "com.atproto.server.createSession") return reply({ did, handle: "alice.test", accessJwt: "a", refreshJwt: "r", didDoc: { service: [{ id: "#atproto_pds", serviceEndpoint: base }] } });
			reached.push([nsid, input.collection ?? req.headers["atproto-proxy"] ?? req.headers.authorization ?? null]);
			if (nsid === "com.atproto.server.getServiceAuth") return reply({ token: `T-${new URL(req.url, "http://x").searchParams.get("lxm")}` });
			reply(nsid === "com.atproto.repo.createRecord" ? { uri: `at://${did}/${input.collection}/1`, cid: "bafy" } : { ok: nsid });
		});
	});
	await new Promise((resolve) => server.listen(0, "127.0.0.1", resolve));
	const base = `http://127.0.0.1:${server.address().port}`;
	const account = ["--app-password", "alice.test", "--password-env", "PW", "--pds", base];
	const env = (host) => ({ PW: "pw", ATLIBS_ALLOW_PRIVATE: "1", ATLIBS_SERVER_STATE: join(dir, `state-${host}`) });
	const outcome = (ran) => ({ result: ran.result, error: ran.error });
	const rust = (args) => atlibs([...args, ...account], { env: env("rust") }).then(({ stdout }) => stdout, (e) => e.stdout || Promise.reject(e)).then((out) => outcome(JSON.parse(out)));
	const create = (collection) => ({ nsid: "com.atproto.repo.createRecord", input: { repo: did, collection, record: {} } });
	const ask = (nsid) => ({ nsid, input: {} });
	const cases = [
		["a", "caller", create("com.example.a"), true],
		["a", "caller", create("com.example.b"), "a may not call com.atproto.repo.createRecord with sessions sa"],
		["b", "caller", create("com.example.b"), true],
		["b", "caller", create("com.example.a"), "b may not call com.atproto.repo.createRecord with sessions sb"],
		["a", "caller", ask("com.example.a.ping"), true],
		["b", "caller", ask("com.example.a.ping"), "b may not call com.example.a.ping with sessions sb"],
		["a", "caller", ask("com.atproto.repo.uploadBlob"), true],
		["b", "caller", ask("com.atproto.repo.uploadBlob"), "b may not call com.atproto.repo.uploadBlob with sessions sb"],
		["c", "caller", ask("com.atproto.repo.listRecords"), true],
		["c", "caller", create("com.example.a"), "c may not call com.atproto.repo.createRecord with no connected session"],
		["chat", "caller", { nsid: "com.example.chat.post", proxy: "did:web:chat.example#chat" }, true],
		["evil", "caller", { nsid: "com.example.chat.post", proxy: "did:web:chat.example#chat" }, "evil may not call com.example.chat.post with sessions evil"],
	];
	try {
		for (const [key, entry, input, expected] of cases) {
			reached.length = 0;
			const args = ["run", join(dir, "modules", `${key}.wasm`), join(dir, "modules", `${key}.json`), entry, JSON.stringify(input)];
			const js = outcome(await nodeCli([...args, ...account], env("node")));
			assert.equal(reached.length, expected === true ? 1 : 0, `${key} ${JSON.stringify(input)}: ${JSON.stringify(reached)}`);
			assert.deepEqual(await rust(args), js, `${key} ${JSON.stringify(input)}`);
			const refused = entry === "caller" ? js.result.q.error : js.error;
			if (expected === true) assert.equal(refused, undefined, `${key}: ${refused}`);
			else assert.match(refused, new RegExp(expected));
		}
		// a composition: each instance is held to its own module's session
		const ctx = createNodeContext(contextOptions(join(dir, "state-node")));
		try {
			const comp = await loadComposition(ctx, JSON.parse(readFileSync(join(dir, "comp.json"), "utf8")), { load: staged(new URL(`file://${dir}/`), async (url) => new Response(await readFile(url))) });
			const session = await login({ identifier: "alice.test", password: "pw", service: base });
			const call = hostCall({ ctx, xrpc: session.xrpc, pds: session.pds });
			const policies = Object.fromEntries(await Promise.all(["a", "b", "h"].map(async (key) => [key, await modulePolicy(ctx, key, comp.modules[key], join(dir, "modules"))])));
			const callFor = (inst) => sessionCall(policies[inst.module], inst.module, call);
			for (const [instance, collection, refused] of [["a", "com.example.a"], ["a", "com.example.b", "a may not call com.atproto.repo.createRecord with sessions sa"], ["b", "com.example.b"], ["b", "com.example.a", "b may not call com.atproto.repo.createRecord with sessions sb"]]) {
				const input = JSON.stringify(create(collection));
				const js = await runCompose(ctx, comp, instance, "caller", JSON.parse(input), { callFor });
				const { result } = await rust(["run", join(dir, "comp.json"), instance, "caller", input]);
				assert.deepEqual(result.q, js.result.q, `${instance} -> ${collection}`);
				assert.equal(js.result.q.error, refused, `${instance} -> ${collection}`);
			}
			// the host's http serviceAuth: minted with h's own session (callFor), though h lists no getServiceAuth
			const http = { [base]: { serviceAuth: { aud: "did:web:svc.example", lxm: "x.y.z" } } };
			const fetchIt = JSON.stringify({ nsid: "com.fedfork.exec.http.fetch", input: { url: `${base}/svc` } });
			reached.length = 0;
			const js = await runCompose(ctx, comp, "h", "caller", JSON.parse(fetchIt), { callFor, http });
			assert.equal(js.result.q.ok, true, JSON.stringify(js.result.q));
			assert.deepEqual(reached, [["com.atproto.server.getServiceAuth", "Bearer a"], ["", "Bearer T-x.y.z"]]);
			reached.length = 0;
			const { result } = await atlibs(["run", join(dir, "comp.json"), "h", "caller", fetchIt, ...account], { env: { ...env("rust"), ATLIBS_HTTP: JSON.stringify(http) } }).then(({ stdout }) => JSON.parse(stdout));
			const seen = (q) => ({ ok: q.ok, status: q.output?.status, body: q.output?.body });
			assert.deepEqual(seen(result.q), seen(js.result.q));
			assert.deepEqual(reached, [["com.atproto.server.getServiceAuth", "Bearer a"], ["", "Bearer T-x.y.z"]]);
		} finally {
			ctx.close();
		}
	} finally {
		server.close();
	}
});