From b2ab892fa1d7df368b3fde471e45c5d9dde83f42 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 27 Oct 2026 07:15:09 +0000 Subject: [PATCH 1/2] Workers AI streamed tool calls whole; Node.js module stubs that bind named imports MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two defects in #5: - env.AI.run(…, { stream: true, tools }) passed each OpenAI chunk on by itself, so a tool call streamed in fragments (a name, then pieces of the arguments' JSON, keyed by index) came out as several half calls, with empty names and unparsed fragments. The stream now gathers the fragments by index and sends the whole calls once, before [DONE], as { response: "", tool_calls: [{ name, arguments }] }. - The stubs for Node.js modules without a browser polyfill were CommonJS whose members were made on first read; esbuild copies a CommonJS module's own properties once, so named imports (`import { spawn } from "node:child_process"`) were undefined, and only `mod.member` threw as meant. The stub is now an ES module exporting a throwing function for every export the module has on the bundling Node.js (classes extend them), with a default export that answers any member (Object.prototype's names included) the same way. Tests (p14): streamed tool-call fragments become the whole calls; a bundle importing node:child_process and node:readline gets throwing named, namespace and default members, and extends a stubbed class at load. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01SXJQA3LbggJHi5qZBpUKe8 --- packages/cellb/src/bindings/workers-ai.ts | 36 ++++++++--- packages/cellb/src/project/browser-bundle.ts | 40 ++++++++---- packages/cellb/test/p14-workers-ai.test.ts | 67 +++++++++++++++++++- 3 files changed, 120 insertions(+), 23 deletions(-) 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"), + }); + }); +}); From 5c5f868dd652ed96e94ae427837cd18969d77be0 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 27 Oct 2026 07:33:06 +0000 Subject: [PATCH 2/2] p11f: give celld up to a minute to wake a queue it took over MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit "Queues: a delayed message sent on cellb is delivered by celld" failed now and then in CI (#5, #7). 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, and the test polled for 20 s. Reproduced every time by sending at :54 of a minute (delivered 59 s after celld started); cellb writes the same entry celld writes for its own delayed messages. The test now polls for up to 80 s (timeout 150 s). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01SXJQA3LbggJHi5qZBpUKe8 --- packages/cellb/e2e/p11f-celld-one-bucket.test.ts | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) 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