History

fix(runtime-dynamic-worker): preserve Blob/Uint8Array across the dispatcher boundary (#483)

The sandbox→host RPC bridge JSON.stringify'd every tool argument before crossing the Worker RPC boundary, and JSON.parse'd it on the host side. Binary values were silently destroyed: `new Blob([...])` arrived as `{}`, and `new Uint8Array([...])` arrived as a numeric-keyed object. Multipart upload tools (OpenAI Files API, anything with a `multipart/form-data` request body) then produced malformed parts and upstream servers returned 400. Switches the dispatcher from `(path: string, argsJson: string) => string` to `(path: string, args: unknown) => WorkerRpcResponse`, so structured clone carries `Uint8Array` / `ArrayBuffer` natively. Workers RPC excludes `Blob` / `File` from its allow-list, so we add a small codec that walks args and results in both directions and round-trips them as tagged ArrayBuffer envelopes (`{ __executorBinary: 1, kind, type, name?, lastModified?, buffer }`). The same codec runs symmetrically on tool results so a tool returning a Blob (e.g. file downloads) survives the host→sandbox hop. Adds repro tests for both directions: - Uint8Array arg survives intact - Blob arg survives with type - Uint8Array result returns to user code - Blob result returns with type and bytes Existing dispatcher tests update from the JSON-string contract to the structured-value contract; nothing outside this package consumed the dropped `encodeWorkerRpcResponse` / `decodeWorkerRpcResponse` helpers.

Rhys Sullivan committed 6 months ago

Commit 292b6331 parent 88b493c3 files · +291 −32Browse files

Unified · Split

3 files
  1. packages/kernel/runtime-dynamic-worker/src/executor.ts +114 −17
  2. packages/kernel/runtime-dynamic-worker/src/invocation.test.ts +113 −11
  3. packages/kernel/runtime-dynamic-worker/src/module-template.ts +64 −4
packages/kernel/runtime-dynamic-worker/src/executor.ts+114−17
@@ -136,7 +136,12 @@ const renderTransportMessage = (value: unknown): string => { · 135 unmodified lines
136136 return value.message;
137137 }
138138
139− if (typeof value === "object" && value !== null && "message" in value && typeof value.message === "string") {
139+ if (
140+ typeof value === "object" &&
141+ value !== null &&
142+ "message" in value &&
143+ typeof value.message === "string"
144+ ) {
140145 return value.message;
141146 }
142147
@@ -156,8 +161,12 @@ const renderTransportMessage = (value: unknown): string => { · 13 unmodified lines
156161 };
157162
158163 export const serializeWorkerCause = (cause: Cause.Cause<unknown>): SerializedWorkerError => {
159− const failures = cause.reasons.filter(Cause.isFailReason).map((reason) => serializeWorkerErrorValue(reason.error));
160− const defects = cause.reasons.filter(Cause.isDieReason).map((reason) => serializeWorkerErrorValue(reason.defect));
164+ const failures = cause.reasons
165+ .filter(Cause.isFailReason)
166+ .map((reason) => serializeWorkerErrorValue(reason.error));
167+ const defects = cause.reasons
168+ .filter(Cause.isDieReason)
169+ .map((reason) => serializeWorkerErrorValue(reason.defect));
161170 const interrupted = cause.reasons.some(Cause.isInterruptReason);
162171 const primary = failures[0] ?? defects[0] ?? null;
163172 const kind =
@@ -215,10 +224,93 @@ export const renderWorkerError = (error: SerializedWorkerError): string => { · 51 unmodified lines
215224 return error.message;
216225 };
217226
218−const encodeWorkerRpcResponse = (response: WorkerRpcResponse): string => JSON.stringify(response);
227+export type { WorkerRpcResponse };
228+
229+// ---------------------------------------------------------------------------
230+// Blob/File codec (both directions across the dispatcher boundary)
231+//
232+// Workers RPC's structured-clone allow-list excludes `Blob` / `File`, so
233+// we encode them to a tagged ArrayBuffer envelope and rehydrate on the
234+// far side. Symmetric in both directions: sandbox encodes args + host
235+// rehydrates them; host encodes result + sandbox rehydrates it. The
236+// matching encoder lives inside `module-template.ts` because it runs in
237+// the dynamic Worker isolate. `ArrayBuffer` / typed arrays / primitives
238+// cross structured clone natively.
239+// ---------------------------------------------------------------------------
240+
241+type BinaryEnvelope = {
242+ readonly __executorBinary: 1;
243+ readonly kind: "blob" | "file";
244+ readonly type: string;
245+ readonly name?: string;
246+ readonly lastModified?: number;
247+ readonly buffer: ArrayBuffer;
248+};
249+
250+const isBinaryEnvelope = (value: unknown): value is BinaryEnvelope =>
251+ typeof value === "object" &&
252+ value !== null &&
253+ (value as { __executorBinary?: unknown }).__executorBinary === 1 &&
254+ (value as { buffer?: unknown }).buffer instanceof ArrayBuffer &&
255+ typeof (value as { type?: unknown }).type === "string";
256+
257+const isPlainObject = (value: object): boolean => {
258+ const proto = Object.getPrototypeOf(value);
259+ return proto === Object.prototype || proto === null;
260+};
261+
262+const rehydrateBinary = (value: unknown): unknown => {
263+ if (value === null || typeof value !== "object") return value;
264+ if (value instanceof ArrayBuffer || ArrayBuffer.isView(value)) return value;
265+ if (isBinaryEnvelope(value)) {
266+ if (value.kind === "file" && typeof value.name === "string") {
267+ return new File([value.buffer], value.name, {
268+ type: value.type,
269+ ...(typeof value.lastModified === "number" ? { lastModified: value.lastModified } : {}),
270+ });
271+ }
272+ return new Blob([value.buffer], { type: value.type });
273+ }
274+ if (Array.isArray(value)) return value.map(rehydrateBinary);
275+ if (!isPlainObject(value)) return value;
276+ const out: Record<string, unknown> = {};
277+ for (const [k, v] of Object.entries(value as Record<string, unknown>)) {
278+ out[k] = rehydrateBinary(v);
279+ }
280+ return out;
281+};
219282
220−export const decodeWorkerRpcResponse = (raw: string): WorkerRpcResponse =>
221− JSON.parse(raw) as WorkerRpcResponse;
283+// Async because `Blob.arrayBuffer()` is async. Used on tool results before
284+// the dispatcher hands them back to the sandbox.
285+const encodeBinary = async (value: unknown): Promise<unknown> => {
286+ if (value === null || typeof value !== "object") return value;
287+ if (value instanceof ArrayBuffer || ArrayBuffer.isView(value)) return value;
288+ if (typeof File !== "undefined" && value instanceof File) {
289+ return {
290+ __executorBinary: 1 as const,
291+ kind: "file" as const,
292+ type: value.type,
293+ name: value.name,
294+ lastModified: value.lastModified,
295+ buffer: await value.arrayBuffer(),
296+ };
297+ }
298+ if (value instanceof Blob) {
299+ return {
300+ __executorBinary: 1 as const,
301+ kind: "blob" as const,
302+ type: value.type,
303+ buffer: await value.arrayBuffer(),
304+ };
305+ }
306+ if (Array.isArray(value)) return Promise.all(value.map(encodeBinary));
307+ if (!isPlainObject(value)) return value;
308+ const out: Record<string, unknown> = {};
309+ for (const [k, v] of Object.entries(value as Record<string, unknown>)) {
310+ out[k] = await encodeBinary(v);
311+ }
312+ return out;
313+};
222314
223315 // ---------------------------------------------------------------------------
224316 // ToolDispatcher — bridges RPC calls back to SandboxToolInvoker
@@ -227,7 +319,12 @@ export const decodeWorkerRpcResponse = (raw: string): WorkerRpcResponse => · 2 unmodified lines
227319 /**
228320 * An `RpcTarget` passed to the dynamic Worker so that sandboxed code can
229321 * invoke tools on the host. The dynamic worker calls
230− * `__dispatcher.call(path, argsJson)` over Workers RPC.
322+ * `__dispatcher.call(path, args)` over Workers RPC. `Uint8Array` /
323+ * `ArrayBuffer` cross structured clone natively; `Blob` / `File` are
324+ * encoded sandbox-side as a tagged envelope and rehydrated here via
325+ * `rehydrateBinary` before the invoker sees them. JSON serialization on
326+ * this hop would replace those values with `"{}"` or numeric-keyed
327+ * objects, which is what broke `multipart/form-data` uploads.
231328 *
232329 * Each call is wrapped in an `executor.tool.rpc_dispatch` span so the
233330 * tool-invocation shell (Workers RPC roundtrip → local invoker →
@@ -248,15 +345,17 @@ export class ToolDispatcher extends RpcTarget { · 14 unmodified lines
248345 this.#runPromise = runPromise;
249346 }
250347
251− async call(path: string, argsJson: string): Promise<string> {
252− const args = argsJson ? JSON.parse(argsJson) : undefined;
253−
348+ async call(path: string, args: unknown): Promise<WorkerRpcResponse> {
349+ const decodedArgs = rehydrateBinary(args);
254350 return this.#runPromise(
255− this.#invoker.invoke({ path, args }).pipe(
256− Effect.map(
257− (value): WorkerRpcResponse => ({
258− ok: true,
259− result: value,
351+ this.#invoker.invoke({ path, args: decodedArgs }).pipe(
352+ Effect.flatMap((value) =>
353+ Effect.tryPromise({
354+ try: (): Promise<WorkerRpcResponse> =>
355+ encodeBinary(value).then((result) => ({ ok: true, result })),
356+ // Encoding failed (e.g. Blob.arrayBuffer rejected) — surface
357+ // it as a normal failure envelope rather than throwing.
358+ catch: (cause) => cause,
260359 }),
261360 ),
262361 Effect.catchCause((cause) =>
@@ -265,11 +364,9 @@ export class ToolDispatcher extends RpcTarget { · 2 unmodified lines
265364 error: serializeWorkerCause(cause),
266365 }),
267366 ),
268− Effect.map(encodeWorkerRpcResponse),
269367 Effect.withSpan("executor.tool.rpc_dispatch", {
270368 attributes: {
271369 "mcp.tool.name": path,
272− "executor.tool.args_length": argsJson.length,
273370 },
274371 }),
275372 ),
packages/kernel/runtime-dynamic-worker/src/invocation.test.ts+113−11
@@ -6,7 +6,6 @@ import * as Effect from "effect/Effect"; · 5 unmodified lines
66 import type { SandboxToolInvoker } from "@executor-js/codemode-core";
77 import {
88 ToolDispatcher,
9− decodeWorkerRpcResponse,
109 makeDynamicWorkerExecutor,
1110 renderWorkerError,
1211 serializeWorkerCause,
@@ -35,15 +34,15 @@ describe("ToolDispatcher", () => { · 22 unmodified lines
3534 const invoker = makeInvoker(({ args }) => args);
3635 const dispatcher = new ToolDispatcher(invoker, Effect.runPromise);
3736
38− const result = await dispatcher.call("test.tool", '{"key":"value"}');
39− expect(decodeWorkerRpcResponse(result)).toEqual({ ok: true, result: { key: "value" } });
37+ const result = await dispatcher.call("test.tool", { key: "value" });
38+ expect(result).toEqual({ ok: true, result: { key: "value" } });
4039 });
4140
4241 it("serializes tagged failures into a structured error envelope", async () => {
4342 const dispatcher = new ToolDispatcher(failingInvoker("tool broke"), Effect.runPromise);
4443
45− const result = await dispatcher.call("broken.tool", "{}");
46− expect(decodeWorkerRpcResponse(result)).toMatchObject({
44+ const result = await dispatcher.call("broken.tool", {});
45+ expect(result).toMatchObject({
4746 ok: false,
4847 error: {
4948 kind: "fail",
@@ -68,8 +67,8 @@ describe("ToolDispatcher", () => { · 18 unmodified lines
6867 Effect.runPromise,
6968 );
7069
71− const result = await dispatcher.call("broken.tool", "{}");
72− expect(decodeWorkerRpcResponse(result)).toEqual({
70+ const result = await dispatcher.call("broken.tool", {});
71+ expect(result).toEqual({
7372 ok: false,
7473 error: {
7574 kind: "fail",
@@ -94,8 +93,8 @@ describe("ToolDispatcher", () => { · 18 unmodified lines
9493 const invoker = makeInvoker(({ args }) => args);
9594 const dispatcher = new ToolDispatcher(invoker, Effect.runPromise);
9695
97− const result = await dispatcher.call("test.tool", "");
98− expect(decodeWorkerRpcResponse(result)).toEqual({ ok: true, result: undefined });
96+ const result = await dispatcher.call("test.tool", undefined);
97+ expect(result).toEqual({ ok: true, result: undefined });
9998 });
10099
101100 it("passes the tool path correctly", async () => {
@@ -106,7 +105,7 @@ describe("ToolDispatcher", () => { · 4 unmodified lines
106105 });
107106 const dispatcher = new ToolDispatcher(invoker, Effect.runPromise);
108107
109− await dispatcher.call("my.deep.tool.path", "{}");
108+ await dispatcher.call("my.deep.tool.path", {});
110109 expect(capturedPath).toBe("my.deep.tool.path");
111110 });
112111 });
@@ -145,7 +144,10 @@ describe("makeDynamicWorkerExecutor", () => { · 32 unmodified lines
145144 const invoker = makeInvoker(() => null);
146145
147146 const result = await Effect.runPromise(
148− executor.execute(["Use this snippet.", "", "```ts", "async () => 42", "```"].join("\n"), invoker),
147+ executor.execute(
148+ ["Use this snippet.", "", "```ts", "async () => 42", "```"].join("\n"),
149+ invoker,
150+ ),
149151 );
150152
151153 expect(result.error).toBeUndefined();
@@ -319,4 +321,104 @@ describe("makeDynamicWorkerExecutor", () => { · 167 unmodified lines
319321
320322 expect(result.error).toBeDefined();
321323 });
324+
325+ // Multipart/form-data uploads (OpenAI Files API, any spec with a
326+ // `multipart/form-data` body) need Blob/File/Uint8Array values to
327+ // survive the sandbox→host RPC hop intact. JSON.stringify turns a Blob
328+ // into "{}" and a Uint8Array into a numeric-keyed object, so
329+ // `coerceFormDataRecord` produces a malformed multipart part and the
330+ // upstream server 400s.
331+ it.effect("preserves Uint8Array tool args across the dispatcher boundary", () =>
332+ Effect.gen(function* () {
333+ const executor = makeDynamicWorkerExecutor({ loader });
334+ let captured: { file?: unknown } = {};
335+ const invoker = makeInvoker(({ args }) => {
336+ captured = (args ?? {}) as { file?: unknown };
337+ return null;
338+ });
339+
340+ yield* executor.execute(
341+ `async () => {
342+ const bytes = new Uint8Array([0xde, 0xad, 0xbe, 0xef]);
343+ await tools.uploads.send({ file: bytes });
344+ }`,
345+ invoker,
346+ );
347+
348+ expect(captured.file).toBeInstanceOf(Uint8Array);
349+ expect(Array.from(captured.file as Uint8Array)).toEqual([0xde, 0xad, 0xbe, 0xef]);
350+ }),
351+ );
352+
353+ it.effect("preserves Blob tool args across the dispatcher boundary", () =>
354+ Effect.gen(function* () {
355+ const executor = makeDynamicWorkerExecutor({ loader });
356+ let captured: { file?: unknown } = {};
357+ const invoker = makeInvoker(({ args }) => {
358+ captured = (args ?? {}) as { file?: unknown };
359+ return null;
360+ });
361+
362+ const result = yield* executor.execute(
363+ `async () => {
364+ const file = new Blob(["hello multipart"], { type: "text/plain" });
365+ await tools.uploads.send({ file });
366+ }`,
367+ invoker,
368+ );
369+
370+ expect(result.error).toBeUndefined();
371+ expect(captured.file).toBeInstanceOf(Blob);
372+ const blob = captured.file as Blob;
373+ expect(blob.type).toBe("text/plain");
374+ const body = yield* Effect.promise(() => blob.text());
375+ expect(body).toBe("hello multipart");
376+ }),
377+ );
378+
379+ // Symmetric direction: tool RESULT contains Blob/Uint8Array/File. Workers
380+ // RPC has the same "Could not serialize Blob" limit on the way back, so
381+ // tool implementations that return file-like data need the host→sandbox
382+ // codec too. This pins down which types survive both directions.
383+ it.effect("returns Uint8Array tool results to the sandbox intact", () =>
384+ Effect.gen(function* () {
385+ const executor = makeDynamicWorkerExecutor({ loader });
386+ const invoker = makeInvoker(() => new Uint8Array([0xca, 0xfe, 0xba, 0xbe]));
387+
388+ const result = yield* executor.execute(
389+ `async () => {
390+ const bytes = await tools.download.fetch({});
391+ if (!(bytes instanceof Uint8Array)) return { kind: typeof bytes, ctor: bytes && bytes.constructor && bytes.constructor.name };
392+ return { kind: 'Uint8Array', length: bytes.length, bytes: Array.from(bytes) };
393+ }`,
394+ invoker,
395+ );
396+
397+ expect(result.error).toBeUndefined();
398+ expect(result.result).toEqual({
399+ kind: "Uint8Array",
400+ length: 4,
401+ bytes: [0xca, 0xfe, 0xba, 0xbe],
402+ });
403+ }),
404+ );
405+
406+ it.effect("returns Blob tool results to the sandbox intact", () =>
407+ Effect.gen(function* () {
408+ const executor = makeDynamicWorkerExecutor({ loader });
409+ const invoker = makeInvoker(() => new Blob(["DOWNLOAD"], { type: "text/plain" }));
410+
411+ const result = yield* executor.execute(
412+ `async () => {
413+ const blob = await tools.download.fetch({});
414+ if (!(blob instanceof Blob)) return { kind: typeof blob };
415+ return { kind: 'Blob', type: blob.type, text: await blob.text() };
416+ }`,
417+ invoker,
418+ );
419+
420+ expect(result.error).toBeUndefined();
421+ expect(result.result).toEqual({ kind: "Blob", type: "text/plain", text: "DOWNLOAD" });
422+ }),
423+ );
322424 });
packages/kernel/runtime-dynamic-worker/src/module-template.ts+64−4
@@ -7,6 +7,13 @@ · 6 unmodified lines
77 * 2. Creates a recursive `tools` Proxy that dispatches calls via RPC.
88 * 3. Executes the normalised user code with a `Promise.race` timeout.
99 * 4. Returns `{ result, error?, logs }`.
10+ *
11+ * Tool args cross the dispatcher boundary via Workers RPC, not JSON, so
12+ * `Uint8Array` / `ArrayBuffer` survive intact. `Blob` / `File` aren't on
13+ * RPC's structured-clone allow-list, so we encode them to a tagged
14+ * envelope (`__encodeBinary`) and the host rehydrates them. This is
15+ * required for any tool whose upstream wants `multipart/form-data`
16+ * (OpenAI Files API, etc.).
1017 */
1118 export const buildExecutorModule = (body: string, timeoutMs: number): string =>
1219 [
@@ -65,6 +72,58 @@ export const buildExecutorModule = (body: string, timeoutMs: number): string => · 52 unmodified lines
6572 " };",
6673 " };",
6774 "",
75+ // Blob/File aren't on Workers RPC's structured-clone allow-list, so
76+ // we encode them to a tagged ArrayBuffer envelope before the call and
77+ // the host rehydrates them — and symmetrically the host encodes
78+ // Blob/File results which we rehydrate here. ArrayBuffer / typed
79+ // arrays cross natively.
80+ " const __isPlainObject = (v) => {",
81+ " const p = Object.getPrototypeOf(v);",
82+ " return p === Object.prototype || p === null;",
83+ " };",
84+ " const __isBinaryEnvelope = (v) =>",
85+ " v && typeof v === 'object' && v.__executorBinary === 1 &&",
86+ " v.buffer instanceof ArrayBuffer && typeof v.type === 'string';",
87+ " const __encodeBinary = async (value) => {",
88+ " if (value === null || typeof value !== 'object') return value;",
89+ " if (value instanceof ArrayBuffer || ArrayBuffer.isView(value)) return value;",
90+ " if (typeof File !== 'undefined' && value instanceof File) {",
91+ " return {",
92+ " __executorBinary: 1,",
93+ " kind: 'file',",
94+ " type: value.type,",
95+ " name: value.name,",
96+ " lastModified: value.lastModified,",
97+ " buffer: await value.arrayBuffer(),",
98+ " };",
99+ " }",
100+ " if (value instanceof Blob) {",
101+ " return { __executorBinary: 1, kind: 'blob', type: value.type, buffer: await value.arrayBuffer() };",
102+ " }",
103+ " if (Array.isArray(value)) return Promise.all(value.map(__encodeBinary));",
104+ " if (!__isPlainObject(value)) return value;",
105+ " const out = {};",
106+ " for (const [k, v] of Object.entries(value)) out[k] = await __encodeBinary(v);",
107+ " return out;",
108+ " };",
109+ " const __decodeBinary = (value) => {",
110+ " if (value === null || typeof value !== 'object') return value;",
111+ " if (value instanceof ArrayBuffer || ArrayBuffer.isView(value)) return value;",
112+ " if (__isBinaryEnvelope(value)) {",
113+ " if (value.kind === 'file' && typeof value.name === 'string') {",
114+ " return new File([value.buffer], value.name, {",
115+ " type: value.type,",
116+ " ...(typeof value.lastModified === 'number' ? { lastModified: value.lastModified } : {}),",
117+ " });",
118+ " }",
119+ " return new Blob([value.buffer], { type: value.type });",
120+ " }",
121+ " if (Array.isArray(value)) return value.map(__decodeBinary);",
122+ " if (!__isPlainObject(value)) return value;",
123+ " const out = {};",
124+ " for (const [k, v] of Object.entries(value)) out[k] = __decodeBinary(v);",
125+ " return out;",
126+ " };",
68127 " const __makeToolsProxy = (path = []) => new Proxy(() => undefined, {",
69128 " get(_target, prop) {",
70129 " if (prop === 'then' || typeof prop === 'symbol') return undefined;",
@@ -73,13 +132,14 @@ export const buildExecutorModule = (body: string, timeoutMs: number): string => · 2 unmodified lines
73132 " apply(_target, _thisArg, args) {",
74133 " const toolPath = path.join('.');",
75134 " if (!toolPath) throw new Error('Tool path missing in invocation');",
76− " return __dispatcher.call(toolPath, JSON.stringify(args[0] ?? {})).then((raw) => {",
77− " const data = JSON.parse(raw);",
135+ " return (async () => {",
136+ " const encoded = await __encodeBinary(args[0]);",
137+ " const data = await __dispatcher.call(toolPath, encoded);",
78138 " if (!data.ok) {",
79139 " throw (data.error && data.error.primary !== null ? data.error.primary : data.error.message);",
80140 " }",
81− " return data.result;",
82− " });",
141+ " return __decodeBinary(data.result);",
142+ " })();",
83143 " },",
84144 " });",
85145 " const tools = __makeToolsProxy();",

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t