johnandersen777.bsky.social/atlibs

Switch repository

Loading…

import assert from "node:assert/strict";
import { test } from "node:test";

import { decodeCbor, decodeFrame, encodeCbor, encodeFrame, runModule, subscribeXrpc, wsUrl } from "./index.js";

const READ = "com.atproto.repo.getRecord";
const SUB = "com.atproto.sync.subscribeRepos";
const DELETE = "com.atproto.repo.deleteRecord";
const mod = (extra) => ({ module: new Uint8Array(), calls: [READ, SUB, DELETE], ...extra });
// The handler stands in for the wasm export: ctx.wasm.run(entry, input) -> handler(input).
const ctxOf = (handler, storage) => ({ storage, wasm: { run: async ({ input }) => handler(input) } });
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

// A subscription stand-in: push() a message, end() it, fail() it; records sends and close.
function channel() {
	const queue = [];
	const sent = [];
	let wake;
	let finished = false;
	let failure;
	const self = {
		sent,
		closed: false,
		push: (m) => (queue.push(m), wake?.()),
		end: () => ((finished = true), wake?.()),
		fail: (e) => ((failure = e), (finished = true), wake?.()),
		send: async (data) => void sent.push(data),
		close: () => ((self.closed = true), self.end()),
		async *[Symbol.asyncIterator]() {
			for (;;) {
				while (!queue.length && !finished) await new Promise((r) => (wake = r));
				if (queue.length) yield queue.shift();
				else if (failure) throw failure;
				else return;
			}
		},
	};
	return self;
}

test("results arrive one by one as calls finish, with state carried and no barrier", async () => {
	const seen = [];
	const handler = ({ event, state }) => {
		seen.push(event.type === "result" ? event.id : event.type);
		if (event.type === "start") return { calls: [{ id: "slow", nsid: READ, params: { ms: 30 } }, { id: "fast", nsid: READ, params: { ms: 1 } }], state: { n: 0 } };
		return { state: { n: state.n + 1 }, result: { last: event.id, n: state.n + 1, output: event.output } };
	};
	const call = async (nsid, { params }) => (await sleep(Number(params.ms)), { ms: params.ms });
	const ran = await runModule(ctxOf(handler), mod(), "go", {}, call);
	assert.deepEqual(seen, ["start", "fast", "slow"]);
	assert.deepEqual(ran.result, { last: "slow", n: 2, output: { ms: "30" } });
	assert.equal(ran.events, 3);
	assert.deepEqual(ran.calls.map((c) => c.id), ["fast", "slow"]);
});

test("calls past inFlight queue FIFO; refusals (allowlist, rkey guard, bad call, budget) are result events", async () => {
	let live = 0;
	let peak = 0;
	const order = [];
	const call = async (nsid, { params }) => {
		peak = Math.max(peak, ++live);
		order.push(params.i);
		await sleep(5);
		live--;
		return {};
	};
	const errors = {};
	const handler = ({ event }) => {
		if (event.type === "start") {
			return {
				calls: [
					...[0, 1, 2, 3, 4].map((i) => ({ id: `c${i}`, nsid: READ, params: { i } })),
					{ id: "nope", nsid: "com.example.not.allowed" },
					{ id: "del", nsid: DELETE, input: { repo: "did:plc:a", collection: "x.y.z", rkey: "k" } },
					{ nsid: READ },
					{ id: "c0", nsid: READ },
					{ id: "over", nsid: READ },
				],
			};
		}
		if (!event.ok) errors[event.id ?? "noid"] = event.error;
		return {};
	};
	const ran = await runModule(ctxOf(handler), mod(), "go", {}, call, { budget: { inFlight: 2, calls: 6 } });
	assert.equal(peak, 2);
	assert.deepEqual(order, ["0", "1", "2", "3", "4"]);
	assert.match(errors.nope, /does not allow/);
	assert.match(errors.del, /was not created by this module/);
	assert.match(errors.noid, /needs an id/);
	assert.match(errors.c0, /already pending/);
	assert.match(errors.over, /exceeded 6 calls/);
	assert.equal(ran.error, undefined);
});

test("a subscription yields messages, takes sends in order, and ends with closed", async () => {
	const feed = channel();
	const got = [];
	let request;
	const call = async (nsid, req) => ((request = req), feed);
	const handler = ({ event, state }) => {
		if (event.type === "start") return { calls: [{ id: "feed", nsid: SUB, params: { cursor: 5 }, subscribe: true, framing: "cbor" }], send: [{ id: "feed", data: "hello" }], state: { count: 0 } };
		if (event.type === "message") {
			got.push(event.data);
			return { state: { count: state.count + 1 }, send: [{ id: "feed", data: { ack: event.data.seq } }] };
		}
		if (event.type === "closed") return { result: { count: state.count, closed: event } };
		return {};
	};
	const running = runModule(ctxOf(handler), mod(), "watch", {}, call);
	await sleep(5);
	feed.push({ seq: 1 });
	feed.push({ seq: 2 });
	await sleep(5);
	feed.end();
	const ran = await running;
	assert.equal(request.subscribe, true);
	assert.equal(request.framing, "cbor");
	assert.deepEqual(request.params, { cursor: "5" });
	assert.ok(request.signal instanceof AbortSignal);
	assert.deepEqual(got, [{ seq: 1 }, { seq: 2 }]);
	assert.deepEqual(feed.sent, ["hello", { ack: 1 }, { ack: 2 }]);
	assert.deepEqual(ran.result, { count: 2, closed: { type: "closed", id: "feed" } });
	assert.deepEqual(ran.calls, [{ id: "feed", nsid: SUB, subscribe: true, messages: 2, ok: true }]);
});

test("cancel closes a subscription, timers fire, send to a closed id is refused, done ends the run", async () => {
	const feed = channel();
	const seen = [];
	const handler = ({ event }) => {
		seen.push(event.type === "result" ? `result:${event.ok}` : event.type);
		if (event.type === "start") return { calls: [{ id: "feed", nsid: SUB, subscribe: true }], timers: [{ id: "stop", ms: 10 }] };
		if (event.type === "timer" && event.id === "stop") return { cancel: ["feed"] };
		if (event.type === "closed") return { result: event.error, timers: [{ id: "again", ms: 1 }], send: [{ id: "feed", data: "late" }] };
		if (event.type === "result") return {};
		return { done: true };
	};
	const ran = await runModule(ctxOf(handler), mod(), "watch", {}, async () => feed);
	assert.deepEqual(seen, ["start", "timer", "closed", "result:false", "timer"]);
	assert.equal(ran.result, "cancelled");
	assert.equal(feed.closed, true);
	assert.deepEqual(feed.sent, []);
});

test("a subscription that fails closes with its error; a stream without send refuses sends", async () => {
	const feed = channel();
	const plain = { async *[Symbol.asyncIterator]() { yield 1; } };
	const events = [];
	const handler = ({ event }) => {
		events.push(event);
		if (event.type === "start") return { calls: [{ id: "a", nsid: SUB, subscribe: true }, { id: "b", nsid: SUB, subscribe: true }], send: [{ id: "b", data: 1 }] };
		return {};
	};
	let n = 0;
	const running = runModule(ctxOf(handler), mod(), "watch", {}, async () => (n++ ? plain : feed));
	await sleep(5);
	feed.fail(new Error("relay went away"));
	await running;
	const byType = (t, id) => events.find((e) => e.type === t && e.id === id);
	assert.equal(byType("closed", "a").error, "relay went away");
	assert.match(byType("result", "b").error, /cannot send/);
	assert.equal(byType("message", "b").data, 1);
});

test("abort stops a run: in-flight calls get the signal and subscriptions close", async () => {
	const feed = channel();
	let callSignal;
	const call = async (nsid, req) => {
		if (req.subscribe) return feed;
		callSignal = req.signal;
		return new Promise(() => {});
	};
	const handler = ({ event }) => (event.type === "start" ? { calls: [{ id: "hang", nsid: READ }, { id: "feed", nsid: SUB, subscribe: true }] } : {});
	const stop = new AbortController();
	setTimeout(() => stop.abort(), 10);
	const ran = await runModule(ctxOf(handler), mod(), "go", {}, call, { signal: stop.signal });
	assert.equal(ran.error, "aborted");
	assert.equal(callSignal.aborted, true);
	assert.equal(feed.closed, true);
});

test("per-call timeout fails the call; events and wall budgets end the run with an error", async () => {
	const timeout = await runModule(
		ctxOf(({ event }) => (event.type === "start" ? { calls: [{ id: "t", nsid: READ, timeoutMs: 5 }] } : { result: event.error })),
		mod(), "go", {}, () => new Promise(() => {}),
	);
	assert.match(timeout.result, /timed out after 5 ms/);
	// a huge timeoutMs is clamped to wallMs: no setTimeout overflow (fires at once) nor a Rust panic
	const huge = await runModule(
		ctxOf(({ event }) => (event.type === "start" ? { calls: [{ id: "t", nsid: READ, timeoutMs: 1e300 }] } : { result: event.ok ? "ok" : event.error })),
		mod(), "go", {}, () => new Promise((r) => setTimeout(() => r({}), 30)), { budget: { wallMs: 2000 } },
	);
	assert.equal(huge.result, "ok");
	const timers = await runModule(ctxOf(({ event }) => (event.type === "start" ? { timers: [{ id: "x", ms: 1e300 }] } : { result: "fired" })), mod(), "go", {}, async () => ({}), { budget: { wallMs: 30 } });
	assert.equal(timers.error, "Module exceeded 30 ms");

	const feed = channel();
	for (let i = 0; i < 10; i++) feed.push(i);
	const ignored = await runModule(ctxOf(({ event }) => (event.type === "start" ? { timers: [{ id: "x", ms: 20 }] } : { result: "fired", done: true })), mod(), "go", {}, async () => ({}), { budget: { wallMs: "5", events: -1, calls: Number.NaN } });
	assert.equal(ignored.result, "fired", "a budget value that is not a positive finite number leaves the default");
	const chatty = await runModule(ctxOf(({ event }) => (event.type === "start" ? { calls: [{ id: "f", nsid: SUB, subscribe: true }] } : {})), mod(), "go", {}, async () => feed, { budget: { events: 4 } });
	assert.equal(chatty.error, "Module exceeded 4 events");
	assert.equal(feed.closed, true);

	const slow = await runModule(ctxOf(({ event }) => (event.type === "start" ? { timers: [{ id: "x", ms: 1000 }] } : {})), mod(), "go", {}, async () => ({}), { budget: { wallMs: 10 } });
	assert.equal(slow.error, "Module exceeded 10 ms");
});

test("a host may set calls, events and wallMs unlimited; callsPerMinute refuses past the rate", async () => {
	// 150 calls, past the default 100, each answered; done after the last
	const many = ({ event, state }) => {
		if (event.type === "start") return { calls: Array.from({ length: 150 }, (_, i) => ({ id: `c${i}`, nsid: READ })) };
		const ok = (state ?? 0) + (event.ok ? 1 : 0);
		return ok === 150 ? { result: ok, done: true } : { state: ok };
	};
	const unlimited = await runModule(ctxOf(many), mod(), "go", {}, async () => ({}), { budget: { calls: Infinity, events: Infinity, wallMs: Infinity } });
	assert.equal(unlimited.error, undefined);
	assert.equal(unlimited.result, 150);
	// no deadline: a timer and a per-call timeout still fire (min(ms, wallMs)), a huge timer is clamped and only
	// the signal ends the run
	const timed = await runModule(
		ctxOf(({ event }) => (event.type === "start" ? { timers: [{ id: "x", ms: 30 }], calls: [{ id: "t", nsid: READ, timeoutMs: 5 }] } : event.type === "timer" ? { result: "fired", done: true } : {})),
		mod(), "go", {}, () => new Promise(() => {}), { budget: { wallMs: Infinity } },
	);
	assert.equal(timed.error, undefined);
	assert.equal(timed.result, "fired");
	assert.match(timed.calls[0].error, /timed out after 5 ms/);
	const stop = new AbortController();
	setTimeout(() => stop.abort(), 50);
	const served = await runModule(ctxOf(({ event }) => (event.type === "start" ? { timers: [{ id: "x", ms: 1e300 }] } : { result: "fired" })), mod(), "go", {}, async () => ({}), { budget: { wallMs: Infinity }, signal: stop.signal });
	assert.equal(served.error, "aborted");
	assert.equal(served.result, null);
	// Infinity is only for those keys: an unlimited inFlight leaves the default
	const capped = await runModule(ctxOf(({ event }) => (event.type === "start" ? { calls: [{ id: "a", nsid: READ }, { id: "b", nsid: READ }] } : {})), mod(), "go", {}, async () => ({}), { budget: { calls: 1, inFlight: Infinity } });
	assert.match(capped.calls.find((c) => c.id === "b").error, /^Module exceeded 1 calls$/);
	// past callsPerMinute: refused like the calls cap, and the run goes on
	const rate = ({ event, state }) => {
		if (event.type === "start") return { calls: Array.from({ length: 5 }, (_, i) => ({ id: `c${i}`, nsid: READ })) };
		const seen = [...(state ?? []), event.ok ? "ok" : event.error];
		return seen.length === 5 ? { result: seen, done: true } : { state: seen };
	};
	const limited = await runModule(ctxOf(rate), mod(), "go", {}, async () => ({}), { budget: { calls: Infinity, callsPerMinute: 3 } });
	assert.equal(limited.error, undefined);
	assert.deepEqual(limited.result.filter((r) => r === "ok").length, 3);
	assert.deepEqual(limited.result.filter((r) => r !== "ok"), ["Module exceeded 3 calls per minute", "Module exceeded 3 calls per minute"]);
});

test("a full queue drops the oldest messages and counts them on the next one", async () => {
	const feed = channel();
	const seen = [];
	let release;
	const gate = new Promise((r) => (release = r));
	const ctx = {
		wasm: {
			run: async ({ input: { event } }) => {
				if (event.type === "start") return { calls: [{ id: "f", nsid: SUB, subscribe: true }] };
				if (event.type === "message") {
					await gate; // the handler is slow, so messages pile up
					seen.push([event.data, event.dropped]);
				}
				return {};
			},
		},
	};
	const running = runModule(ctx, mod(), "go", {}, async () => feed, { budget: { queue: 3 } });
	await sleep(5);
	for (let i = 0; i < 8; i++) feed.push(i);
	await sleep(5);
	feed.end();
	await sleep(5);
	release();
	await running;
	// 0 is being handled; 1..7 race for 3 slots: 1..4 dropped, 5, 6, 7 kept; closed queues behind them.
	assert.deepEqual(seen, [[0, undefined], [5, 4], [6, undefined], [7, undefined]]);
});

test("inputs become input events and keep the run open until they end", async () => {
	async function* source() {
		yield "a";
		await sleep(5);
		yield "b";
	}
	const ran = await runModule(ctxOf(({ event, state }) => (event.type === "input" ? { state: [...(state ?? []), event.data], result: [...(state ?? []), event.data] } : {})), mod(), "go", {}, async () => ({}), { inputs: source() });
	assert.deepEqual(ran.result, ["a", "b"]);
});

test("handler errors end the run as data", async () => {
	const failed = await runModule(ctxOf(() => ({ error: "bad input" })), mod(), "go", {}, async () => ({}));
	assert.equal(failed.error, "bad input");
});

test("event-stream frames: CBOR round trip, error frames, CIDs and bytes in the atproto JSON model", () => {
	const cid = "bafyreie5737gdxlw5i64vzichcalba3z2v5n6icifvx5xytvske7mr3hpm";
	const frame = encodeCbor({ op: 1, t: "#commit" }, { seq: 7, repo: "did:plc:a", blocks: { $bytes: "AQID" }, commit: { $link: cid }, tooBig: false, ops: [] });
	assert.deepEqual(decodeFrame(frame.buffer, SUB), { $type: `${SUB}#commit`, seq: 7, repo: "did:plc:a", blocks: { $bytes: "AQID" }, commit: { $link: cid }, tooBig: false, ops: [] });
	assert.throws(() => decodeFrame(encodeCbor({ op: -1 }, { error: "FutureCursor", message: "too far" }).buffer), /FutureCursor: too far/);
	assert.deepEqual(decodeFrame('{"kind":"commit"}'), { kind: "commit" });
	assert.deepEqual(decodeCbor(encodeCbor(-500, 70000, "é", [null, true])), [-500, 70000, "é", [null, true]]);
	assert.deepEqual(decodeCbor(encodeFrame({ $type: "com.fedproxy.temp.xrpc.subscribe#response", id: 1 }, "cbor")), [{ op: 1, t: "#response" }, { id: 1 }]);
	assert.equal(encodeFrame({ a: 1 }), '{"a":1}');
	assert.deepEqual([...encodeFrame({ $bytes: "AQID" })], [1, 2, 3]);
	assert.equal(wsUrl("https://pds.example/", SUB, { cursor: "1" }), `wss://pds.example/xrpc/${SUB}?cursor=1`);
});

test("CBOR: bounded nesting, 1 MB byte strings, ints past 2^32, $type without #", () => {
	// ~1 MB of 0x81 (a one-item array, nested) errors instead of overflowing the stack
	assert.throws(() => decodeCbor(new Uint8Array(1 << 20).fill(0x81)), /nested deeper than 128/);
	const deep = (n) => Uint8Array.from([...new Array(n).fill(0x81), 0]);
	assert.equal(decodeCbor(deep(128)).length, 1);
	assert.throws(() => decodeCbor(deep(129)), /nested deeper/);
	assert.throws(() => decodeCbor(Uint8Array.from(new Array(200).fill([0xa1, 0x00]).flat())), /nested deeper/);
	assert.throws(() => decodeCbor(Uint8Array.from([0x5a, 0xff, 0xff, 0xff, 0xff])), /truncated/);
	const mb = { $bytes: Buffer.alloc(1 << 20, 7).toString("base64").replace(/=+$/, "") };
	assert.deepEqual(decodeCbor(encodeCbor(mb, "x".repeat(1 << 20))), [mb, "x".repeat(1 << 20)]);
	const big = { time: 1731000000000, neg: -5000000000 };
	assert.deepEqual(decodeCbor(encodeCbor(big)), [big]);
	const noHash = encodeFrame({ $type: SUB, x: 1 }, "cbor");
	assert.deepEqual(decodeCbor(noHash), [{ op: 1 }, { x: 1 }]);
	assert.deepEqual(decodeFrame(noHash.buffer, SUB), { $type: SUB, x: 1 });
});

test("subscribeXrpc decodes frames from a WebSocket, sends, and throws on an abnormal close", async () => {
	const sockets = [];
	class FakeSocket {
		constructor(url, options) {
			Object.assign(this, { url, options, readyState: 0, bufferedAmount: 0, sent: [] });
			sockets.push(this);
			setTimeout(() => ((this.readyState = 1), this.onopen?.()));
		}
		send(data) {
			this.sent.push(data);
		}
		close(code = 1000) {
			this.readyState = 3;
			this.onclose?.({ code });
		}
	}
	const stream = subscribeXrpc("wss://x/xrpc/a.b.c", { nsid: SUB, framing: "cbor", WebSocket: FakeSocket });
	const ws = sockets[0];
	await stream.send({ $type: "#response", ok: true });
	assert.deepEqual(decodeCbor(ws.sent[0]), [{ op: 1, t: "#response" }, { ok: true }]);
	ws.onmessage({ data: encodeCbor({ op: 1, t: "#identity" }, { seq: 1 }).buffer });
	ws.onmessage({ data: '{"kind":"account"}' });
	setTimeout(() => ws.onclose({ code: 1006, reason: "gone" }), 5);
	const got = [];
	await assert.rejects(async () => {
		for await (const m of stream) got.push(m);
	}, /closed: 1006 gone/);
	assert.deepEqual(got, [{ $type: `${SUB}#identity`, seq: 1 }, { kind: "account" }]);
	subscribeXrpc("wss://x/", { headers: { authorization: "Bearer t" }, WebSocket: FakeSocket });
	assert.deepEqual(sockets[1].options, { headers: { authorization: "Bearer t" } });
});

test("frames sent with done go out before the stream closes", async () => {
	const order = [];
	const feed = channel();
	feed.send = async (data) => (await sleep(5), order.push(`send ${data}`));
	feed.close = () => (order.push("close"), feed.end());
	const handler = ({ event }) => (event.type === "start" ? { calls: [{ id: "f", nsid: SUB, subscribe: true }], timers: [{ id: "t", ms: 1 }] } : { send: [{ id: "f", data: "bye" }], done: true });
	await runModule(ctxOf(handler), mod(), "go", {}, async () => feed);
	assert.deepEqual(order, ["send bye", "close"]);
});

test("a cancel after sends flushes them, then closes; no messages after closed", async () => {
	const order = [];
	const feed = channel();
	feed.send = async (data) => (await sleep(5), order.push(`send ${data}`));
	feed.close = () => (order.push("close"), feed.end());
	const seen = [];
	const handler = ({ event }) => {
		seen.push(event.type);
		if (event.type === "start") return { calls: [{ id: "f", nsid: SUB, subscribe: true }], send: [{ id: "f", data: "a" }], timers: [{ id: "t", ms: 1 }] };
		if (event.type === "timer" && event.id === "t") return { send: [{ id: "f", data: "b" }], cancel: ["f"], timers: [{ id: "after", ms: 30 }] };
		return {};
	};
	const running = runModule(ctxOf(handler), mod(), "go", {}, async () => feed);
	await sleep(8);
	feed.push("late"); // arrives while the cancel is flushing
	assert.equal((await running).error, undefined);
	assert.deepEqual(order, ["send a", "send b", "close"]);
	assert.ok(seen.indexOf("closed") > -1 && !seen.slice(seen.indexOf("closed")).includes("message"));
});

test("overflow close: a full queue closes the subscription instead of dropping its messages", async () => {
	const feed = channel();
	const other = channel();
	const seen = [];
	let release, entered;
	const gate = new Promise((r) => (release = r));
	const inside = new Promise((r) => (entered = r));
	const ctx = {
		wasm: {
			run: async ({ input: { event } }) => {
				if (event.type === "start") return { calls: [{ id: "req", nsid: SUB, subscribe: true, overflow: "close" }, { id: "fire", nsid: SUB, subscribe: true }] };
				if (event.type === "message" && event.data === "r0") {
					entered();
					await gate;
				}
				seen.push(event.type === "message" ? `${event.id}:${event.data}${event.dropped ? `+${event.dropped}` : ""}` : `${event.type}:${event.id}:${event.error ?? ""}`);
				return {};
			},
		},
	};
	let n = 0;
	const running = runModule(ctx, mod(), "go", {}, async () => (n++ ? other : feed), { budget: { queue: 3 } });
	feed.push("r0");
	await inside; // r0 is being handled while the rest pile up
	feed.push("r1");
	other.push("f1");
	other.push("f2");
	other.push("f3"); // full: drops f1 (a drop subscription), never r1
	// all four queued before r2 (under swap.mjs each subscription reaches the queue on its own task, so the
	// order across the two is not kept: only each one's own order is asserted)
	await sleep(20);
	feed.push("r2"); // full again: req closes rather than lose r2
	await sleep(20);
	other.end();
	release();
	const ran = await running;
	assert.equal(feed.closed, true);
	const of = (id) => seen.filter((s) => s.split(":").includes(id));
	assert.deepEqual(of("req"), ["req:r0", "req:r1", "closed:req:overflow"]);
	assert.deepEqual(of("fire"), ["fire:f2+1", "fire:f3", "closed:fire:"]);
	assert.deepEqual(ran.calls.find((c) => c.id === "req"), { id: "req", nsid: SUB, subscribe: true, messages: 3, ok: false, error: "overflow" });
	await assert.doesNotReject(async () => assert.match((await runModule(ctxOf(({ event }) => (event.type === "start" ? { calls: [{ id: "x", nsid: SUB, subscribe: true, overflow: "maybe" }] } : { result: event.error })), mod(), "go", {}, async () => feed)).result, /overflow must be drop or close/));
});

test("budget.log keeps the last call records; the result counts the rest", async () => {
	const handler = ({ event }) => (event.type === "start" ? { calls: [0, 1, 2, 3, 4].map((n) => ({ id: `c${n}`, nsid: READ })) } : {});
	const ran = await runModule(ctxOf(handler), mod(), "go", {}, async () => ({}), { budget: { log: 2 } });
	assert.deepEqual(ran.calls.map((c) => c.id), ["c3", "c4"]);
	assert.equal(ran.callsDropped, 3);
	const all = await runModule(ctxOf(handler), mod(), "go", {}, async () => ({}));
	assert.equal(all.calls.length, 5);
	assert.equal(all.callsDropped, undefined);
});

test("subscribe serviceAuth reaches call(); input only for host pseudo-call sessions; bad shapes are refused", async () => {
	const IROH = "com.fedfork.exec.iroh.connect";
	const requests = [];
	const errors = {};
	const handler = ({ event }) => {
		if (event.type === "start") {
			return {
				calls: [
					{ id: "hdr", nsid: SUB, service: "https://relay.example", subscribe: true, serviceAuth: {} },
					{ id: "qry", nsid: SUB, service: "https://relay.example", subscribe: true, serviceAuth: { param: "service_auth" } },
					{ id: "iroh", nsid: IROH, input: { ticket: "abc", command: "id" }, subscribe: true },
					{ id: "in", nsid: SUB, input: { x: 1 }, subscribe: true },
					{ id: "nosvc", nsid: SUB, subscribe: true, serviceAuth: {} },
					{ id: "odd", nsid: SUB, service: "https://relay.example", subscribe: true, serviceAuth: { param: "a b" } },
					{ id: "extra", nsid: SUB, service: "https://relay.example", subscribe: true, serviceAuth: { header: "x" } },
				],
			};
		}
		if (event.type === "result" && !event.ok) errors[event.id] = event.error;
		return {};
	};
	await runModule(ctxOf(handler), mod({ calls: [SUB, IROH] }), "go", {}, async (nsid, request) => (requests.push([nsid, request]), []), { budget: { subscriptions: 10 } });
	const byNsid = (n, key) => requests.filter(([nsid]) => nsid === n).map(([, r]) => r[key]);
	assert.deepEqual(byNsid(SUB, "serviceAuth"), [{}, { param: "service_auth" }]);
	assert.deepEqual(byNsid(IROH, "input"), [{ ticket: "abc", command: "id" }]);
	assert.match(errors.in, /subscribe takes params, not input/);
	assert.match(errors.nosvc, /serviceAuth needs an https service/);
	assert.match(errors.odd, /serviceAuth must be \{\} or \{ param \}/);
	assert.match(errors.extra, /serviceAuth must be \{\} or \{ param \}/);
});

test("withServiceAuth: a header where the WebSocket takes one, else the token in the named query param", async () => {
	const { withServiceAuth } = await import("./index.js");
	const mint = async (service, nsid) => `tok:${new URL(service).host}:${nsid}`;
	const req = { subscribe: true, service: "https://relay.example", params: { cursor: "1" } };
	assert.deepEqual(await withServiceAuth(SUB, { ...req, serviceAuth: {} }, mint), { ...req, headers: { authorization: `Bearer tok:relay.example:${SUB}` } });
	assert.deepEqual(await withServiceAuth(SUB, { ...req, serviceAuth: { param: "service_auth" } }, mint, { canHeader: false }), { ...req, params: { cursor: "1", service_auth: `tok:relay.example:${SUB}` } });
	assert.deepEqual(await withServiceAuth(SUB, req, mint, { canHeader: false }), req, "no serviceAuth: untouched, nothing minted");
	await assert.rejects(withServiceAuth(SUB, { ...req, serviceAuth: {} }, mint, { canHeader: false }), /browsers cannot set WebSocket headers; serviceAuth needs \{ param \}/);
	await assert.rejects(withServiceAuth(SUB, { ...req, headers: { a: "b" } }, mint, { canHeader: false }), /browsers cannot set WebSocket headers/);
});

// review 3b #1: what a cancel or a re-armed timer replaces must not arrive later, even when its event
// is already queued (the handler is slow, so events queue behind it).
test("cancel drops a queued result and queued messages; a re-armed timer fires once", async () => {
	const slowAfter = (handler, id) => ({ wasm: { run: async ({ input }) => {
		const out = handler(input);
		if (input.event.id === id && input.event.type !== "start") await sleep(80);
		return out;
	} } });
	const late = async (nsid, { params }) => (await sleep(Number(params.ms)), {});
	const seen = [];
	const note = ({ event }) => seen.push(`${event.type}:${event.id ?? ""}`);
	let ran = await runModule(slowAfter((input) => {
		note(input);
		if (input.event.type === "start") return { calls: [{ id: "a", nsid: READ, params: { ms: 5 } }, { id: "b", nsid: READ, params: { ms: 30 } }] };
		if (input.event.id === "a") return { cancel: ["b"] };
		return {};
	}, "a"), mod(), "go", {}, late);
	assert.deepEqual(seen, ["start:", "result:a"], "b's result was queued behind a; the cancel drops it");
	assert.equal(ran.error, undefined);
	seen.length = 0;
	await runModule(slowAfter((input) => {
		note(input);
		if (input.event.type === "start") return { calls: [{ id: "a", nsid: READ, params: { ms: 5 } }], timers: [{ id: "t", ms: 30 }] };
		if (input.event.id === "a") return { timers: [{ id: "t", ms: 100 }] };
		return {};
	}, "a"), mod(), "go", {}, late);
	assert.deepEqual(seen, ["start:", "result:a", "timer:t"], "t fired while a ran; re-armed, only the new firing arrives");
	seen.length = 0;
	ran = await runModule({ wasm: { run: async ({ input: { event } }) => {
		seen.push(`${event.type}:${event.id ?? ""}${event.data ? `:${event.data.n}` : ""}`);
		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 {};
	} } }, mod(), "go", {}, async () => ({ async *[Symbol.asyncIterator]() {
		for (let n = 0; n < 3; n++) yield { n }, await sleep(10);
		await sleep(1e4);
	} }));
	assert.deepEqual(seen, ["start:", "message:s:0", "closed:s"]);
});

// review 3b #4: a ledger read that fails in the rkey guard refuses that call; the run does not throw.
test("a failing ledger read refuses the write instead of throwing", async () => {
	const PUT = "com.atproto.repo.putRecord";
	const storage = { managed_records: { get: async () => { throw new Error("database is locked"); }, put: async () => {}, delete: async () => {} } };
	const ran = await runModule(ctxOf(({ event }) => (event.type === "start" ? { calls: [{ id: "p", nsid: PUT, input: { repo: "did:plc:a", collection: "x.y.z", rkey: "k", record: {} } }] } : { result: event, done: true }), storage), mod({ calls: [PUT] }), "go", {}, async () => ({}));
	assert.equal(ran.error, undefined);
	assert.equal(ran.result.ok, false);
	assert.match(ran.result.error, /database is locked/);
});

// review 3c: map keys sort by their UTF-8 length, then bytewise (subscribe.js sorted UTF-16 lengths).
test("encodeCbor orders non-ASCII map keys by encoded length", () => {
	// "é" is 1 UTF-16 unit but 2 bytes: z (1 byte), then ab and é (2 bytes; 0x61 < 0xc3).
	assert.deepEqual([...encodeCbor({ é: 1, ab: 2, z: 3 })], [0xa3, 0x61, 0x7a, 0x03, 0x62, 0x61, 0x62, 0x02, 0x62, 0xc3, 0xa9, 0x01]);
	assert.deepEqual(decodeCbor(encodeCbor({ é: 1, ab: 2, z: 3 })), [{ z: 3, ab: 2, é: 1 }]);
});

test("http.fetch: declared ∩ host-allowed origins, host headers injected and replacing the module's, refusals are call errors", async () => {
	const HTTP = "com.fedfork.exec.http.fetch";
	const seen = [];
	const fetch = async (url, init) => (seen.push([url, init.method, init.headers, init.body ?? null, init.redirect]), new Response(`hi ${init.method}`, { status: 201, headers: { "content-type": "text/plain", "mcp-session-id": "s1" } }));
	const req = (id, input) => ({ id, nsid: HTTP, input });
	const results = {};
	const handler = ({ event }) => {
		if (event.type === "start") {
			return {
				calls: [
					req("ok", { method: "POST", url: "https://api.example/v1/messages?x=1", headers: { "content-type": "application/json", "X-Api-Key": "module-guess" }, body: "{}" }),
					req("get", { url: "https://Mcp.Example:443/mcp" }),
					req("undeclared", { url: "https://other.example/" }),
					req("unallowed", { url: "https://declared.example/" }),
					req("creds", { url: "https://u:p@api.example/" }),
					req("ftp", { url: "ftp://api.example/" }),
					req("method", { method: "TRACE", url: "https://api.example/" }),
					req("hdrs", { url: "https://api.example/", headers: { "bad name": "x" } }),
					req("body", { method: "POST", url: "https://api.example/", body: { a: 1 } }),
					req("getbody", { url: "https://api.example/", body: "x" }),
					req("noinput", undefined),
					{ id: "sub", nsid: HTTP, input: { url: "https://api.example/" }, subscribe: true },
				],
			};
		}
		results[event.id] = event.ok ? event.output : event.error;
		return { result: results };
	};
	const ctx = { ...ctxOf(handler), http: { fetch } };
	const http = { "https://api.example": { headers: { "x-api-key": "SECRET" } }, "https://mcp.example": {}, "https://unlisted.example": {} };
	const ran = await runModule(ctx, mod({ calls: [HTTP], http: ["https://api.example", "https://mcp.example", "https://declared.example"] }), "go", {}, async () => assert.fail("never the XRPC call"), { http });
	assert.equal(ran.error, undefined);
	const r = ran.result;
	assert.deepEqual(r.ok, { status: 201, headers: { "content-type": "text/plain", "mcp-session-id": "s1" }, body: "hi POST" });
	assert.equal(r.get.body, "hi GET");
	assert.deepEqual(seen.sort(), [
		["https://Mcp.Example:443/mcp", "GET", {}, null, "manual"],
		// a JSON body goes as its JSON
		["https://api.example/", "POST", { "x-api-key": "SECRET" }, '{"a":1}', "manual"],
		["https://api.example/v1/messages?x=1", "POST", { "content-type": "application/json", "x-api-key": "SECRET" }, "{}", "manual"],
	]);
	const P = `${HTTP}: `;
	assert.equal(r.undeclared, `${P}https://other.example is not in the module's http list`);
	assert.equal(r.unallowed, `${P}this host does not allow https://declared.example`);
	assert.equal(r.creds, `${P}url must be a URL without credentials`);
	assert.equal(r.ftp, `${P}ftp://* is not in the module's http list`);
	assert.equal(r.method, `${P}method must be one of GET, HEAD, POST, PUT, PATCH, DELETE`);
	assert.equal(r.hdrs, `${P}headers must map header names to strings`);
	assert.equal(r.getbody, `${P}GET takes no body`);
	assert.equal(r.noinput, `${P}input must be { method?, url, headers?, body? }`);
	assert.equal(r.sub, `${P}cannot be subscribed to (a response arrives whole)`);

	// "*" reaches any origin the host allows, and none without the host's allowlist
	const star = await runModule(ctx, mod({ calls: [HTTP], http: ["*"] }), "go", {}, async () => {}, { http: { "https://mcp.example": {} } });
	assert.equal(star.result.get.body, "hi GET");
	assert.equal(star.result.ok, `${P}this host does not allow https://api.example`);
	const none = await runModule(ctx, mod({ calls: [HTTP], http: ["*"] }), "go", {}, async () => {});
	assert.equal(none.result.get, `${P}this host does not allow https://mcp.example`);
	const undeclared = await runModule(ctx, mod({ calls: [HTTP] }), "go", {}, async () => {}, { http });
	assert.equal(undeclared.result.get, `${P}https://mcp.example is not in the module's http list`);
});

test("http.fetch: another scheme goes to the host's fetcher for it, allowed as <scheme>://* in both lists, without host headers", async () => {
	const HTTP = "com.fedfork.exec.http.fetch";
	const seen = [];
	// a stub transport fetcher (atlibs-webrtc's host.fetchers()): what it got, and a response with mixed-case headers
	const fetchers = { "webrtc:": async ({ method, url, headers, body, signal }) => (seen.push([method, url, headers, body ?? null, signal instanceof AbortSignal]), { status: 200, headers: { "Content-Type": "application/json", "Mcp-Session-Id": "s1" }, body: `{"via":"webrtc"}` }) };
	const url = "webrtc:///eyJyZWNlaXB0Ijp7fX0/shell/mcp";
	const req = (id, input) => ({ id, nsid: HTTP, input });
	const results = {};
	const handler = ({ event }) => {
		if (event.type === "start") {
			return {
				calls: [
					req("post", { method: "POST", url, headers: { "content-type": "application/json", "x-api-key": "module's own" }, body: "{}" }),
					req("iroh", { url: "iroh:///e30/shell/mcp" }),
					req("creds", { url: "webrtc://u:p@x/y" }),
					req("https", { url: "https://api.example/" }),
				],
			};
		}
		results[event.id] = event.ok ? event.output : event.error;
		return { result: results };
	};
	const ctx = { ...ctxOf(handler), http: { fetch: async () => assert.fail("never ctx.http.fetch for webrtc:") } };
	// the host's headers and serviceAuth never ride a scheme fetch
	const http = { "webrtc://*": { headers: { "x-api-key": "SECRET" } }, "iroh://*": {} };
	const run = (declared, options) => runModule(ctx, mod({ calls: [HTTP], http: declared }), "go", {}, async () => assert.fail("never the XRPC call"), options);
	const ran = await run(["webrtc://*", "iroh://*"], { http, fetchers });
	assert.equal(ran.error, undefined);
	const P = `${HTTP}: `;
	assert.deepEqual(ran.result.post, { status: 200, headers: { "content-type": "application/json", "mcp-session-id": "s1" }, body: `{"via":"webrtc"}` });
	assert.deepEqual(seen, [["POST", url, { "content-type": "application/json", "x-api-key": "module's own" }, "{}", true]]);
	assert.equal(ran.result.iroh, `${P}no fetcher for iroh:`);
	assert.equal(ran.result.creds, `${P}url must be a URL without credentials`);
	assert.equal(ran.result.https, `${P}https://api.example is not in the module's http list`);
	// a module's "*" is http(s) only; a scheme the host does not allow is refused before any fetcher
	const star = await run(["*", "webrtc://host.example"], { http, fetchers });
	assert.equal(star.result.post, `${P}webrtc://* is not in the module's http list`);
	const unallowed = await run(["webrtc://*"], { http: { "iroh://*": {} }, fetchers });
	assert.equal(unallowed.result.post, `${P}this host does not allow webrtc://*`);
	assert.equal(seen.length, 1);
});

test("a call's service may be a did#service (a public read of that service); bad ones and input are refused", async () => {
	const seen = [];
	const errors = {};
	const handler = ({ event }) => {
		if (event.type === "start") {
			return {
				calls: [
					{ id: "ok", nsid: READ, service: "did:web:api.example%3A8443#svc_1", params: { a: "1" } },
					{ id: "empty", nsid: READ, service: "did:web:api.example#" },
					{ id: "odd", nsid: READ, service: "did:web:api.example#a b" },
					{ id: "input", nsid: READ, service: "did:web:api.example#svc", input: {} },
				],
			};
		}
		if (!event.ok) errors[event.id] = event.error;
		return {};
	};
	await runModule(ctxOf(handler), mod(), "go", {}, async (nsid, request) => (seen.push(request.service), {}));
	assert.deepEqual(seen, ["did:web:api.example%3A8443#svc_1"]);
	assert.deepEqual(errors, {
		empty: "Module call service must be an https origin, a DID or a did#service",
		odd: "Module call service must be an https origin, a DID or a did#service",
		input: "Module call to a DID service is a read: no input",
	});
});

test("a did#service call may name its endpoints (then it takes input); they reach call() as given; bad ones are refused", async () => {
	const seen = [];
	const errors = {};
	const endpoints = [{ kind: "iroh", address: "endpointabc234" }, { kind: "https", address: "https://bidder.example" }];
	const handler = ({ event }) => {
		if (event.type === "start") {
			return {
				calls: [
					{ id: "ok", nsid: READ, service: "did:plc:bidder#pdr_temp_market", endpoints, input: { a: 1 } },
					{ id: "bare", nsid: READ, service: "did:plc:bidder", endpoints },
					{ id: "https", nsid: READ, service: "https://bidder.example", endpoints },
					{ id: "none", nsid: READ, service: "did:plc:bidder#pdr_temp_market", endpoints: [] },
					{ id: "extra", nsid: READ, service: "did:plc:bidder#pdr_temp_market", endpoints: [{ kind: "iroh", address: "e", x: 1 }] },
					{ id: "kind", nsid: READ, service: "did:plc:bidder#pdr_temp_market", endpoints: [{ kind: "Iroh!", address: "e" }] },
					{ id: "space", nsid: READ, service: "did:plc:bidder#pdr_temp_market", endpoints: [{ kind: "iroh", address: "a b" }] },
				],
			};
		}
		if (!event.ok) errors[event.id] = event.error;
		return {};
	};
	await runModule(ctxOf(handler), mod(), "go", {}, async (nsid, request) => (seen.push([request.service, request.endpoints, request.input]), {}));
	assert.deepEqual(seen, [["did:plc:bidder#pdr_temp_market", endpoints, { a: 1 }]]);
	const bad = "Module call endpoints must be 1 to 8 { kind, address } (kind a URL scheme name, address up to 2048 printable ASCII)";
	assert.deepEqual(errors, {
		bare: "Module call endpoints need a did#service service",
		https: "Module call endpoints need a did#service service",
		none: bad,
		extra: bad,
		kind: bad,
		space: bad,
	});
});

test("ioBytes: the host's per-event wasm I/O cap reaches ctx.wasm.run, 1 MiB by default, never past 16 MiB", async () => {
	for (const [budget, want] of [[undefined, 1 << 20], [{ ioBytes: 8 << 20 }, 8 << 20], [{ ioBytes: 1 << 30 }, 16 << 20], [{ ioBytes: Infinity }, 1 << 20]]) {
		const seen = [];
		const ctx = { wasm: { run: async ({ ioBytes }) => (seen.push(ioBytes), { done: true }) } };
		await runModule(ctx, mod(), "go", {}, async () => ({}), budget && { budget });
		assert.deepEqual(seen, [want], JSON.stringify(budget));
	}
});

test("a call's encoding names the content-type of its { $bytes } input and reaches call(); bad ones are refused", async () => {
	const seen = [];
	const errors = {};
	const handler = ({ event }) => {
		if (event.type === "start") {
			return {
				calls: [
					{ id: "ok", nsid: READ, input: { $bytes: "AAEC" }, encoding: "application/x-git-packed-objects" },
					{ id: "plain", nsid: READ, input: { $bytes: "AAEC" } },
					{ id: "odd", nsid: READ, input: { $bytes: "AAEC" }, encoding: "not a type" },
					{ id: "json", nsid: READ, input: { a: 1 }, encoding: "application/json" },
				],
			};
		}
		if (!event.ok) errors[event.id] = event.error;
		return {};
	};
	await runModule(ctxOf(handler), mod(), "go", {}, async (nsid, request) => (seen.push([request.input, request.encoding]), {}));
	assert.deepEqual(seen, [[{ $bytes: "AAEC" }, "application/x-git-packed-objects"], [{ $bytes: "AAEC" }, undefined]]);
	assert.deepEqual(errors, { odd: "Module call encoding must be a MIME type", json: "Module call encoding needs a $bytes input" });
});

test("held values: put, append, get, delete in order at plan time; spliced into an http.fetch body and the result; bounded by heldBytes", async () => {
	const H = "com.fedfork.exec.held.";
	const HTTP = "com.fedfork.exec.http.fetch";
	const errors = {};
	const outputs = {};
	const sent = [];
	const handler = ({ event }) => {
		if (event.type === "start") {
			return {
				calls: [
					{ id: "a1", nsid: `${H}append`, input: { name: "t", items: [{ role: "user", content: "hi" }] } },
					{ id: "a2", nsid: `${H}append`, input: { name: "t", items: [{ role: "assistant", content: "yo" }] } },
					{ id: "send", nsid: HTTP, input: { method: "POST", url: "https://api.example/v1", body: { model: "m", messages: { $held: "t" } } } },
					{ id: "p", nsid: `${H}put`, input: { name: "n", value: 7 } },
					{ id: "g", nsid: `${H}get`, input: { name: "n" } },
					{ id: "d", nsid: `${H}delete`, input: { name: "n" } },
					{ id: "g2", nsid: `${H}get`, input: { name: "n" } },
					{ id: "big", nsid: `${H}put`, input: { name: "b", value: "x".repeat(200) } },
					{ id: "notlist", nsid: `${H}append`, input: { name: "t2", items: 1 } },
					{ id: "missing", nsid: HTTP, input: { method: "POST", url: "https://api.example/v1", body: { m: { $held: "nope" } } } },
					{ id: "odd", nsid: `${H}grow`, input: { name: "t" } },
				],
				result: { messages: { $held: "t" } },
			};
		}
		if (event.ok) outputs[event.id] = event.output;
		else errors[event.id] = event.error;
		return {};
	};
	const ctx = { ...ctxOf(handler), http: { fetch: async (url, init) => (sent.push(init.body), new Response("{}")) } };
	const calls = [`${H}put`, `${H}append`, `${H}get`, `${H}delete`, `${H}grow`, HTTP];
	const ran = await runModule(ctx, mod({ calls, http: ["https://api.example"] }), "go", {}, async () => ({}), { http: { "https://api.example": {} }, budget: { heldBytes: 150 } });
	assert.equal(ran.error, undefined);
	assert.deepEqual(sent, ['{"model":"m","messages":[{"role":"user","content":"hi"},{"role":"assistant","content":"yo"}]}']);
	assert.deepEqual(ran.result, { messages: [{ role: "user", content: "hi" }, { role: "assistant", content: "yo" }] });
	assert.deepEqual({ a1: outputs.a1, a2: outputs.a2, p: outputs.p, g: outputs.g, d: outputs.d, g2: outputs.g2 }, { a1: { length: 1, bytes: 32 }, a2: { length: 2, bytes: 68 }, p: {}, g: { value: 7 }, d: {}, g2: { value: null } });
	assert.deepEqual(errors, {
		big: "com.fedfork.exec.held.put: held values past 150 bytes",
		notlist: "com.fedfork.exec.held.append: items must be a list",
		missing: "com.fedfork.exec.http.fetch: no held value nope",
		odd: "com.fedfork.exec.held.grow is not a held-value call (put, append, get, delete)",
	});
});

test("held values are copies: a later append grows neither the module's put value (onEvent's output) nor an earlier get's answer", async () => {
	const H = "com.fedfork.exec.held.";
	const outputs = {};
	let seen;
	const handler = ({ event }) => {
		if (event.type === "start") {
			return {
				calls: [
					{ id: "p", nsid: `${H}put`, input: { name: "t", value: [{ n: 1 }] } },
					{ id: "g1", nsid: `${H}get`, input: { name: "t" } },
					{ id: "a", nsid: `${H}append`, input: { name: "t", items: [{ n: 2 }] } },
					{ id: "g2", nsid: `${H}get`, input: { name: "t" } },
					{ id: "a2", nsid: `${H}append`, input: { name: "t", items: [{ n: 3 }] } },
				],
			};
		}
		outputs[event.id] = event.output;
		return {};
	};
	const onEvent = (event, output) => event.type === "start" && (seen = output);
	const ran = await runModule(ctxOf(handler), mod({ calls: [`${H}put`, `${H}get`, `${H}append`] }), "go", {}, async () => ({}), { onEvent });
	assert.equal(ran.error, undefined);
	assert.deepEqual(outputs.g1, { value: [{ n: 1 }] });
	assert.deepEqual(outputs.g2, { value: [{ n: 1 }, { n: 2 }] });
	assert.deepEqual(seen.calls[0].input.value, [{ n: 1 }]);
	assert.deepEqual(seen.calls[2].input.items, [{ n: 2 }]);
});

test("a host starts a run with held values (counted against heldBytes); { $held: [names] } concatenates held lists", async () => {
	const H = "com.fedfork.exec.held.";
	const HTTP = "com.fedfork.exec.http.fetch";
	const sent = [];
	const errors = {};
	const handler = ({ event }) => {
		if (event.type === "start") {
			return {
				calls: [
					{ id: "a", nsid: `${H}append`, input: { name: "t", items: [{ n: 2 }] } },
					{ id: "send", nsid: HTTP, input: { method: "POST", url: "https://api.example/v1", body: { messages: { $held: ["s", "t"] } } } },
					{ id: "notlist", nsid: HTTP, input: { method: "POST", url: "https://api.example/v1", body: { m: { $held: ["s", "n"] } } } },
					{ id: "missing", nsid: HTTP, input: { method: "POST", url: "https://api.example/v1", body: { m: { $held: ["nope"] } } } },
				],
				result: { t: { $held: "t" }, n: { $held: "n" } },
			};
		}
		if (!event.ok) errors[event.id] = event.error;
		return {};
	};
	const ctx = { ...ctxOf(handler), http: { fetch: async (url, init) => (sent.push(init.body), new Response("{}")) } };
	const m = mod({ calls: [`${H}append`, HTTP], http: ["https://api.example"] });
	const held = { s: [{ n: 0 }], t: [{ n: 1 }], n: 7 };
	const ran = await runModule(ctx, m, "go", {}, async () => ({}), { http: { "https://api.example": {} }, held });
	assert.equal(ran.error, undefined);
	assert.deepEqual(ran.result, { t: [{ n: 1 }, { n: 2 }], n: 7 });
	assert.deepEqual(held.t, [{ n: 1 }], "the host's value is copied, not grown");
	assert.deepEqual(sent, ['{"messages":[{"n":0},{"n":1},{"n":2}]}']);
	assert.deepEqual(errors, { notlist: `${HTTP}: held value n is not a list`, missing: `${HTTP}: no held value nope` });
	const big = await runModule(ctx, m, "go", {}, async () => ({}), { held: { t: "x".repeat(50) }, budget: { heldBytes: 20 } });
	assert.equal(big.error, "held: com.fedfork.exec.held.put: held values past 20 bytes");
	assert.equal((await runModule(ctx, m, "go", {}, async () => ({}), { held: { "bad name": 1 } })).error, "held: com.fedfork.exec.held.put: input needs a name");
});

test("http.fetch serviceAuth: the host mints a token per request with the run's session; no session or no token is a call error", async () => {
	const HTTP = "com.fedfork.exec.http.fetch";
	const sent = [];
	const handler = ({ event, state }) => {
		if (event.type === "start") {
			const req = (id) => ({ id, nsid: HTTP, input: { method: "POST", url: "https://ai.example/v1/chat", headers: { Authorization: "Bearer module-guess", "content-type": "application/json" }, body: "{}" } });
			return { calls: [req("a"), req("b")] };
		}
		const seen = { ...state, [event.id]: event.ok ? event.output.body : event.error };
		return { result: seen, state: seen };
	};
	const ctx = { ...ctxOf(handler), http: { fetch: async (url, init) => (sent.push(init.headers), new Response("ok")) } };
	const m = mod({ calls: [HTTP], http: ["https://ai.example"] });
	const http = { "https://ai.example": { headers: { "x-extra": "1" }, serviceAuth: { aud: "did:web:console.ai.example", lxm: "example.ai.dispatch" } } };
	const minted = [];
	const session = async (nsid, request) => (minted.push([nsid, request.params]), { token: `tok-${minted.length}` });
	const ran = await runModule(ctx, m, "go", {}, session, { http });
	assert.equal(ran.error, undefined);
	assert.deepEqual(minted, [["com.atproto.server.getServiceAuth", { aud: "did:web:console.ai.example", lxm: "example.ai.dispatch" }], ["com.atproto.server.getServiceAuth", { aud: "did:web:console.ai.example", lxm: "example.ai.dispatch" }]]);
	assert.deepEqual(sent.map((h) => h.authorization).sort(), ["Bearer tok-1", "Bearer tok-2"]);
	assert.ok(sent.every((h) => h["x-extra"] === "1" && h["content-type"] === "application/json" && !("Authorization" in h)));
	const signedOut = await runModule(ctx, m, "go", {}, async () => Promise.reject(new Error("no session")), { http });
	assert.equal(signedOut.result.a, `${HTTP}: https://ai.example needs a service-auth token (aud did:web:console.ai.example, lxm example.ai.dispatch) from the run's session: no session (an OAuth session's scope must include rpc:example.ai.dispatch?aud=did:web:console.ai.example)`);
	const empty = await runModule(ctx, m, "go", {}, async () => ({}), { http });
	assert.equal(empty.result.a, `${HTTP}: com.atproto.server.getServiceAuth answered no service-auth token`);
});