diff --git a/packages/cellb/e2e/p11f-celld-one-bucket.test.ts b/packages/cellb/e2e/p11f-celld-one-bucket.test.ts index ca8abe9..8886cae 100644 --- a/packages/cellb/e2e/p11f-celld-one-bucket.test.ts +++ b/packages/cellb/e2e/p11f-celld-one-bucket.test.ts @@ -529,8 +529,11 @@ describe.skipIf(!CELLD)( await w.close(); celld = await celldDev(dir); try { + // celld wakes cells from its wake index by minute (wake/entries//…): a + // message due in a later minute than celld started in is woken at its next scan, up + // to a minute on. So up to 80 s here; within one minute it is a few seconds. let got: unknown[] = []; - for (let i = 0; i < 40 && got.length < 2; i++) { + for (let i = 0; i < 160 && got.length < 2; i++) { await new Promise((r) => setTimeout(r, 500)); got = await (await fetch(`${celld.base}/`)).json(); } @@ -559,7 +562,7 @@ describe.skipIf(!CELLD)( it( "Queues: a delayed message sent on cellb is delivered by celld", () => handoff("cellb→celld"), - 120_000, + 150_000, // celld may take up to a minute to wake the queue (above) ); // A workflow instance is a cell (celld's own workflow code on both sides): a sleeping or diff --git a/packages/cellb/src/bindings/workers-ai.ts b/packages/cellb/src/bindings/workers-ai.ts index 3584106..1da40b1 100644 --- a/packages/cellb/src/bindings/workers-ai.ts +++ b/packages/cellb/src/bindings/workers-ai.ts @@ -119,6 +119,8 @@ export function workersAi(opts: WorkersAiOptions, fetchImpl: typeof fetch = (r, } interface OpenAiToolCall { + /** In a streamed chunk: which call a fragment belongs to. */ + index?: number; id?: string; function?: { name?: string; arguments?: string }; } @@ -137,23 +139,33 @@ function toWorkersAiToolCall(tc: OpenAiToolCall) { /** * OpenAI chat-completion chunks (`data: {"choices":[{"delta":…}]}`) → Workers AI's stream - * (`data: {"response":"…"}`, then `data: [DONE]`); usage, and tool calls when a chunk carries - * them, ride along as Workers AI sends them. + * (`data: {"response":"…"}`, then `data: [DONE]`), usage riding along. Tool calls stream in + * fragments (a name, then pieces of the arguments' JSON, keyed by `index`): they are gathered and + * sent whole, once, before `[DONE]`, as `{ response: "", tool_calls: [{ name, arguments }] }`. */ function toWorkersAiStream(): TransformStream { const decoder = new TextDecoder(); const encoder = new TextEncoder(); let buffer = ""; let done = false; + const calls = new Map(); const emit = (c: TransformStreamDefaultController, data: string) => c.enqueue(encoder.encode(`data: ${data}\n\n`)); + const finish = (c: TransformStreamDefaultController) => { + if (done) return; + done = true; + if (calls.size) { + const toolCalls = [...calls.entries()] + .sort(([a], [b]) => a - b) + .map(([, call]) => toWorkersAiToolCall({ function: call })); + emit(c, JSON.stringify({ response: "", tool_calls: toolCalls })); + } + emit(c, "[DONE]"); + }; const line = (c: TransformStreamDefaultController, l: string) => { if (!l.startsWith("data:")) return; const data = l.slice(5).trim(); - if (data === "[DONE]") { - done = true; - return emit(c, "[DONE]"); - } + if (data === "[DONE]") return finish(c); let chunk: { choices?: { delta?: { content?: string | null; tool_calls?: OpenAiToolCall[] } }[]; usage?: unknown; @@ -164,10 +176,16 @@ function toWorkersAiStream(): TransformStream { return; } const delta = chunk.choices?.[0]?.delta; + for (const [i, tc] of (delta?.tool_calls ?? []).entries()) { + const index = tc.index ?? i; + const call = calls.get(index) ?? { name: "", arguments: "" }; + call.name += tc.function?.name ?? ""; + call.arguments += tc.function?.arguments ?? ""; + calls.set(index, call); + } const out: Json = { response: delta?.content ?? "" }; - if (delta?.tool_calls?.length) out.tool_calls = delta.tool_calls.map(toWorkersAiToolCall); if (chunk.usage) out.usage = chunk.usage; - if (out.response === "" && !out.tool_calls && !out.usage) return; + if (out.response === "" && !out.usage) return; emit(c, JSON.stringify(out)); }; return new TransformStream({ @@ -183,7 +201,7 @@ function toWorkersAiStream(): TransformStream { flush(c) { buffer += decoder.decode(); if (buffer) line(c, buffer.trim()); - if (!done) emit(c, "[DONE]"); + finish(c); }, }); } diff --git a/packages/cellb/src/project/browser-bundle.ts b/packages/cellb/src/project/browser-bundle.ts index 87f6cff..5abcf16 100644 --- a/packages/cellb/src/project/browser-bundle.ts +++ b/packages/cellb/src/project/browser-bundle.ts @@ -16,20 +16,34 @@ const pkg = resolve(import.meta.dirname, "../.."); const requireFromPkg = createRequire(resolve(pkg, "package.json")); /** - * A Node.js module with no browser polyfill, as CommonJS so that any named import binds: each - * export is a function (a class too) that throws when called, the module's default is the same. + * A Node.js module with no browser polyfill, as an ES module: every export the module has on + * this Node.js (the bundler runs on it) is a function that throws when called (a class to extend + * too), and the default export answers any member the same way. */ -const nodeStub = (name: string) => ` -const fail = (what) => { throw new Error(${JSON.stringify(name)} + (what ? "." + what : "") + " is not available in browsers (nodejs_compat)"); }; -const exports_ = {}; -module.exports = new Proxy(exports_, { - get(target, key) { - if (typeof key === "symbol" || key === "__esModule" || key === "then") return undefined; - if (!(key in target)) target[key] = function () { fail(String(key)); }; - return target[key]; - }, -}); -`; +function nodeStub(name: string): string { + let names: string[] = []; + try { + names = Object.keys(requireFromPkg(name)).filter((n) => n !== "default"); + } catch { + // not a module of this Node.js: the default export only + } + const what = JSON.stringify(name); + return [ + `const fail = (key) => { throw new Error(${what} + "." + key + " is not available in browsers (nodejs_compat)"); };`, + "const stub = (key) => function () { fail(key); };", + ...names.map( + (n, i) => + `const s${i} = stub(${JSON.stringify(n)}); export { s${i} as ${JSON.stringify(n)} };`, + ), + "export default new Proxy(Object.create(null), {", + " get(target, key) {", + ' if (typeof key === "symbol" || key === "then" || key === "__esModule") return undefined;', + " if (!Object.hasOwn(target, key)) target[key] = stub(String(key));", + " return target[key];", + " },", + "});", + ].join("\n"); +} const GLOBALS_SHIM = `${WORKERS_GLOBAL_NAMES.map((n) => `export const ${n} = globalThis.__cellsGlobals.${n};`).join(" ")} export const scheduler = ${SCHEDULER_EXPRESSION};`; diff --git a/packages/cellb/test/p14-workers-ai.test.ts b/packages/cellb/test/p14-workers-ai.test.ts index ce505ef..82f9a80 100644 --- a/packages/cellb/test/p14-workers-ai.test.ts +++ b/packages/cellb/test/p14-workers-ai.test.ts @@ -5,7 +5,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { MemFilesApi } from "@statewalker/webrun-files-mem"; import { afterAll, beforeAll, describe, expect, it } from "vitest"; -import { loadNodeProject, nodeSqliteOpener } from "../src/node.js"; +import { bundleWorkerForBrowser, loadNodeProject, nodeSqliteOpener } from "../src/node.js"; // Workers AI (`env.AI`) served by an OpenAI-compatible endpoint: a fake one here, which echoes // what it was asked and records every request. @@ -28,6 +28,23 @@ async function fakeOpenAi() { ); } const last = body.messages.at(-1).content; + if (body.tools?.length && last === "weather?" && body.stream) { + // As OpenAI-compatible servers stream tool calls: the name first, then the arguments in + // fragments, each keyed by the call's index. + res.writeHead(200, { "content-type": "text/event-stream" }); + const chunk = (delta: unknown, finish: string | null = null) => + res.write(`data: ${JSON.stringify({ choices: [{ delta, finish_reason: finish }] })}\n\n`); + chunk({ + tool_calls: [{ index: 0, id: "c1", function: { name: "weather", arguments: "" } }], + }); + for (const part of ['{"ci', 'ty":"Pa', 'ris"}']) + chunk({ tool_calls: [{ index: 0, function: { arguments: part } }] }); + chunk({ + tool_calls: [{ index: 1, id: "c2", function: { name: "time", arguments: "{}" } }], + }); + chunk({}, "tool_calls"); + return res.end("data: [DONE]\n\n"); + } if (body.tools?.length && last === "weather?") { res.writeHead(200, { "content-type": "application/json" }); return res.end( @@ -94,6 +111,11 @@ export default { tools: [{ name: "weather", description: "Weather", parameters: { type: "object" } }], }), stream: await read(await ai.run("@cf/meta/llama-3.1-8b-instruct", { prompt: "go", stream: true })), + streamedTools: await read(await ai.run("@cf/meta/llama-3.1-8b-instruct", { + messages: [{ role: "user", content: "weather?" }], + tools: [{ name: "weather", description: "Weather", parameters: { type: "object" } }], + stream: true, + })), raw: { status: raw.status, type: raw.headers.get("content-type"), lines: await read(raw.body) }, embed: await ai.run("@cf/baai/bge-small-en-v1.5", { text: ["ab", "abcd"] }), refused, @@ -144,6 +166,11 @@ describe("P14 Workers AI over an OpenAI-compatible endpoint", () => { 'data: {"response":"go "}', "data: [DONE]", ]); + // Tool calls stream in fragments; Workers AI's stream carries the whole calls, once. + expect(r.streamedTools).toEqual([ + 'data: {"response":"","tool_calls":[{"name":"weather","arguments":{"city":"Paris"}},{"name":"time","arguments":{}}]}', + "data: [DONE]", + ]); // The endpoint's own response, as Flue's provider reads it. expect(r.raw.status).toBe(200); expect(r.raw.type).toBe("text/event-stream"); @@ -174,3 +201,41 @@ describe("P14 Workers AI over an OpenAI-compatible endpoint", () => { } }); }); + +describe("P14 Node.js modules a browser bundle has no polyfill for", () => { + it("load as stubs whose named and default exports throw when called", async () => { + const dir = mkdtempSync(join(tmpdir(), "p14-stub-")); + writeFileSync( + join(dir, "index.js"), + `import { spawn, ChildProcess } from "node:child_process"; + import * as cp from "node:child_process"; + import readline from "node:readline"; + class Mine extends ChildProcess {} // a class to extend, as libraries do at load + export const probe = () => { + const out = {}; + const calls = { + spawn, + namespace: () => cp.exec(), + defaultMember: () => readline.createInterface(), + toString: () => readline.toString(), // Object.prototype's names are stubs too + subclass: () => new Mine(), + }; + for (const [k, f] of Object.entries(calls)) + try { f(); out[k] = "ran"; } catch (e) { out[k] = e.message; } + return out; + };`, + ); + const code = await bundleWorkerForBrowser(join(dir, "index.js"), { nodeCompat: true }); + const g = globalThis as { __cellsGlobals?: object }; + g.__cellsGlobals ??= {}; // the Workers globals a browser host installs; this code reads none + const mod = await import(`data:text/javascript;base64,${Buffer.from(code).toString("base64")}`); + const fails = (key: string) => `${key} is not available in browsers (nodejs_compat)`; + expect(mod.probe()).toEqual({ + spawn: fails("node:child_process.spawn"), + namespace: fails("node:child_process.exec"), + defaultMember: fails("node:readline.createInterface"), + toString: fails("node:readline.toString"), + subclass: fails("node:child_process.ChildProcess"), + }); + }); +});