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

import { loadComposition, runCompose } from "../../../atlibs/javascript/src/compose/index.js";
import { runModule } from "../../../atlibs/javascript/src/run-module/index.js";
import { load } from "../../../atlibs/rust/src/harness.mjs";
import { fetcher, mcpServer, noFetchers, streamableHttp } from "./mcp-stub.mjs";

const { module, run } = await load(new URL("../../target/wasm32-unknown-unknown/release/asyncai.wasm", import.meta.url));
const manifest = JSON.parse(await readFile(new URL("module.json", import.meta.url), "utf8"));
const mod = { module, ...manifest };
const MCP = "com.fedfork.exec.mcp.message";
const INFER = "https://inference.test/v1/messages";
const CHAT = "https://inference.test/v1/chat/completions";
const UNLIMITED = { calls: Infinity, events: Infinity, wallMs: 30_000 };

// A scripted Messages API: each request gets the next answer (an object: a 200 JSON body), recorded with its headers.
function inference(answers) {
	const requests = [];
	const respond = async (init) => {
		requests.push({ headers: init.headers, body: JSON.parse(init.body) });
		const next = answers.shift();
		if (next instanceof Response) return next;
		return Response.json(typeof next === "function" ? next(requests.at(-1).body) : next);
	};
	return { requests, respond };
}
const text = (t) => ({ content: [{ type: "text", text: t }], stop_reason: "end_turn", usage: { input_tokens: 10, output_tokens: 2 } });
const uses = (...calls) => ({ content: calls.map(([id, name, input]) => ({ type: "tool_use", id, name, input })), stop_reason: "tool_use", usage: { input_tokens: 5, output_tokens: 5 } });

// ctx: the wasm, and a fetch that answers the inference endpoint and any stub MCP origins.
const ctxOf = (infer, origins = {}) => ({
	wasm: { run: async ({ entry, input }) => run(entry, input) },
	http: {
		fetch: async (url, init) => {
			if (url === INFER || url === CHAT) return infer.respond(init);
			const server = origins[new URL(url).origin];
			if (!server) throw new Error(`no stub for ${url}`);
			return server(init);
		},
	},
});
const noScheme = await noFetchers(runModule, { wasm: { run: async ({ entry, input }) => run(entry, input) } }, mod);
const allow = (...origins) => Object.fromEntries([["https://inference.test", { headers: { "x-api-key": "HOST-KEY" } }], ...origins.map((o) => [o, {}])]);

test("module has no imports and exports run", () => {
	assert.equal(WebAssembly.Module.imports(module).length, 0);
	assert.ok(WebAssembly.Module.exports(module).some((e) => e.name === "run"));
});

test("remote (Streamable HTTP, SSE) and two in-process servers: namespaced tools, tool results, end_turn", async () => {
	const docs = streamableHttp(mcpServer({ "search.docs": (args) => ({ content: [{ type: "text", text: `found ${args.q}` }] }), "list.more": () => ({ content: [] }) }, { version: "2025-06-18", pageSize: 1 }), { sse: true, session: "sess-1" });
	const code = mcpServer({ add: ({ a, b }) => ({ content: [{ type: "text", text: String(a + b) }], structuredContent: { sum: a + b } }) });
	const kcp = mcpServer({ add: () => ({ content: [{ type: "text", text: "kcp add" }], isError: true }) });
	const infer = inference([
		uses(["t1", "docs__search_docs", { q: "mcp" }], ["t2", "code__add", { a: 2, b: 3 }], ["t3", "kcp__add", {}], ["t4", "nope__x", {}]),
		text("2 + 3 = 5"),
	]);
	const ctx = ctxOf(infer, { "https://mcp.test": docs.handle });
	const composition = await loadComposition(ctx, {
		$type: "com.fedfork.exec.composition",
		name: "agent",
		instances: [{ name: "agent", module: "asyncai" }],
		handlers: [{ name: "code" }, { name: "kcp" }],
		edges: [
			{ from: "agent", nsid: MCP, handler: "code", reason: "in-process code server" },
			{ from: "agent", nsid: "com.example.mcp.kcp", handler: "kcp", reason: "a second in-process server (anyHooks)" },
		],
	}, { load: async () => mod });
	const mcp = [{ name: "docs", url: "https://mcp.test/mcp" }, { name: "code", hook: MCP }, { name: "kcp", hook: "com.example.mcp.kcp" }];
	const ran = await runCompose(ctx, composition, "agent", "run", { prompt: "add 2 and 3", endpoint: "https://inference.test", model: "test-model", mcp: JSON.stringify(mcp) }, {
		handlers: { code: (message) => code(message), kcp: (message) => kcp(message) },
		http: allow("https://mcp.test"),
		budget: UNLIMITED,
	});
	assert.equal(ran.error, undefined);
	const r = ran.result;
	assert.equal(r.stop, "end_turn");
	assert.equal(r.text, "2 + 3 = 5");
	assert.equal(r.steps, 2);
	assert.deepEqual(r.usage, { input_tokens: 15, output_tokens: 7 });
	assert.deepEqual(r.tools, ["docs__search_docs", "docs__list_more", "code__add", "kcp__add"]);
	assert.deepEqual(r.servers.map((s) => [s.name, s.phase, s.protocolVersion]), [["docs", "ready", "2025-06-18"], ["code", "ready", "2025-11-25"], ["kcp", "ready", "2025-11-25"]]);

	// the host's key went with every inference request; the wasm set only its own headers
	const [first, second] = infer.requests;
	assert.equal(first.headers["x-api-key"], "HOST-KEY");
	assert.equal(first.headers["anthropic-version"], "2026-10-21");
	assert.equal(first.body.model, "test-model");
	assert.deepEqual(first.body.tools[2], { name: "code__add", input_schema: { type: "object" }, description: "add" });
	assert.deepEqual(first.body.messages, [{ role: "user", content: "add 2 and 3" }]);
	assert.deepEqual(second.body.messages.at(-1), {
		role: "user",
		content: [
			{ type: "tool_result", tool_use_id: "t1", content: "found mcp" },
			{ type: "tool_result", tool_use_id: "t2", content: '{"sum":5}' },
			{ type: "tool_result", tool_use_id: "t3", content: "kcp add", is_error: true },
			{ type: "tool_result", tool_use_id: "t4", content: "unknown tool nope__x", is_error: true },
		],
	});
	assert.deepEqual(r.messages, [...second.body.messages, { role: "assistant", content: [{ type: "text", text: "2 + 3 = 5" }] }]);

	// Streamable HTTP: initialize without a session, then the session id and protocol version on every request
	assert.deepEqual(docs.seen.map((s) => [s.method, s.session ?? null, s.version ?? null]), [
		["initialize", null, null],
		["notifications/initialized", "sess-1", "2025-06-18"],
		["tools/list", "sess-1", "2025-06-18"],
		["tools/list", "sess-1", "2025-06-18"],
		["tools/call", "sess-1", "2025-06-18"],
	]);
	assert.ok(docs.seen.every((s) => s.accept === "application/json, text/event-stream"));
	assert.equal(docs.seen[3].params.cursor, "1");
	// the graph shows the routed MCP messages
	assert.deepEqual(new Set(ran.graph.map((g) => `${g.nsid} ${g.handler}`)), new Set([`${MCP} code`, "com.example.mcp.kcp kcp"]));
});

test("no composition: the default hook answers null, so no tools; earlier messages (held by the host) are the memory", async () => {
	const infer = inference([text("hi again")]);
	const earlier = [{ role: "user", content: "hello" }, { role: "assistant", content: [{ type: "text", text: "hi" }] }];
	const ran = await runModule(ctxOf(infer), mod, "run", { prompt: "and again", endpoint: "https://inference.test/", system: "be brief", maxTokens: "100" }, async () => assert.fail("no XRPC"), { http: allow(), budget: UNLIMITED, held: { messages: earlier } });
	assert.equal(ran.error, undefined);
	assert.equal(ran.result.text, "hi again");
	assert.deepEqual(ran.result.servers, [{ name: "mcp", phase: "skipped", tools: [], hook: MCP }]);
	const body = infer.requests[0].body;
	assert.equal(body.tools, undefined);
	assert.equal(body.system, "be brief");
	assert.equal(body.max_tokens, 100);
	assert.equal(body.model, "claude-opus-5-5");
	assert.deepEqual(body.messages, [...earlier, { role: "user", content: "and again" }]);
});

test("the step cap: the last call may not use tools, and an unanswered tool_use gets an error result", async () => {
	const code = mcpServer({ add: () => ({ content: [{ type: "text", text: "1" }] }) });
	const loop = () => uses([`t${Math.random()}`.replace(".", ""), "mcp__add", {}]);
	const infer = inference([loop, loop, loop]);
	const ctx = ctxOf(infer);
	const composition = await loadComposition(ctx, { $type: "com.fedfork.exec.composition", name: "a", instances: [{ name: "agent", module: "asyncai" }], handlers: [{ name: "code" }], edges: [{ from: "agent", nsid: MCP, handler: "code", reason: "tools" }] }, { load: async () => mod });
	const ran = await runCompose(ctx, composition, "agent", "run", { prompt: "go", endpoint: "https://inference.test", maxSteps: 3 }, { handlers: { code }, http: allow(), budget: UNLIMITED });
	assert.equal(ran.result.stop, "max_steps");
	assert.equal(ran.result.steps, 3);
	const last = infer.requests.at(-1).body;
	assert.deepEqual(last.tool_choice, { type: "none" });
	assert.match(last.messages.at(-1).content.at(-1).text, /Tool budget exhausted/);
	assert.equal(infer.requests[1].body.tool_choice, undefined);
	const closing = ran.result.messages.at(-1);
	assert.equal(closing.role, "user");
	assert.deepEqual(closing.content.map((b) => [b.type, b.content, b.is_error]), [["tool_result", "not run: the step cap was reached", true]]);
});

test("a 529 is retried after retry-after; a refused origin and a failing MCP server are errors in the result", async () => {
	const busy = () => new Response("overloaded", { status: 529, headers: { "retry-after": "0" } });
	const infer = inference([busy(), text("ok")]);
	const broken = streamableHttp(() => ({ jsonrpc: "2.0", id: "init.0", error: { code: -32600, message: "nope" } }));
	const ran = await runModule(ctxOf(infer, { "https://mcp.test": broken.handle }), mod, "run", { prompt: "x", endpoint: "https://inference.test", mcp: [{ name: "b", url: "https://mcp.test/mcp" }] }, async () => {}, { http: allow("https://mcp.test"), budget: UNLIMITED });
	assert.equal(ran.result.text, "ok");
	assert.equal(ran.result.steps, 1);
	assert.equal(infer.requests.length, 2);
	assert.deepEqual(ran.result.servers, [{ name: "b", phase: "error", tools: [], url: "https://mcp.test/mcp", error: "nope (-32600)" }]);

	const refused = await runModule(ctxOf(inference([])), mod, "run", { prompt: "x", endpoint: "https://inference.test", mcp: [{ name: "b", url: "https://mcp.test/mcp" }] }, async () => {}, { http: {}, budget: UNLIMITED });
	assert.equal(refused.result.stop, "error");
	assert.equal(refused.result.error, "inference: com.fedfork.exec.http.fetch: this host does not allow https://inference.test");
	assert.equal(refused.result.servers[0].error, "com.fedfork.exec.http.fetch: this host does not allow https://mcp.test");
});

test("bad input fails the run", async () => {
	for (const [input, error] of [
		[{}, "prompt is required"],
		[{ prompt: "x", mcp: [{ name: "a b" }] }, 'mcp server name "a b": 1 to 32 of A-Z a-z 0-9 _ -'],
		[{ prompt: "x", mcp: [{ name: "a" }, { name: "a" }] }, "mcp server name a is used twice"],
		[{ prompt: "x", maxSteps: "0" }, "maxSteps must be a whole number from 1 to 1000"],
		[{ prompt: "x", mcp: "[" }, /^mcp: /],
	]) {
		const ran = await runModule(ctxOf(inference([])), mod, "run", input, async () => {}, { budget: UNLIMITED });
		if (error instanceof RegExp) assert.match(ran.error, error);
		else assert.equal(ran.error, error);
	}
});

test("the transcript lives in the host: each inference body has it whole, no event carries it, the result gets it", async () => {
	const big = "x".repeat(400_000);
	// three tool turns, each result ~400 KB: past 1 MiB of transcript, which would have ended a run that kept it in state
	const code = streamableHttp(mcpServer({ echo: () => ({ content: [{ type: "text", text: big }] }) }));
	const infer = inference([uses(["t1", "code__echo", {}]), uses(["t2", "code__echo", {}]), uses(["t3", "code__echo", {}]), text("done")]);
	const ctx = ctxOf(infer, { "https://mcp.test": code.handle });
	let largest = 0;
	const sized = { ...ctx, wasm: { run: async (call) => ((largest = Math.max(largest, JSON.stringify(call.input).length)), ctx.wasm.run(call)) } };
	const ran = await runModule(sized, mod, "run", { prompt: "go", endpoint: "https://inference.test", model: "m", mcp: [{ name: "code", url: "https://mcp.test/mcp" }] }, async () => {}, { http: allow("https://mcp.test"), budget: UNLIMITED });
	assert.equal(ran.error, undefined, ran.error);
	assert.equal(ran.result.text, "done");
	assert.equal(infer.requests.length, 4);
	assert.equal(infer.requests.at(-1).body.messages.length, 7, "prompt, then three assistant/tool-result pairs");
	assert.ok(JSON.stringify(infer.requests.at(-1).body).length > 1_200_000);
	assert.deepEqual(ran.result.messages.slice(0, 1), [{ role: "user", content: "go" }]);
	assert.equal(ran.result.messages.length, 8);
	assert.ok(largest < 600_000, `the largest event input was ${largest} bytes`);
});

// Compaction. Every tool_result in a request answers a tool_use in the assistant message before it.
const paired = (messages) =>
	messages.every((m, i) => m.role !== "user" || !Array.isArray(m.content) || m.content.filter((b) => b.type === "tool_result").every((r) => messages[i - 1]?.content?.some?.((b) => b.type === "tool_use" && b.id === r.tool_use_id)));
const SUMMARY = /^This session continues an earlier conversation that was compacted/;

test("client compaction: past the threshold the transcript becomes a summary and the last tool round, thinking dropped", async () => {
	const code = mcpServer({ echo: ({ n }) => ({ content: [{ type: "text", text: `echo ${n}` }] }) });
	const used = (tokens, ...calls) => ({ ...uses(...calls), content: [{ type: "thinking", thinking: "", signature: "sig" }, ...uses(...calls).content], usage: { input_tokens: tokens, output_tokens: 10 } });
	const infer = inference([
		used(100, ["t1", "mcp__echo", { n: 1 }]),
		used(600, ["t2", "mcp__echo", { n: 2 }], ["t3", "mcp__echo", { n: 3 }]), // 610 > 1000 * 0.5: the next request compacts first
		{ content: [{ type: "thinking", thinking: "" }, { type: "text", text: "notes <summary>echoed 1, then asked for 2 and 3</summary>" }], stop_reason: "end_turn", usage: { input_tokens: 700, output_tokens: 20 } },
		used(200, ["t4", "mcp__echo", { n: 4 }]), // under the threshold again
		text("done"),
	]);
	const ctx = ctxOf(infer);
	const composition = await loadComposition(ctx, { $type: "com.fedfork.exec.composition", name: "a", instances: [{ name: "agent", module: "asyncai" }], handlers: [{ name: "code" }], edges: [{ from: "agent", nsid: MCP, handler: "code", reason: "tools" }] }, { load: async () => mod });
	const outputs = [];
	const ran = await runCompose(ctx, composition, "agent", "run", { prompt: "go", endpoint: "https://inference.test", contextWindow: 1000, compactAt: "0.5" }, { handlers: { code }, http: allow(), budget: UNLIMITED, onEvent: (_, out) => outputs.push(structuredClone(out)) });
	assert.equal(ran.error, undefined, ran.error);
	const r = ran.result;
	assert.equal(r.stop, "end_turn");
	assert.equal(r.text, "done");
	assert.equal(r.steps, 4, "the summary request is not a step");
	assert.equal(r.compactions, 1);
	assert.deepEqual(r.usage, { input_tokens: 1610, output_tokens: 52 });
	assert.equal(infer.requests.length, 5);

	// the summary request: the whole transcript so far (append-only), the instruction after the last tool results
	const summarize = infer.requests[2].body;
	assert.equal(summarize.messages.length, 5);
	assert.deepEqual(summarize.messages.slice(0, 4), infer.requests[1].body.messages.concat(summarize.messages[3]).slice(0, 4));
	const last = summarize.messages.at(-1).content;
	assert.deepEqual(last.slice(0, 2).map((b) => [b.type, b.tool_use_id]), [["tool_result", "t2"], ["tool_result", "t3"]]);
	assert.match(last.at(-1).text, /<summary><\/summary> block/);
	assert.deepEqual(summarize.tools, infer.requests[1].body.tools, "the same tools: the cached prefix holds");

	// after it: the summary (with the prompt), then the last tool round verbatim but for its thinking
	const after = infer.requests[3].body.messages;
	assert.equal(after.length, 3);
	assert.equal(after[0].role, "user");
	assert.match(after[0].content, SUMMARY);
	assert.match(after[0].content, /echoed 1, then asked for 2 and 3\n\nThe user's latest request: go$/);
	assert.doesNotMatch(after[0].content, /notes|<summary>/);
	assert.deepEqual(after[1], { role: "assistant", content: uses(["t2", "mcp__echo", { n: 2 }], ["t3", "mcp__echo", { n: 3 }]).content });
	assert.deepEqual(after[2], { role: "user", content: [{ type: "tool_result", tool_use_id: "t2", content: "echo 2" }, { type: "tool_result", tool_use_id: "t3", content: "echo 3" }] });
	for (const req of infer.requests) assert.ok(paired(req.body.messages), "no tool_result without its tool_use");
	assert.deepEqual(r.messages, [...infer.requests[4].body.messages, { role: "assistant", content: [{ type: "text", text: "done" }] }]);
	assert.equal(infer.requests[4].body.messages.length, 5);

	// a page sees it: the summary request's call, then the put that restarts the held transcript
	const calls = outputs.flatMap((o) => o.calls ?? []);
	assert.ok(calls.some((c) => c.id === "compact.1" && c.nsid === "com.fedfork.exec.http.fetch"));
	const put = calls.find((c) => c.nsid === "com.fedfork.exec.held.put");
	assert.deepEqual(put.input, { name: "messages", value: [after[0]] });
});

test("client compaction at the start: earlier turns past the threshold are summarized before the first step", async () => {
	const earlier = [{ role: "user", content: "hello" }, { role: "assistant", content: [{ type: "text", text: "x".repeat(4000) }] }];
	const infer = inference([{ content: [{ type: "text", text: "they said hello" }], stop_reason: "end_turn", usage: { input_tokens: 1000, output_tokens: 5 } }, text("hi again")]);
	const ran = await runModule(ctxOf(infer), mod, "run", { prompt: "and again", endpoint: "https://inference.test", contextWindow: "1000" }, async () => {}, { http: allow(), budget: UNLIMITED, held: { messages: earlier } });
	assert.equal(ran.error, undefined, ran.error);
	assert.equal(ran.result.compactions, 1);
	assert.equal(ran.result.steps, 1);
	assert.deepEqual(infer.requests[0].body.messages.at(-1).content.map((b) => b.type), ["text", "text"], "the prompt, then the instruction");
	assert.equal(infer.requests[0].body.messages.at(-1).content[0].text, "and again");
	const [first, ...rest] = infer.requests[1].body.messages;
	assert.deepEqual(rest, []);
	assert.match(first.content, SUMMARY);
	assert.match(first.content, /they said hello\n\nThe user's latest request: and again$/);
	assert.equal(ran.result.messages.length, 2, "the next run's memory is the compacted one");
});

test("compaction off, a failed summary, and bad options", async () => {
	const big = { content: [{ type: "text", text: "a" }], stop_reason: "end_turn", usage: { input_tokens: 999_999, output_tokens: 1 } };
	const off = await runModule(ctxOf(inference([big])), mod, "run", { prompt: "x", endpoint: "https://inference.test", compact: false }, async () => {}, { http: allow(), budget: UNLIMITED });
	assert.equal(off.result.compactions, 0);

	const code = mcpServer({ echo: () => ({ content: [{ type: "text", text: "e" }] }) });
	const infer = inference([{ ...uses(["t1", "mcp__echo", {}]), usage: { input_tokens: 999_999 } }, uses(["t2", "mcp__echo", {}])]);
	const ctx = ctxOf(infer);
	const composition = await loadComposition(ctx, { $type: "com.fedfork.exec.composition", name: "a", instances: [{ name: "agent", module: "asyncai" }], handlers: [{ name: "code" }], edges: [{ from: "agent", nsid: MCP, handler: "code", reason: "tools" }] }, { load: async () => mod });
	const failed = await runCompose(ctx, composition, "agent", "run", { prompt: "go", endpoint: "https://inference.test" }, { handlers: { code }, http: allow(), budget: UNLIMITED });
	assert.equal(failed.result.stop, "error");
	assert.equal(failed.result.error, 'compaction: no summary (stop_reason "tool_use")');

	for (const [input, error] of [
		[{ prompt: "x", compact: "sometimes" }, "compact must be client, server or off"],
		[{ prompt: "x", compactAt: 1.5 }, "compactAt must be a number above 0, at most 1"],
		[{ prompt: "x", contextWindow: 0 }, "contextWindow must be a whole number from 1 to 100000000"],
	]) assert.equal((await runModule(ctxOf(inference([])), mod, "run", input, async () => {}, { budget: UNLIMITED })).error, error);
});

test("server compaction: the beta header and edit, and the held transcript restarts at the compaction block", async () => {
	const code = mcpServer({ echo: () => ({ content: [{ type: "text", text: "e" }] }) });
	const compacted = { content: [{ type: "text", text: "old" }, { type: "compaction", content: "S" }, { type: "tool_use", id: "t1", name: "mcp__echo", input: {} }], stop_reason: "tool_use", usage: { input_tokens: 9 } };
	const infer = inference([compacted, text("done")]);
	const ctx = ctxOf(infer);
	const composition = await loadComposition(ctx, { $type: "com.fedfork.exec.composition", name: "a", instances: [{ name: "agent", module: "asyncai" }], handlers: [{ name: "code" }], edges: [{ from: "agent", nsid: MCP, handler: "code", reason: "tools" }] }, { load: async () => mod });
	const ran = await runCompose(ctx, composition, "agent", "run", { prompt: "go", endpoint: "https://inference.test", compact: "server", contextWindow: 10_000, compactPrompt: "keep ids" }, { handlers: { code }, http: allow(), budget: UNLIMITED });
	assert.equal(ran.result.error, undefined, ran.result.error);
	assert.equal(ran.result.compactions, 1);
	const [first, second] = infer.requests;
	assert.equal(first.headers["anthropic-beta"], "compact-2026-01-12");
	assert.deepEqual(first.body.context_management, { edits: [{ type: "compact_20260112", trigger: { type: "input_tokens", value: 50_000 }, instructions: "keep ids" }] });
	assert.deepEqual(second.body.messages, [
		{ role: "assistant", content: compacted.content.slice(1) },
		{ role: "user", content: [{ type: "tool_result", tool_use_id: "t1", content: "e" }] },
	]);
});

// OpenAI chat completions (co/core and any compatible endpoint): the same loop, that API's shapes.
const chat = (message, finish, usage = { prompt_tokens: 10, completion_tokens: 2, total_tokens: 12 }) => ({ id: "c", object: "chat.completion", choices: [{ index: 0, message: { role: "assistant", ...message }, finish_reason: finish }], usage });
const fn = (id, name, args) => ({ id, type: "function", function: { name, arguments: JSON.stringify(args) } });
const OPENAI = { provider: "openai", endpoint: "https://inference.test/v1", model: "stub" };

test("openai: function tools from MCP inputSchema, tool_calls to tool messages, the system message held apart", async () => {
	const code = mcpServer({ add: ({ a, b }) => ({ content: [{ type: "text", text: String(a + b) }] }), boom: () => ({ content: [{ type: "text", text: "no" }], isError: true }) });
	const infer = inference([chat({ content: null, tool_calls: [fn("c1", "code__add", { a: 2, b: 3 }), fn("c2", "code__boom", {})] }, "tool_calls"), chat({ content: "5" }, "stop")]);
	const ctx = ctxOf(infer);
	const composition = await loadComposition(ctx, { $type: "com.fedfork.exec.composition", name: "a", instances: [{ name: "agent", module: "asyncai" }], handlers: [{ name: "code" }], edges: [{ from: "agent", nsid: MCP, handler: "code", reason: "tools" }] }, { load: async () => mod });
	const earlier = [{ role: "user", content: "hello" }, { role: "assistant", content: "hi" }];
	const ran = await runCompose(ctx, composition, "agent", "run", { ...OPENAI, prompt: "add 2 and 3", system: "be brief", mcp: [{ name: "code" }] }, { handlers: { code }, http: allow(), budget: UNLIMITED, held: { messages: earlier } });
	assert.equal(ran.error, undefined, ran.error);
	const r = ran.result;
	assert.deepEqual([r.stop, r.text, r.steps], ["end_turn", "5", 2]);
	assert.deepEqual(r.usage, { prompt_tokens: 20, completion_tokens: 4, total_tokens: 24 });
	const [first, second] = infer.requests;
	assert.equal(first.headers["x-api-key"], "HOST-KEY");
	assert.equal(first.headers["anthropic-version"], undefined);
	assert.deepEqual(first.body.tools[0], { type: "function", function: { name: "code__add", parameters: { type: "object" }, description: "add" } });
	assert.equal(first.body.system, undefined);
	assert.deepEqual(first.body.messages, [{ role: "system", content: "be brief" }, ...earlier, { role: "user", content: "add 2 and 3" }]);
	assert.deepEqual(second.body.messages.slice(-3), [
		{ role: "assistant", content: "", tool_calls: [fn("c1", "code__add", { a: 2, b: 3 }), fn("c2", "code__boom", {})] },
		{ role: "tool", tool_call_id: "c1", content: "5" },
		{ role: "tool", tool_call_id: "c2", content: "Error: no" },
	]);
	// the memory is the transcript without the system message
	assert.deepEqual(r.messages, [...second.body.messages.slice(1), { role: "assistant", content: "5" }]);
});

test("openai: a streamed answer is collected (content, tool calls by index, usage); the step cap", async () => {
	const sse = (...chunks) => new Response(chunks.map((c) => `data: ${JSON.stringify(c)}\n\n`).join("") + "data: [DONE]\n\n", { headers: { "content-type": "text/event-stream" } });
	const delta = (d, finish = null) => ({ object: "chat.completion.chunk", choices: [{ index: 0, delta: d, finish_reason: finish }] });
	const code = mcpServer({ add: () => ({ content: [{ type: "text", text: "1" }] }) });
	const infer = inference([
		sse(delta({ role: "assistant", content: "let me " }), delta({ content: "add", tool_calls: [{ index: 0, id: "c1", type: "function", function: { name: "mcp__add", arguments: "{\"a\":" } }] }), delta({ tool_calls: [{ index: 0, function: { arguments: "1}" } }] }, "tool_calls"), { choices: [], usage: { prompt_tokens: 7, completion_tokens: 3, total_tokens: 10 } }),
		chat({ content: "", tool_calls: [fn("c2", "mcp__add", {})] }, "tool_calls"),
	]);
	const ctx = ctxOf(infer);
	const composition = await loadComposition(ctx, { $type: "com.fedfork.exec.composition", name: "a", instances: [{ name: "agent", module: "asyncai" }], handlers: [{ name: "code" }], edges: [{ from: "agent", nsid: MCP, handler: "code", reason: "tools" }] }, { load: async () => mod });
	const ran = await runCompose(ctx, composition, "agent", "run", { ...OPENAI, prompt: "go", stream: "true", maxSteps: 2 }, { handlers: { code }, http: allow(), budget: UNLIMITED });
	assert.equal(ran.error, undefined, ran.error);
	const [first, last] = infer.requests;
	assert.deepEqual([first.body.stream, first.body.stream_options, first.headers.accept], [true, { include_usage: true }, "text/event-stream"]);
	assert.deepEqual(last.body.messages.slice(0, 3), [
		{ role: "user", content: "go" },
		{ role: "assistant", content: "let me add", tool_calls: [fn("c1", "mcp__add", { a: 1 })] },
		{ role: "tool", tool_call_id: "c1", content: "1" },
	]);
	assert.equal(last.body.tool_choice, "none");
	assert.match(last.body.messages.at(-1).content, /Tool budget exhausted/);
	assert.equal(ran.result.stop, "max_steps");
	assert.deepEqual(ran.result.messages.at(-1), { role: "tool", tool_call_id: "c2", content: "Error: not run: the step cap was reached" });
	assert.equal(ran.result.usage.prompt_tokens, 17);
});

test("openai: client compaction summarizes from the answer's usage; bad options", async () => {
	const code = mcpServer({ add: () => ({ content: [{ type: "text", text: "x".repeat(400) }] }) });
	const big = { prompt_tokens: 900, completion_tokens: 10, total_tokens: 910 };
	const infer = inference([chat({ content: "", tool_calls: [fn("c1", "mcp__add", {})] }, "tool_calls", big), chat({ content: "<summary>added once</summary>" }, "stop"), chat({ content: "done" }, "stop")]);
	const ctx = ctxOf(infer);
	const composition = await loadComposition(ctx, { $type: "com.fedfork.exec.composition", name: "a", instances: [{ name: "agent", module: "asyncai" }], handlers: [{ name: "code" }], edges: [{ from: "agent", nsid: MCP, handler: "code", reason: "tools" }] }, { load: async () => mod });
	const ran = await runCompose(ctx, composition, "agent", "run", { ...OPENAI, prompt: "go", contextWindow: 1000 }, { handlers: { code }, http: allow(), budget: UNLIMITED });
	assert.equal(ran.error, undefined, ran.error);
	assert.equal(ran.result.compactions, 1);
	const summary = infer.requests[1].body.messages;
	assert.equal(summary.at(-1).role, "user", "the instruction after the tool message");
	assert.match(summary.at(-1).content, /<summary>/);
	const after = infer.requests[2].body.messages;
	assert.match(after[0].content, SUMMARY);
	assert.deepEqual(after.slice(1).map((m) => m.role), ["assistant", "tool"], "the last tool round, kept");
	const bad = async (input) => (await runModule(ctxOf(inference([])), mod, "run", { prompt: "x", ...input }, async () => {}, { http: allow() })).error;
	assert.match(await bad({ provider: "openai" }), /model is required with provider openai/);
	assert.match(await bad({ provider: "gemini" }), /provider gemini: anthropic or openai/);
	assert.match(await bad({ ...OPENAI, compact: "server" }), /compact server is Anthropic's/);
	assert.match(await bad({ stream: true }), /stream is for provider openai/);
});

// Dynamic MCP servers (dynamicMcp): the model connects and disconnects them mid-run with the built-in tools.
const resultsOf = (body) => Object.fromEntries(body.messages.at(-1).content.map((b) => [b.tool_use_id, [b.content, b.is_error === true]]));
const names = (body) => (body.tools ?? []).map((t) => t.name);

test("dynamic MCP: mcp_connect each descriptor kind, tools from the next inference, mcp_disconnect, gates and clashes", { skip: noScheme }, async () => {
	const code = mcpServer({ add: () => ({ content: [{ type: "text", text: "3" }] }) });
	const kcp = mcpServer({ bid: () => ({ content: [{ type: "text", text: "kcp bid" }] }) });
	const docs = streamableHttp(mcpServer({ search: ({ q }) => ({ content: [{ type: "text", text: `found ${q}` }] }) }), { session: "s1" });
	// the VM guest agents, as the host's webrtc: and iroh: fetchers reach them: stateless JSON (no session)
	const vm = streamableHttp(mcpServer({ exec: ({ command }) => ({ content: [{ type: "text", text: `ran ${command}` }], structuredContent: { exitCode: 0 } }), read: () => ({ content: [], structuredContent: { output: "", done: false } }) }, { pageSize: 1 }));
	const iroh = streamableHttp(mcpServer({ exec: () => ({ content: [] }) }));
	const [webrtcFetch, irohFetch] = [fetcher(vm.handle), fetcher(iroh.handle)];
	const VM = "webrtc:///eyJyZWNlaXB0Ijp7fX0/shell/mcp";
	const connect = (id, name, server) => [id, "mcp_connect", { name, server }];
	const infer = inference([
		uses(
			connect("c1", "docs", { url: "https://mcp.test/allowed/mcp" }),
			connect("c2", "vm", { url: VM }),
			connect("c3", "kcp", { hook: "com.example.mcp.kcp" }),
			connect("c4", "other", { url: "https://mcp.test/allowedx/mcp" }),
			connect("c5", "held", { nsid: "com.fedfork.exec.held.put" }),
			connect("c6", "code", { url: "https://mcp.test/allowed/mcp" }),
			connect("c7", "gone", { hook: "com.example.mcp.unrouted" }),
			connect("c8", "both", { url: "https://mcp.test/allowed/mcp", hook: "com.example.mcp.kcp" }),
			["c9", "docs__search", { q: "early" }],
			connect("c10", "iroh", { url: "iroh:///eyJ0aWNrZXQiOiJUIn0/shell/mcp" }),
		),
		(body) => {
			assert.deepEqual(names(body), ["code__add", "vm0__exec", "mcp_connect", "mcp_disconnect", "docs__search", "vm__exec", "vm__read", "kcp__bid"]);
			return uses(["d1", "docs__search", { q: "mcp" }], ["d2", "vm__exec", { command: "ls" }], ["d3", "mcp_disconnect", { name: "docs" }], ["d4", "mcp_disconnect", { name: "nope" }]);
		},
		(body) => {
			assert.deepEqual(names(body), ["code__add", "vm0__exec", "mcp_connect", "mcp_disconnect", "vm__exec", "vm__read", "kcp__bid"]);
			return text("done");
		},
	]);
	const ctx = ctxOf(infer, { "https://mcp.test": docs.handle });
	const composition = await loadComposition(ctx, {
		$type: "com.fedfork.exec.composition",
		name: "agent",
		instances: [{ name: "agent", module: "asyncai" }],
		handlers: [{ name: "code" }, { name: "kcp" }],
		edges: [
			{ from: "agent", nsid: MCP, handler: "code", reason: "tools" },
			{ from: "agent", nsid: "com.example.mcp.kcp", handler: "kcp", reason: "a server the model may connect" },
		],
	}, { load: async () => mod });
	const dynamicMcp = ["https://mcp.test/allowed", "com.example.mcp.kcp", "com.example.mcp.unrouted", "webrtc:///"];
	const mcp = [{ name: "code" }, { name: "vm0", server: { url: "iroh:///eyJ0aWNrZXQiOiJUIn0/shell/mcp" } }];
	const http = { ...allow("https://mcp.test"), "webrtc://*": {}, "iroh://*": {} };
	const ran = await runCompose(ctx, composition, "agent", "run", { prompt: "go", endpoint: "https://inference.test", mcp, dynamicMcp }, { handlers: { code, kcp }, http, fetchers: { "webrtc:": webrtcFetch.fetch, "iroh:": irohFetch.fetch }, budget: UNLIMITED });
	assert.equal(ran.error, undefined, ran.error);
	const r = ran.result;
	assert.deepEqual([r.stop, r.text, r.steps, r.error], ["end_turn", "done", 3, undefined]);

	// the first turn's results: three connected, the rest refused; a new server's tools are not offered yet
	const first = resultsOf(infer.requests[1].body);
	assert.deepEqual(JSON.parse(first.c1[0]), { connected: "docs", protocolVersion: "2025-11-25", tools: ["docs__search"] });
	assert.deepEqual(JSON.parse(first.c2[0]).tools, ["vm__exec", "vm__read"], "tools/list followed its cursor");
	assert.deepEqual(JSON.parse(first.c3[0]).tools, ["kcp__bid"]);
	assert.deepEqual(first.c4, ["not connected: https://mcp.test/allowedx/mcp is not allowed here (allowed: https://mcp.test/allowed, com.example.mcp.kcp, com.example.mcp.unrouted, webrtc:///)", true]);
	assert.deepEqual(first.c5, ["not connected: mcp server held: a server is {url} or {hook} (no nsid descriptors: a host transport is a webrtc:// or iroh:// url)", true]);
	assert.match(first.c6[0], /^not connected: an MCP server named code is connected/);
	assert.match(first.c7[0], /^not connected: .*com.example.mcp.unrouted/);
	assert.deepEqual(first.c8, ["not connected: mcp server both: one of url or hook", true]);
	assert.deepEqual(first.c9, ["unknown tool docs__search", true]);
	assert.match(first.c10[0], /^not connected: iroh:\/\/\/.* is not allowed here/, "webrtc:/// allows no iroh URL");

	// a tool call out before its server's disconnect still answers
	const second = resultsOf(infer.requests[2].body);
	assert.deepEqual(second, {
		d1: ["found mcp", false],
		d2: ['{"exitCode":0}', false],
		d3: ["disconnected docs: its tools are gone from your next turn on", false],
		d4: ['no MCP server named "nope"', true],
	});

	// the host fetchers got each message as a POST of the server's URL; no session (stateless)
	assert.deepEqual(webrtcFetch.urls, Array(5).fill(VM));
	assert.deepEqual(vm.seen.map((s) => [s.method, s.session ?? null, s.version ?? null]), [["initialize", null, null], ["notifications/initialized", null, "2025-11-25"], ["tools/list", null, "2025-11-25"], ["tools/list", null, "2025-11-25"], ["tools/call", null, "2025-11-25"]]);
	assert.deepEqual(iroh.seen.map((s) => s.method), ["initialize", "notifications/initialized", "tools/list"]);
	assert.deepEqual(docs.seen.map((s) => [s.method, s.session ?? null]), [["initialize", null], ["notifications/initialized", "s1"], ["tools/list", "s1"], ["tools/call", "s1"]]);

	// the live set in the result: the start servers and the connected ones, less the disconnected and failed
	assert.deepEqual(r.servers.map((s) => [s.name, s.phase, s.dynamic ?? false]), [["code", "ready", false], ["vm0", "ready", false], ["vm", "ready", true], ["kcp", "ready", true]]);
	assert.deepEqual(r.servers[2], { name: "vm", phase: "ready", tools: ["exec", "read"], url: VM, protocolVersion: "2025-11-25", dynamic: true });
	assert.deepEqual(r.tools, names(infer.requests[2].body));
});

test("a VM's MCP server over webrtc://: each request's deadline is past the tool call's own waitMs or timeoutMs", { skip: noScheme }, async () => {
	const vm = streamableHttp(mcpServer({
		read: ({ waitMs }) => ({ content: [], structuredContent: { output: "tick\n", cursor: 5, done: false, waitMs } }),
		exec: () => ({ content: [], structuredContent: { exitCode: 0, stdout: "", stderr: "", truncated: false } }),
	}));
	const infer = inference([uses(["r1", "vm__read", { id: "j1", waitMs: 30_000 }], ["r2", "vm__exec", { command: "make", timeoutMs: 600_000 }], ["r3", "vm__exec", { command: "ls" }], ["r4", "vm__read", { id: "j1", waitMs: 1e12 }]), text("done")]);
	const outputs = [];
	const ran = await runModule(ctxOf(infer), mod, "run", { prompt: "follow", endpoint: "https://inference.test", mcp: [{ name: "vm", url: "webrtc:///e30/shell/mcp" }] }, async () => {}, {
		http: { ...allow(), "webrtc://*": {} },
		fetchers: { "webrtc:": fetcher(vm.handle).fetch },
		budget: UNLIMITED,
		onEvent: (_, output) => outputs.push(output),
	});
	assert.equal(ran.error, undefined, ran.error);
	assert.equal(ran.result.text, "done");
	const deadlines = outputs.flatMap((o) => o.calls ?? []).filter((c) => c.nsid === "com.fedfork.exec.http.fetch" && c.input.url.startsWith("webrtc:")).map((c) => [c.id, c.timeoutMs]);
	assert.deepEqual(deadlines, [["init.0", 120_000], ["inited.0", 120_000], ["list.0.0", 120_000], ["tool.2", 150_000], ["tool.3", 720_000], ["tool.4", 120_000], ["tool.5", 720_000]]);
	assert.deepEqual(resultsOf(infer.requests[1].body).r1, ['{"cursor":5,"done":false,"output":"tick\\n","waitMs":30000}', false]);
});

test("dynamic MCP: true allows urls; the connected set survives a compaction; off by default; bad options", async () => {
	const docs = streamableHttp(mcpServer({ search: () => ({ content: [{ type: "text", text: "found" }] }) }));
	const infer = inference([
		uses(["c1", "mcp_connect", { name: "docs", server: { url: "https://mcp.test/mcp" } }], ["c2", "mcp_connect", { name: "vm", server: { hook: "com.example.mcp.vm" } }]),
		{ ...uses(["t1", "docs__search", {}]), usage: { input_tokens: 600, output_tokens: 10 } }, // past 1000 * 0.5: the next request compacts first
		text("<summary>searched the docs</summary>"),
		(body) => {
			assert.ok(names(body).includes("docs__search"), "the connected server's tools after the compaction");
			assert.match(body.messages[0].content, /MCP servers connected with mcp_connect, still connected \(their tools are <name>__<tool>\): docs\.\n\nThe user's latest request: go$/);
			return text("done");
		},
	]);
	const ran = await runModule(ctxOf(infer, { "https://mcp.test": docs.handle }), mod, "run", { prompt: "go", endpoint: "https://inference.test", dynamicMcp: "true", contextWindow: 1000, compactAt: 0.5 }, async () => assert.fail("no host call"), { http: allow("https://mcp.test"), budget: UNLIMITED });
	assert.equal(ran.error, undefined, ran.error);
	assert.deepEqual([ran.result.text, ran.result.compactions], ["done", 1]);
	assert.match(resultsOf(infer.requests[1].body).c2[0], /com.example.mcp.vm is not allowed here \(allowed: url\)/);
	assert.deepEqual(ran.result.servers.map((s) => [s.name, s.dynamic]), [["mcp", undefined], ["docs", true]]);

	// off by default: no built-in tools, so an mcp_connect is an unknown tool
	const off = inference([uses(["c1", "mcp_connect", { name: "docs", server: { url: "https://mcp.test/mcp" } }]), text("ok")]);
	const plain = await runModule(ctxOf(off), mod, "run", { prompt: "go", endpoint: "https://inference.test" }, async () => {}, { http: allow(), budget: UNLIMITED });
	assert.equal(off.requests[0].body.tools, undefined);
	assert.deepEqual(resultsOf(off.requests[1].body).c1, ["unknown tool mcp_connect", true]);
	assert.equal(plain.result.text, "ok");

	for (const dynamicMcp of [5, [""], { url: true }]) {
		const bad = await runModule(ctxOf(inference([])), mod, "run", { prompt: "x", dynamicMcp }, async () => {}, { budget: UNLIMITED });
		assert.match(bad.error, /^dynamicMcp must be true, false or a list/);
	}
	for (const [server, error] of [
		[{ name: "a", input: {} }, "mcp server a: a server is {url} or {hook} (no nsid descriptors: a host transport is a webrtc:// or iroh:// url)"],
		[{ name: "a", server: { nsid: "com.fedfork.exec.webrtc.mcp" } }, "mcp server a: a server is {url} or {hook} (no nsid descriptors: a host transport is a webrtc:// or iroh:// url)"],
		[{ name: "a", url: "mcp.test/mcp" }, "mcp server a: url must be <scheme>://... (https, or a host transport's webrtc:// or iroh://)"],
	]) {
		const bad = await runModule(ctxOf(inference([])), mod, "run", { prompt: "x", mcp: [server] }, async () => {}, { budget: UNLIMITED });
		assert.equal(bad.error, error);
	}
});