Fix #5: Workers AI streamed tool calls, Node.js module stubs for named imports; p11f queue hand-over timing

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/<UTC minute>/…): 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

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<Uint8Array, Uint8Array> {
   const decoder = new TextDecoder();
   const encoder = new TextEncoder();
   let buffer = "";
   let done = false;
+  const calls = new Map<number, { name: string; arguments: string }>();
   const emit = (c: TransformStreamDefaultController<Uint8Array>, data: string) =>
     c.enqueue(encoder.encode(`data: ${data}\n\n`));
+  const finish = (c: TransformStreamDefaultController<Uint8Array>) => {
+    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<Uint8Array>, 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<Uint8Array, Uint8Array> {
       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<Uint8Array, Uint8Array> {
     flush(c) {
       buffer += decoder.decode();
       if (buffer) line(c, buffer.trim());
-      if (!done) emit(c, "[DONE]");
+      finish(c);
     },
   });
 }

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};`;
 

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"),
+    });
+  });
+});