johnandersen777.bsky.social/atlibs-agent-reference-spa

Switch repository

Loading…

// node --test test/*.test.mjs  (after `npm ci` and scripts/stage.sh: the published modules, staged in dist/)
// The module -> MCP adapter and the agent with the real modules, runModule, policyCall, code mode (asc in a
// Node worker) and the asyncai agent; only the user's PDS (its OAuth session and its public reads) and the
// inference API are stubbed. The official MCP SDK client talks to the adapter over Streamable HTTP, as a
// remote client would through the ingress.
import assert from "node:assert/strict";
import { mkdtemp } from "node:fs/promises";
import { createServer } from "node:http";
import { tmpdir } from "node:os";
import { join } from "node:path";
import { test } from "node:test";

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";

// The page loads @publicdomainrelay/* from jsdelivr through its import map, and Node from node_modules (the same
// npm packages).
const ROOT = new URL("../dist/", import.meta.url);
const spa = await import("../spa.js");
const { actionsServer, toolName } = await import("@publicdomainrelay/atlibs/mcp/index.js");
const { requestScope } = await import("@publicdomainrelay/atlibs/oauth/index.js");
const { createNodeContext, fileFetch } = await import("@publicdomainrelay/atlibs/server/index.js");
// a hook default (the requester's user-data) loads from beside its module with the global fetch, as the page's
// does from its origin: dist/ here
const netFetch = globalThis.fetch;
globalThis.fetch = (url, init) => (String(url).startsWith("file:") ? fileFetch(url, init) : netFetch(url, init));

const did = "did:plc:lpfuqerea3deuoyrn7ojser4";
const PDS = "https://pds.test";
const DOMAIN = { uri: `at://${did}/at.marque.domain/example.com`, cid: "bafydomain", value: { $type: "at.marque.domain", domain: "example.com", status: "active" } };

const created = [];
// the user's computer.socialweb.vm.kept/self, when a test keeps a VM
let keptRecord = null;
const SSH_KEY = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIGp6bWl0ZXN0a2V5Zm9yYXRsaWJzYWdlbnRzcGF0ZXN0 agent";
// The user's PDS, public reads and the session's alike; `seen` records each request as "METHOD nsid".
function pds(seen) {
	return async (url, init = {}) => {
		const u = new URL(url, PDS);
		if (u.origin === "https://plc.directory") return Response.json({ service: [{ id: "#atproto_pds", type: "AtprotoPersonalDataServer", serviceEndpoint: PDS }] });
		const nsid = u.pathname.replace("/xrpc/", "");
		seen.push(`${init.method ?? "GET"} ${nsid}${u.searchParams.get("collection") ? ` ${u.searchParams.get("collection")}` : ""}`);
		if (nsid === "com.atproto.repo.listRecords" && u.searchParams.get("collection") === "at.marque.domain") return Response.json({ records: [DOMAIN] });
		if (nsid === "com.atproto.repo.listRecords" && u.searchParams.get("collection") === "dev.knotless.repo") return Response.json({ records: [{ uri: `at://${did}/dev.knotless.repo/demo`, cid: "bafyrepo", value: { name: "demo" } }] });
		if (nsid === "dev.cocore.account.createApiKey") return Response.json({ id: "k1", name: JSON.parse(init.body).name, key: "cocore_secret" });
		if (nsid === "com.atproto.repo.createRecord" && created.push(JSON.parse(init.body))) return Response.json({ uri: `at://${did}/${JSON.parse(init.body).collection}/3new`, cid: "bafynew" });
		if (nsid === "com.atproto.repo.getRecord" && u.searchParams.get("collection") === "computer.socialweb.vm.kept" && keptRecord) return Response.json(keptRecord);
		if (nsid === "com.atproto.repo.putRecord") return Response.json({ uri: `at://${did}/${JSON.parse(init.body).collection}/${JSON.parse(init.body).rkey}`, cid: "bafyput" });
		return Response.json({ error: "RecordNotFound", message: "not here" }, { status: 400 });
	};
}
const sessionOf = (seen) => ({ did, fetchHandler: (path, init) => pds(seen)(new URL(path, PDS).href, init) });

async function setup({ confirm = async () => true, fetch } = {}) {
	const seen = [];
	const ctx = createNodeContext({ id: "agent-spa-test", stateDir: await mkdtemp(join(tmpdir(), "agent-spa-")), fetch: fetch ?? pds(seen) });
	const { modules, policy } = await spa.loadAll(ctx, ROOT, fileFetch);
	const asked = [];
	const servers = spa.mcpServers({ ctx, policy, modules, session: sessionOf(seen), confirm: async (q) => (asked.push(q), confirm(q)) });
	return { ctx, modules, policy, servers, seen, asked };
}
const rpc = (server, method, params, id = 1) => server.message({ jsonrpc: "2.0", id, method, params });

test("one sign-in asks for every lib's session scope; each module keeps its own", async () => {
	const { policy, ctx } = await setup();
	ctx.close?.();
	const scope = requestScope(policy, Object.keys(policy.sessions)).split(" ");
	for (const s of ["repo:dev.knotless.ref?action=create&action=update", "blob:application/x-git-packed-objects", "repo:at.marque.dns?action=create&action=update", "repo:com.fedproxy.sshPublicKey?action=create", "rpc:dev.cocore.account.createApiKey?aud=did:web:cocore.dev%23cocore_console", "rpc:dev.cocore.inference.dispatch?aud=*", "repo:com.publicdomainrelay.temp.market.rfp?action=create&action=delete", "rpc:com.publicdomainrelay.temp.market.submitRfp?aud=*", "repo:computer.socialweb.vm.kept?action=create&action=update"]) assert.ok(scope.includes(s), s);
	assert.deepEqual(Object.keys(policy.sessions).sort(), ["asyncai", "cocore-inference", "cocore-keys", "fedproxy", "knotless", "marque", "requester"]);
	assert.equal(policy.modules.asyncai.session, "asyncai"); // its own: co/core inference by service auth
});

test("module.json actions become MCP tools: names, schemas, read-only hints", async () => {
	const { servers, ctx } = await setup();
	ctx.close?.();
	const init = await rpc(servers.knotless, "initialize", { protocolVersion: "2025-06-18", capabilities: {}, clientInfo: { name: "t", version: "0" } });
	assert.deepEqual(init.result, { protocolVersion: "2025-06-18", capabilities: { tools: {} }, serverInfo: { name: "atlibs-knotless", version: "0.1.0" } });
	assert.equal(await servers.knotless.message({ jsonrpc: "2.0", method: "notifications/initialized" }), null);
	const knotless = (await rpc(servers.knotless, "tools/list")).result.tools;
	assert.deepEqual(knotless.map((t) => [t.name, t.annotations.readOnlyHint]), [["create_repo", false], ["commit", false]]);
	assert.deepEqual(Object.keys(knotless[1].inputSchema.properties), ["name", "branch", "parent", "message", "author", "changes"]);
	// the model's servers: code mode and config.js TOOLS; cocore (the inference behind asyncai) is none of them
	assert.deepEqual(Object.keys(servers), ["code", "knotless", "marque", "fedproxy", "vm"]);
	assert.equal(servers.fedproxy.readOnly("list_keys"), true);
	assert.equal(servers.fedproxy.readOnly("create_key"), false);
	assert.equal(toolName("cocore", "cocore-keys", "list_keys"), "keys_list_keys");
	assert.equal((await rpc(servers.marque, "tools/call", { name: "nope" })).error.code, -32602);
	assert.equal((await rpc(servers.marque, "resources/list")).error.code, -32601);
});

test("a read runs as the user with no confirm; a write asks first, and a declined one never reaches the PDS", async () => {
	let answer = false;
	const { servers, seen, asked, ctx } = await setup({ confirm: async () => answer });
	const domains = await rpc(servers.marque, "tools/call", { name: "list_domains", arguments: {} });
	assert.equal(domains.result.isError, undefined, JSON.stringify(domains));
	assert.equal(domains.result.structuredContent.domains[0].domain, "example.com");
	assert.deepEqual(asked, []);

	seen.length = 0;
	const key = { handle: "alice.test", name: "agent", service: "agent", key: SSH_KEY };
	const declined = await rpc(servers.fedproxy, "tools/call", { name: "create_key", arguments: { ...key, did: "did:plc:mallory" } });
	assert.equal(declined.result.isError, true);
	assert.match(declined.result.content[0].text, /declined/);
	assert.deepEqual(asked, [{ server: "fedproxy", tool: "create_key", module: "fedproxy", entry: "create_key", arguments: key, caller: undefined }]);
	assert.deepEqual(seen, []);

	answer = true;
	const added = await rpc(servers.fedproxy, "tools/call", { name: "create_key", arguments: key });
	ctx.close?.();
	assert.equal(added.result.isError, undefined, JSON.stringify(added));
	assert.deepEqual(seen, ["POST com.atproto.repo.createRecord"]);
});

test("each tool is held to its own module's session: a narrowed scope fails the call", async () => {
	const { ctx, modules, policy, seen } = await setup();
	const narrowed = { ...policy, sessions: { ...policy.sessions, "cocore-keys": policy.sessions.marque } };
	const server = actionsServer({ name: "cocore", modules: [modules["cocore-keys"]], confirm: async () => true, run: (mod, entry, args) => spa.runAction({ ctx, policy: narrowed, session: sessionOf(seen) }, mod, entry, args) });
	const out = await rpc(server, "tools/call", { name: "keys_create_key", arguments: { name: "agent" } });
	ctx.close?.();
	assert.equal(out.result.isError, true);
	assert.match(out.result.content[0].text, /cocore-keys may not call dev.cocore.account.createApiKey/);
});

test("the MCP SDK client over Streamable HTTP sees only the public tools; write tools still ask the user", async () => {
	const { servers, asked, ctx } = await setup({ confirm: async () => false });
	const isPublic = (server, tool) => servers[server].readOnly(tool);
	const forward = spa.publicMcp(servers, isPublic);
	// what the ingress hands the page for each request, and back
	const http = createServer(async (req, res) => {
		let body = "";
		for await (const chunk of req) body += chunk;
		const out = await forward({ method: req.method, path: new URL(req.url, "http://x").pathname, headers: req.headers, body, callerDid: "did:plc:remote" });
		res.writeHead(out.status, { "content-type": out.contentType, ...out.headers }).end(typeof out.body === "string" ? out.body : JSON.stringify(out.body));
	});
	await new Promise((ok) => http.listen(0, "127.0.0.1", ok));
	const base = `http://127.0.0.1:${http.address().port}`;
	try {
		const client = new Client({ name: "remote", version: "0" });
		await client.connect(new StreamableHTTPClientTransport(new URL(`${base}/mcp/fedproxy`)));
		assert.deepEqual((await client.listTools()).tools.map((t) => t.name), ["list_keys"]);
		await assert.rejects(client.callTool({ name: "create_key", arguments: { name: "x" } }), /Unknown tool/);
		await client.close();

		// opted in: listed, and the user is still asked (here: declines)
		const optIn = spa.publicMcp(servers, (server, tool) => isPublic(server, tool) || tool === "create_key");
		const out = await optIn({ method: "POST", path: "/mcp/fedproxy", body: JSON.stringify({ jsonrpc: "2.0", id: 7, method: "tools/call", params: { name: "create_key", arguments: { handle: "alice.test", name: "agent", service: "agent", key: SSH_KEY } } }), callerDid: "did:plc:remote" });
		assert.equal(out.body.result.isError, true);
		assert.equal(asked.at(-1).caller, "did:plc:remote");

		const code = new Client({ name: "remote", version: "0" });
		await code.connect(new StreamableHTTPClientTransport(new URL(`${base}/mcp/code`)));
		assert.deepEqual((await code.listTools()).tools.map((t) => t.name), ["run_assemblyscript"]);
		await code.close();
		assert.equal((await forward({ method: "GET", path: "/mcp/code" })).status, 405);
		// CORS on every answer, and the browser's preflight
		for (const r of [await forward({ method: "GET", path: "/mcp/code" }), await forward({ method: "POST", path: "/mcp/code", body: '{"jsonrpc":"2.0","method":"notifications/initialized"}' }), await forward({ method: "POST", path: "/nope" })]) assert.deepEqual(r.headers, { "access-control-allow-origin": "*", "access-control-expose-headers": "mcp-session-id" });
		const pre = await forward({ method: "OPTIONS", path: "/mcp/marque" });
		assert.equal(pre.status, 204);
		assert.match(pre.headers["access-control-allow-headers"], /mcp-protocol-version/);
		assert.match(pre.headers["access-control-allow-methods"], /POST/);
		assert.deepEqual((await forward({ method: "POST", path: "/" })).body.servers, ["/mcp/code", "/mcp/knotless", "/mcp/marque", "/mcp/fedproxy", "/mcp/vm"]);
	} finally {
		http.close();
		ctx.close?.();
	}
});

// code mode calls a lib server's read-only tool (one JSON-RPC message) and filters what it returns; a write is refused
const toolCall = (name, args) => `Json.object().set("jsonrpc", Json.string("2.0")).set("id", Json.number(1)).set("method", Json.string("tools/call")).set("params", Json.object().set("name", Json.string("${name}")).set("arguments", ${args}))`;
const CODE_CALLS = `export function main(run: Run): void {
  if (run.start) { print("hi"); run.call("d", "com.fedfork.exec.mcp.marque", ${toolCall("list_domains", "Json.object()")}); return; }
  run.finish(Json.object().set("n", Json.number(17 * 23)).set("domain", run.output.get("result").get("structuredContent").get("domains").at(0).get("domain")));
}`;
const CODE_WRITES = `export function main(run: Run): void {
  if (run.start) { run.call("w", "com.fedfork.exec.mcp.knotless", ${toolCall("create_repo", 'Json.object().set("name", Json.string("x"))')}); return; }
  run.finish(run.output.get("error").get("message"));
}`;

test("the agent: asyncai with code mode and the lib servers in-process; inference stubbed", { timeout: 120_000 }, async () => {
	const requests = [];
	const turns = [
		{
			content: [
				{ type: "tool_use", id: "a", name: "code__run_assemblyscript", input: { code: CODE_CALLS } },
				{ type: "tool_use", id: "w", name: "code__run_assemblyscript", input: { code: CODE_WRITES } },
				{ type: "tool_use", id: "b", name: "marque__list_domains", input: {} },
				{ type: "tool_use", id: "c", name: "knotless__create_repo", input: { name: "demo" } },
			],
			stop_reason: "tool_use",
		},
		{ content: [{ type: "text", text: "391; example.com; you declined the repo" }], stop_reason: "end_turn" },
	];
	const seen = [];
	const stub = pds(seen);
	const fetch = async (url, init) => (url.startsWith("https://inference.test/") ? (requests.push({ headers: init.headers, body: JSON.parse(init.body) }), Response.json({ ...turns.shift(), usage: { input_tokens: 1, output_tokens: 1 } })) : stub(url, init));
	const { ctx, modules, servers, asked } = await setup({ fetch, confirm: async () => false });
	const composition = await spa.agentComposition(ctx, modules.asyncai, Object.keys(servers));
	const progress = [];
	const ran = await spa.runAgent(ctx, composition, servers, { prompt: "go", system: spa.systemPrompt(did), provider: "custom", endpoint: "https://inference.test", format: "anthropic", apiKey: "PAGE-KEY", model: "m" }, { progress: (r) => progress.push(r.length) });
	ctx.close?.();
	assert.equal(ran.error, undefined, ran.error);
	const r = ran.result;
	assert.equal(r.stop, "end_turn", r.error);
	assert.deepEqual(r.servers.map((s) => [s.name, s.phase]), [["code", "ready"], ["knotless", "ready"], ["marque", "ready"], ["fedproxy", "ready"], ["vm", "ready"]]);
	const results = Object.fromEntries(r.messages.at(-2).content.map((b) => [b.tool_use_id, b]));
	assert.deepEqual(JSON.parse(results.a.content), { stdout: "hi\n", result: { n: 391, domain: "example.com" } });
	assert.match(JSON.parse(results.w.content).result, /create_repo: code mode calls read-only tools only/);
	assert.match(results.b.content, /example\.com/);
	assert.equal(results.c.is_error, true);
	assert.deepEqual(asked.map((q) => q.tool), ["create_repo"]);
	// each in-process server on its own hook: code mode's, and com.fedfork.exec.mcp.<lib>
	assert.deepEqual([...new Set(ran.graph.map((g) => `${g.nsid} ${g.handler}`))].sort(), ["com.fedfork.exec.mcp.fedproxy fedproxy", "com.fedfork.exec.mcp.knotless knotless", "com.fedfork.exec.mcp.marque marque", "com.fedfork.exec.mcp.message code", "com.fedfork.exec.mcp.vm vm"]);
	// the key went only in the host's headers; the model saw every server's tools
	assert.equal(requests[0].headers["x-api-key"], "PAGE-KEY");
	assert.ok(requests[0].body.tools.some((t) => t.name === "fedproxy__list_keys"));
	assert.ok(!requests[0].body.tools.some((t) => t.name.startsWith("cocore__")));
	// no listing: every PDS read the run made is one record at a computed key, or marque's own list_domains
	assert.ok(!seen.some((s) => s.startsWith("GET com.atproto.repo.listRecords") && !s.endsWith("at.marque.domain")), seen.join("\n"));
	assert.equal(requests[0].body.model, "m");
	assert.ok(progress.length >= 2 && progress.at(-1) === spa.rows(r.messages).length);
});

// co/core's /models on 2026-11-10
const COCORE_MODELS = ["mlx-community/Qwen2.5-7B-Instruct-4bit", "mlx-community/gemma-4-26B-A4B-it-4bit", "mlx-community/Qwen3.5-4B-MLX-4bit", "google/gemma-4-12b", "mlx-community/gemma-3-4b-it-qat-4bit", "mlx-community/Ornith-1.0-9B-4bit", "mlx-community/Qwen2.5-32B-Instruct-4bit", "prism-ml/Ternary-Bonsai-27B-mlx-2bit", "stub"];

test("pickModel: the largest instruct model, else the largest other than stub", () => {
	assert.equal(spa.pickModel(COCORE_MODELS), "mlx-community/Qwen2.5-32B-Instruct-4bit");
	assert.equal(spa.pickModel(COCORE_MODELS.filter((id) => !id.includes("32B"))), "mlx-community/gemma-4-26B-A4B-it-4bit");
	assert.equal(spa.pickModel(["google/gemma-4-12b", "stub", "x/Big-70B"]), "x/Big-70B");
	assert.equal(spa.pickModel(["stub"]), undefined);
});

test("co/core, the default provider: the host mints its service-auth token with asyncai's session; the agent never sees it", { timeout: 120_000 }, async () => {
	const seen = [];
	const stub = pds(seen);
	const sent = [];
	const fetch = async (url, init = {}) => {
		const u = new URL(url);
		if (u.pathname.endsWith("com.atproto.server.getServiceAuth")) return (seen.push(`GET getServiceAuth ${u.searchParams.get("aud")} ${u.searchParams.get("lxm")}`), Response.json({ token: "SVC" }));
		if (u.href === "https://cocore.dev/api/v1/models") return Response.json({ object: "list", data: COCORE_MODELS.map((id) => ({ id, object: "model" })) });
		if (u.origin === "https://cocore.dev") return (sent.push({ headers: init.headers, body: JSON.parse(init.body) }), Response.json({ id: "c", object: "chat.completion", choices: [{ index: 0, message: { role: "assistant", content: "hello" }, finish_reason: "stop" }], usage: { prompt_tokens: 1, completion_tokens: 1 } }));
		return stub(url, init);
	};
	const { ctx, modules, policy, servers } = await setup({ fetch });
	const composition = await spa.agentComposition(ctx, modules.asyncai, Object.keys(servers));
	const host = { ctx, policy, session: { did, fetchHandler: (path, init) => fetch(new URL(path, PDS).href, init) } };
	// no model named: the largest instruct model co/core serves now
	const model = await spa.cocoreModel(host, modules);
	assert.equal(model, "mlx-community/Qwen2.5-32B-Instruct-4bit");
	const ran = await spa.runAgent(ctx, composition, servers, { prompt: "hi", system: "s", model }, { host });
	ctx.close?.();
	assert.equal(ran.error, undefined, ran.error);
	assert.equal(ran.result.stop, "end_turn", ran.result.error);
	assert.ok(seen.includes("GET getServiceAuth did:web:cocore.dev dev.cocore.inference.dispatch"), seen.join("\n"));
	assert.equal(new Headers(sent[0].headers).get("authorization"), "Bearer SVC");
	assert.equal(sent[0].body.model, model);
});

test("inference providers: co/core by service auth (no key), Anthropic and custom by host headers", async () => {
	const today = { actions: [{ entry: "run", fields: [{ key: "prompt" }] }] };
	const withProvider = { actions: [{ entry: "run", fields: [{ key: "prompt" }, { key: "provider" }] }] };
	assert.throws(() => spa.inference({}, today), /co\/core inference needs atlibs-asyncai's provider option/);
	assert.throws(() => spa.inference({ provider: "custom", endpoint: "https://x.test", format: "openai" }, today), /OpenAI-format/);
	assert.deepEqual(spa.inference({}, withProvider), {
		input: { endpoint: "https://cocore.dev/api/v1", model: "claude-opus-5-5", provider: "openai" },
		http: { "https://cocore.dev": { serviceAuth: { aud: "did:web:cocore.dev", lxm: "dev.cocore.inference.dispatch" } } },
	});
	assert.deepEqual(spa.inference({ provider: "anthropic", apiKey: "K" }, today).http, { "https://api.anthropic.com": { headers: { "x-api-key": "K", "anthropic-dangerous-direct-browser-access": "true" } } });
	assert.deepEqual(spa.inference({ provider: "custom", endpoint: "https://api.deepseek.com/anthropic", apiKey: "K", model: "d" }, today), { input: { endpoint: "https://api.deepseek.com/anthropic", model: "d" }, http: { "https://api.deepseek.com": { headers: { "x-api-key": "K" } } } });
	assert.deepEqual(spa.inference({ provider: "custom", endpoint: "https://o.test/v1", format: "openai", apiKey: "T" }, withProvider).http, { "https://o.test": { headers: { authorization: "Bearer T" } } });
	assert.throws(() => spa.inference({ provider: "custom" }, today), /base URL/);
});

const { mcpServer, streamableHttp, fetcher } = await import("./mcp-stub.mjs");
const { webrtcUrl } = await import("@publicdomainrelay/atlibs-webrtc/index.js");
const RECEIPT = { uri: "at://did:plc:bidder/com.publicdomainrelay.temp.market.receipt/3r", cid: "bafyreceipt" };

test("vm is a server: vm-mcp's four tools; vm_create asks first, then rents through the requester (its hook default, as you)", async () => {
	const seen = [];
	// the page's context: managed_records is the ledger release deletes by
	const ctx = createNodeContext({ id: "agent-spa-test", stateDir: await mkdtemp(join(tmpdir(), "agent-spa-")), storage: { managed_records: {} }, fetch: pds(seen) });
	const { modules, policy } = await spa.loadAll(ctx, ROOT, fileFetch);
	const asked = [];
	const observed = [];
	const vm = { attest: async ({ record } = {}) => (record ? { key: "did:key:zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme", record: { ...record, signatures: [{ $type: "network.attested.signature" }] } } : { key: "did:key:zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme" }), calls: {}, fetchers: {} };
	const servers = spa.mcpServers({ ctx, policy, modules, session: sessionOf(seen), vm, confirm: async (q) => (asked.push(q), true), observe: (...a) => observed.push(a) });
	const tools = (await rpc(servers.vm, "tools/list")).result.tools;
	assert.deepEqual(tools.map((t) => [t.name, t.annotations.readOnlyHint]), [["vm_create", false], ["vm_list", true], ["vm_exec", false], ["vm_release", false], ["vm_forget", false]]);
	assert.equal(tools[0].inputSchema.required, undefined, "vm_create needs no argument");
	// hostOptions: the vm lib and the hook defaults it names get the transport, the attestation key and the ledger
	for (const name of ["vm-mcp", "requester", "process-mcp"]) {
		const options = spa.hostOptions({ ctx, policy, session: sessionOf(seen), vm }, name);
		assert.deepEqual([options.attest, options.fetchers, options.ledger, options.budget], [vm.attest, vm.fetchers, ctx.storage.managed_records, { wallMs: 1_200_000, calls: Infinity, events: Infinity }], name);
	}
	assert.equal(spa.hostOptions({ ctx, policy, session: sessionOf(seen), vm }, "marque").attest, undefined);

	const out = await rpc(servers.vm, "tools/call", { name: "vm_create", arguments: { bidders: ["did:plc:bidder"] } });
	ctx.close?.();
	assert.deepEqual(asked.map((q) => [q.tool, q.arguments]), [["vm_create", { bidders: ["did:plc:bidder"] }]]);
	const [server, entry, args, ran] = observed[0];
	assert.deepEqual([server, entry, args.via], ["vm", "vm_create", "webrtc"]);
	// the kept list was written back before renting; the requester wrote the vm record as you, its user-data hook
	// default built the VM's cloud-init: the guest agent's MCP shell
	assert.ok(seen.indexOf("POST com.atproto.repo.putRecord") < seen.indexOf("POST com.atproto.repo.createRecord"), seen.join("\n"));
	const record = created.find((c) => c.collection === "com.publicdomainrelay.temp.compute.vm").record;
	assert.match(record.user_data, /^#cloud-config\n/);
	assert.match(record.user_data, /services\.json/);
	assert.match(record.user_data, /shell\.sock/);
	// the stub PDS has no bidder, so no VM: vm_create says why and what to change, and the requester deleted
	// the record it made (the ledger holds it; the stub PDS answers the delete RecordNotFound)
	assert.equal(out.result.isError, true);
	assert.match(out.result.content[0].text, /^no VM \(.*did:plc:bidder offerings.*\); nothing is kept\. Try vm_create again/);
	assert.equal(ran.error, out.result.content[0].text);
	assert.doesNotMatch(out.result.content[0].text, /not created by this module/);
	assert.ok(seen.includes("POST com.atproto.repo.deleteRecord"), seen.join("\n"));
	assert.ok(!seen.some((s) => s.includes("listRecords")), "never listRecords");
});

test("kept VMs: vm_list's are the list, vm_create adds one, vm_release drops it; the transport signals through requester runs", async () => {
	const good = webrtcUrl({ receipt: RECEIPT, bidder: "did:plc:bidder" }, "shell", "mcp");
	const one = { vm: "vm-3r", bidder: "did:plc:bidder", shell: good };
	const kept = spa.vmsAfter([], "vm", "vm_list", {}, { result: { vms: [one] } });
	assert.deepEqual(kept, [one]);
	const two = spa.vmsAfter(kept, "vm", "vm_create", {}, { result: { ...one, vm: "vm-4s" } });
	assert.deepEqual(two.map((v) => v.vm), ["vm-3r", "vm-4s"]);
	assert.equal(spa.vmsAfter(two, "vm", "vm_release", { vm: "vm-3r" }, { error: "nope" }), two);
	assert.deepEqual(spa.vmsAfter(two, "vm", "vm_release", { vm: "vm-3r" }, { result: {} }).map((v) => v.vm), ["vm-4s"]);
	assert.deepEqual(spa.vmsAfter(two, "vm", "vm_forget", { vm: "vm-4s" }, { result: { forgotten: true } }).map((v) => v.vm), ["vm-3r"]);
	assert.equal(spa.vmsAfter(two, "marque", "vm_release", { vm: "vm-3r" }, { result: {} }), two);

	// the fetchers send a kept VM's own ref (found by receipt URI), and refuse one no kept VM has, at once
	const miscopied = webrtcUrl({ receipt: { ...RECEIPT, cid: "bafyreceip" }, bidder: "did:plc:bidder" }, "shell", "mcp");
	const sent = [];
	const held = spa.keptFetchers({ "webrtc:": async (r) => (sent.push(r.url), { status: 200, headers: {}, body: "" }) }, () => kept);
	await held["webrtc:"]({ method: "POST", url: miscopied, body: "{}" });
	assert.deepEqual(sent, [good]);
	const stranger = webrtcUrl({ receipt: { uri: "at://did:plc:bidder/com.publicdomainrelay.temp.market.receipt/other", cid: "c" } }, "shell", "mcp");
	assert.throws(() => held["webrtc:"]({ method: "POST", url: stranger, body: "{}" }), /no VM kept here has this receipt.*kept: vm-3r/);

	const runs = [];
	const { events, signal } = spa.vmSignals(async (entry, input) => (runs.push([entry, input]), entry === "watch" ? { events: [{ kind: "k" }] } : { ok: true }));
	assert.deepEqual(await events({ receipt: RECEIPT, tag: "shell" }), [{ kind: "k" }]);
	await signal({ receipt: RECEIPT, tag: "shell", offer: "sealed" });
	assert.deepEqual(runs, [
		["watch", { bidder: "did:plc:bidder", receipt: RECEIPT.uri, tags: ["shell"], max_polls: 1 }],
		["signal", { receipt: RECEIPT, tag: "shell", offer: "sealed" }],
	]);
});

test("vm_exec runs a command on a kept VM: process-mcp's http.fetch to its shell, over the host's webrtc: fetcher", async () => {
	const seen = [];
	const good = webrtcUrl({ receipt: RECEIPT, bidder: "did:plc:bidder" }, "shell", "mcp");
	keptRecord = { uri: `at://${did}/computer.socialweb.vm.kept/self`, cid: "bafykept", value: { vms: [{ vm: "vm-3r", bidder: "did:plc:bidder", shell: good, receipt: RECEIPT, records: [] }] } };
	const shell = mcpServer({ exec: ({ command, timeoutMs }) => ({ content: [{ type: "text", text: JSON.stringify({ exitCode: 0, stdout: `ran ${command} in ${timeoutMs}\n`, stderr: "" }) }] }) });
	const transport = fetcher(streamableHttp(shell).handle);
	const ctx = createNodeContext({ id: "agent-spa-test", stateDir: await mkdtemp(join(tmpdir(), "agent-spa-")), storage: { managed_records: {} }, fetch: pds(seen) });
	const { modules, policy } = await spa.loadAll(ctx, ROOT, fileFetch);
	const vm = { attest: async () => ({ key: "did:key:zQ3shokFTS3brHcDQrn82RUDfCZESWL1ZdCEJwekUDPQiYBme" }), calls: {}, fetchers: spa.keptFetchers({ "webrtc:": transport.fetch }, () => keptRecord.value.vms) };
	const servers = spa.mcpServers({ ctx, policy, modules, session: sessionOf(seen), vm, confirm: async () => true });
	const out = await rpc(servers.vm, "tools/call", { name: "vm_exec", arguments: { vm: "vm-3r", command: "uname -a", timeout_s: 5 } });
	keptRecord = null;
	ctx.close?.();
	assert.deepEqual(out.result.structuredContent, { exit: 0, stdout: "ran uname -a in 5000\n", stderr: "" }, JSON.stringify(out.result));
	assert.ok(transport.urls.length > 0 && transport.urls.every((u) => u === good), transport.urls.join("\n"));
});

test("the agent connects a VM's MCP shell mid-run over the host's webrtc: fetcher, runs a command and follows a job", { timeout: 120_000 }, async () => {
	const url = webrtcUrl({ receipt: RECEIPT, bidder: "did:plc:bidder" }, "shell", "mcp");
	const job = { reads: 0 };
	const shell = mcpServer({
		exec: ({ command }) => ({ content: [{ type: "text", text: JSON.stringify({ exitCode: 0, stdout: `ran ${command}\n`, stderr: "", truncated: false }) }] }),
		start: () => ({ content: [{ type: "text", text: '{"id":"j1"}' }] }),
		read: ({ cursor = 0 }) => ({ content: [{ type: "text", text: JSON.stringify(++job.reads < 2 ? { output: "tick 1\n", cursor: 7, done: false } : { output: "tick 2\ntick 3\n", cursor: 21, done: true, exitCode: 0, from: cursor }) }] }),
		write: () => ({ content: [{ type: "text", text: "{}" }] }),
		kill: () => ({ content: [{ type: "text", text: "{}" }] }),
		list: () => ({ content: [{ type: "text", text: '{"jobs":[]}' }] }),
	});
	const transport = fetcher(streamableHttp(shell).handle);
	const turns = [
		{ content: [{ type: "tool_use", id: "c", name: "mcp_connect", input: { name: "shell", server: { url } } }], stop_reason: "tool_use" },
		{ content: [{ type: "tool_use", id: "e", name: "shell__exec", input: { command: "uname -a; id -u" } }, { type: "tool_use", id: "s", name: "shell__start", input: { command: "for i in 1 2 3; do echo tick $i; sleep 2; done" } }], stop_reason: "tool_use" },
		{ content: [{ type: "tool_use", id: "r1", name: "shell__read", input: { id: "j1", cursor: 0, waitMs: 5000 } }], stop_reason: "tool_use" },
		{ content: [{ type: "tool_use", id: "r2", name: "shell__read", input: { id: "j1", cursor: 7, waitMs: 5000 } }], stop_reason: "tool_use" },
		{ content: [{ type: "text", text: "done" }], stop_reason: "end_turn" },
	];
	const bodies = [];
	const seen = [];
	const stub = pds(seen);
	const fetch = async (u, init) => (u.startsWith("https://inference.test/") ? (bodies.push(JSON.parse(init.body)), Response.json({ ...turns.shift(), usage: { input_tokens: 1, output_tokens: 1 } })) : stub(u, init));
	const { ctx, modules, servers } = await setup({ fetch });
	const composition = await spa.agentComposition(ctx, modules.asyncai, Object.keys(servers));
	const ran = await spa.runAgent(ctx, composition, servers, { prompt: "rent a vm", system: spa.systemPrompt(did), provider: "custom", endpoint: "https://inference.test", format: "anthropic", apiKey: "K", model: "m" }, { fetchers: { "webrtc:": transport.fetch } });
	ctx.close?.();
	assert.equal(ran.error, undefined, ran.error);
	const r = ran.result;
	assert.equal(r.stop, "end_turn", r.error);
	const vm = r.servers.find((s) => s.name === "shell");
	assert.deepEqual([vm.phase, vm.dynamic, vm.tools.length], ["ready", true, 6]);
	assert.ok(transport.urls.length >= 5 && transport.urls.every((u) => u === url), transport.urls.join("\n"));
	assert.ok(bodies[0].tools.some((t) => t.name === "mcp_connect"));
	assert.ok(bodies[1].tools.some((t) => t.name === "shell__read"));
	const results = Object.fromEntries(r.messages.flatMap((m) => (Array.isArray(m.content) ? m.content : [])).filter((b) => b.type === "tool_result").map((b) => [b.tool_use_id, b]));
	assert.match(JSON.stringify(results.e.content), /ran uname -a; id -u/);
	assert.match(JSON.stringify(results.r2.content), /\\"done\\":true/);
	assert.deepEqual(spa.connectedOf(spa.rows(r.messages)).map((s) => [s.name, s.url, s.tools.length]), [["shell", url, 6]]);
});

test("rows: OpenAI chat completions transcripts too", () => {
	const list = spa.rows([
		{ role: "user", content: "go" },
		{ role: "assistant", content: null, tool_calls: [{ id: "t1", type: "function", function: { name: "mcp_connect", arguments: '{"name":"vm","server":{"url":"webrtc:///x/shell/mcp"}}' } }] },
		{ role: "tool", tool_call_id: "t1", content: '{"connected":"vm","tools":["vm__exec"]}' },
	]);
	assert.deepEqual(list.map((r) => [r.kind, r.id ?? null]), [["text", null], ["tool_use", "t1"], ["tool_result", "t1"]]);
	assert.deepEqual(spa.connectedOf(list), [{ name: "vm", url: "webrtc:///x/shell/mcp", tools: ["vm__exec"] }]);
});

test("co/core is offered on localhost origins only (its CORS); elsewhere custom endpoints alone", () => {
	for (const host of ["localhost", "127.0.0.1", "[::1]"]) assert.equal(spa.cocoreUsable(host), true, host);
	for (const host of ["atlibs-agent.socialweb.computer", "example.com", "127.0.0.2"]) assert.equal(spa.cocoreUsable(host), false, host);
});