Files
fx/sdk/tests/test-packed-example.mjs
T
Pranit b0d83eb894 Make libfx sessions durable
- `createFxAgent()` returns the agent at once instead of a promise, and does no I/O until a session runs a turn.
- `agent.session(id)` opens a session, and `session.prompt()` returns a turn view with `accepted`, `result`, async iteration over its events, and an NDJSON `readable`.
- Sessions are durable by default: `vercel()` on Vercel, `local()` files in Node.js elsewhere, and `memory()` in browsers. The `durability` option picks one.
- `libfx/durable-local` and `libfx/durable-vercel` bundle their World, so an app installs no other package.
- A turn whose process crashes, redeploys, or reaches its function deadline continues from its last record in the next process that gets work for the session.
- On Vercel, a turn stops at its next model request 30 seconds before the function deadline and continues in a new invocation from the same queue message.
- A model request or tool call still running 2 seconds before the deadline is cut off, and `onEvent` receives `session.deadline`.
- One process runs a session's turns at a time, and a prompt for a busy session queues behind the running turn.
- A process that froze and was replaced stops at its next write and never starts a tool with effects, and `onEvent` receives `session.fenced`.
- After an interruption, a call to an `idempotent: true` tool runs again. Any other call never reruns, and the model gets an error saying it may have partly run.
- `executionId` is the same each time a call runs, including after a crash.
- A prompt sent again with the same `messageId` starts nothing new: its view follows the original turn, or resolves with `repeated: true`.
- `session.stream(cursor)` replays a session's events from any cursor on any server, with `turn_start`, `turn_end`, `turn_resume`, and `turn_yield` lines.
- Every reader of a session's stream shows the same lines, because lines a replaced worker lands late are hidden.
- `session.steer()` and `session.cancel()` reach the session's running turn, and `session.resume()` continues a turn a stopped process left open.
- Every turn end and yield saves a checkpoint, so continuing a session reads one checkpoint and the records after it.
- Checkpoints record the libfx version, tool set, and model. A session or checkpoint saved under different ones still loads, and `onEvent` receives `checkpoint.mismatch`.
- `session.checkpoint()` exports a session's history, and the `checkpoint` option starts a new agent's session from it.
- `agent.wakeHandler()` is the queue delivery route an app mounts on Vercel and subscribes to `__libfx_wkf_workflow_session` in `vercel.json`.
- On Vercel, libfx authenticates to the AI Gateway with the deployment's OIDC token when neither `apiKey` nor `AI_GATEWAY_API_KEY` is set.
- `createFxEngine()` accepts a `persistence` store, and `createMemoryPersistence()` provides one in memory. A store that rejects a stale append with `FxFencedError` stops a second writer.
- A turn whose store append fails stops at once with `FX_JOURNAL_APPEND_FAILED`, and the agent refuses later prompts.
2026-10-07 01:47:00 -04:00

143 lines
7.1 KiB
JavaScript

#!/usr/bin/env node
import { strict as assert } from "node:assert";
import { createServer } from "node:http";
import { spawnSync } from "node:child_process";
import { cp, mkdir, mkdtemp, readdir, rm, writeFile } from "node:fs/promises";
import { tmpdir } from "node:os";
import { createRequire } from "node:module";
import { resolve } from "node:path";
import { fileURLToPath } from "node:url";
const backend = process.argv[3] || "native";
const format = process.argv[4] || "esm";
if (process.argv[2] !== "--installed") {
const temp = await mkdtemp(resolve(tmpdir(), "libfx-packed-"));
function run(command, args, cwd) {
const result = spawnSync(command, args, { cwd, encoding: "utf8", timeout: 120_000 });
if (result.error) throw result.error;
assert.equal(result.status, 0, `${result.stdout}\n${result.stderr}`);
return result.stdout;
}
try {
const input = resolve(process.argv[2]);
const consumer = resolve(temp, "consumer");
await mkdir(consumer);
await writeFile(resolve(consumer, "package.json"), '{"private":true,"type":"module"}\n');
const archive = input.endsWith(".tgz") ? input : resolve(temp,
JSON.parse(run("npm", ["pack", input, "--ignore-scripts", "--pack-destination", temp, "--json"], temp))[0].filename);
run("npm", ["install", "--ignore-scripts", "--no-audit", "--no-fund", archive], consumer);
await cp(fileURLToPath(import.meta.url), resolve(consumer, "example.mjs"));
console.log(run(process.execPath, [...process.execArgv, "example.mjs", "--installed", backend, format], consumer).trim());
} finally {
await rm(temp, { recursive: true, force: true });
}
} else {
const { createFxAgent, createFxEngine, createFxTerminal, getBackendInfo, memory } = format === "cjs" ? createRequire(import.meta.url)("libfx") : await import("libfx");
const { createMcpAdapter } = await import("libfx/mcp");
const { createSkillsAdapter } = await import("libfx/skills");
assert.equal(typeof createMcpAdapter, "function");
assert.equal(typeof createSkillsAdapter, "function");
for (const makeOptions of [
(key, value) => ({ [key]: value }),
(key, value) => Object.create({ [key]: value }),
(key, value) => Object.defineProperty({}, key, { value }),
]) {
for (const invalid of [null, false, 0, "", "other"]) {
await assert.rejects(getBackendInfo(makeOptions("backend", invalid)), TypeError);
await assert.rejects(getBackendInfo(makeOptions("surface", invalid)), TypeError);
const backendError = { name: "TypeError", message: 'backend must be "auto", "native", or "wasm"' };
await assert.rejects(createFxEngine(makeOptions("backend", invalid)), backendError);
await assert.rejects(createFxTerminal(makeOptions("backend", invalid)), backendError);
}
for (const value of [undefined, "auto", "native", "wasm"]) {
assert.deepEqual(await getBackendInfo(makeOptions("backend", value)), await getBackendInfo({ backend: value }));
}
assert.deepEqual(await getBackendInfo(makeOptions("surface", undefined)), await getBackendInfo());
assert.deepEqual(await getBackendInfo(makeOptions("nativeAddon", false)), await getBackendInfo({ nativeAddon: false }));
}
assert.deepEqual(await getBackendInfo({ backend: undefined, surface: undefined }), await getBackendInfo());
assert.deepEqual(await getBackendInfo({ backend: "auto", surface: "agent" }), await getBackendInfo());
for (const select of [getBackendInfo, createFxEngine, createFxTerminal]) {
let reads = 0;
const options = { backend: "native", get nativeAddon() { reads++; this.backend = null; } };
await assert.rejects(select(options), { name: "TypeError", message: 'backend must be "auto", "native", or "wasm"' });
assert.equal(reads, 1);
}
let requestedAuthorization;
let requestedModel;
const server = createServer((request, response) => {
request.resume();
request.on("end", () => {
if (request.method === "GET") {
response.writeHead(200, { "content-type": "application/json" });
response.end('{"object":"list","data":[]}');
return;
}
requestedAuthorization = request.headers.authorization;
requestedModel = request.headers["ai-language-model-id"];
response.writeHead(200, { "content-type": "text/event-stream" });
response.end('data: {"type":"text-delta","delta":"packed"}\n\ndata: {"type":"finish","finishReason":{"unified":"stop","raw":"stop"},"usage":{"inputTokens":{"total":1},"outputTokens":{"total":1}}}\n\ndata: [DONE]\n\n');
});
});
await new Promise((resolveListen) => server.listen(0, "127.0.0.1", resolveListen));
const agent = await createFxEngine({
backend,
fetch(input, init) {
const origin = `http://127.0.0.1:${server.address().port}`;
const url = init?.method === "GET" ? `${origin}/models` : String(input);
assert.equal(new URL(url).origin, origin);
return fetch(url, init);
},
apiKey: "packed-key",
gatewayChatUrl: `http://127.0.0.1:${server.address().port}/chat`,
model: "packed/model",
});
try {
assert.deepEqual(Object.keys(agent).sort(), ["checkpoint", "close", "followUp", "prompt", "resume", "sessionId"]);
const turn = agent.prompt("hello");
let text = "";
for await (const event of turn) if (event.type === "text_delta") text += event.delta;
assert.equal(text, "packed");
assert.deepEqual(await turn.result, { stopReason: "end_turn", usage: { inputTokens: 1, outputTokens: 1 } });
assert.equal(requestedAuthorization, "Bearer packed-key");
assert.equal(requestedModel, "packed/model");
assert.ok((await agent.checkpoint()).length > 48);
await agent.close();
// A durable agent from the package: the local World it bundles keeps the
// session in files, then memory() opts out.
const sessionsDir = await mkdtemp(resolve(tmpdir(), "libfx-packed-sessions-"));
process.env.FX_SESSIONS_DIR = sessionsDir;
try {
for (const durability of [undefined, memory()]) {
const durable = createFxAgent({
backend,
fetch: (input, init) => fetch(init?.method === "GET" ? `http://127.0.0.1:${server.address().port}/models` : String(input), init),
apiKey: "packed-key",
gatewayChatUrl: `http://127.0.0.1:${server.address().port}/chat`,
model: "packed/model",
...(durability ? { durability } : {}),
});
const durableTurn = durable.session().prompt("hello");
let durableText = "";
for await (const event of durableTurn) if (event.type === "text_delta") durableText += event.delta;
assert.equal(durableText, "packed");
assert.equal((await durableTurn.result).stopReason, "end_turn");
await durable.close();
}
assert.ok((await readdir(sessionsDir)).includes("runs"), "the default durability kept the session on disk");
} finally {
delete process.env.FX_SESSIONS_DIR;
await rm(sessionsDir, { recursive: true, force: true });
}
console.log(`${format} ${backend} packed libfx example passed`);
} finally {
await agent.close().catch(() => {});
server.closeAllConnections();
await new Promise((resolveClose) => server.close(resolveClose));
}
}