// 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. 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); });