1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
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);
}
});
