/** * DynamicWorkerExecutor — runs sandboxed code in an isolated Cloudflare * Worker via the WorkerLoader binding. * * Tool calls are dispatched over Workers RPC: the host creates a * `ToolDispatcher` (an `RpcTarget`) that bridges back to the * `SandboxToolInvoker` from codemode-core, and passes it to the * dynamic worker's `evaluate()` entrypoint. */ import { RpcTarget } from "cloudflare:workers";import * as Cause from "effect/Cause";import * as Data from "effect/Data";import * as Effect from "effect/Effect"; import { CodeCompilationError, recoverExecutionBody, SandboxHostTimeoutError, SandboxRuntimeError, stripTypeScript, type CodeExecutor, type ExecuteOutputItem, type ExecuteResult, type SandboxToolInvoker,} from "@executor-js/codemode-core"; import { buildExecutorModule } from "./module-template"; // ---------------------------------------------------------------------------// Errors// --------------------------------------------------------------------------- export class DynamicWorkerExecutionError extends Data.TaggedError("DynamicWorkerExecutionError")<{ readonly message: string;}> {} // ---------------------------------------------------------------------------// Options// --------------------------------------------------------------------------- export type DynamicWorkerExecutorOptions = { readonly loader: WorkerLoader; /** * Timeout in milliseconds for code execution. Defaults to 5 minutes. */ readonly timeoutMs?: number; /** * Extra wall-clock the host waits beyond `timeoutMs` before declaring the * isolate wedged and delivering a `SandboxHostTimeoutError`. Defaults to 30s. * The in-sandbox timer should always fire first for a healthy isolate; this * margin only covers the case where that timer is defeated by a wedged RPC. * Exposed primarily so tests can shrink the backstop; production leaves it * unset. */ readonly hostTimeoutGraceMs?: number; /** * Poll interval for the host-timeout watchdog, in milliseconds. Defaults to * 1s. Exposed for tests; production leaves it unset. */ readonly hostTimeoutPollMs?: number; /** * Controls outbound network access from sandboxed code. * - `null` (default): `fetch()` and `connect()` throw — fully isolated. * - `undefined`: inherits parent Worker's network access. * - A `Fetcher`: all outbound requests route through this handler. */ readonly globalOutbound?: Fetcher | null; /** * Additional modules to make available in the sandbox. * Keys are module specifiers, values are module source code. * The key `"executor.js"` is reserved. */ readonly modules?: Record<string, string>;}; export type SerializedWorkerErrorValue = unknown; export type SerializedWorkerError = { readonly kind: "fail" | "die" | "interrupt" | "mixed" | "unknown"; readonly message: string; readonly primary: SerializedWorkerErrorValue | null; readonly failures: ReadonlyArray<SerializedWorkerErrorValue>; readonly defects: ReadonlyArray<SerializedWorkerErrorValue>; readonly interrupted: boolean;}; type WorkerRpcSuccess = { readonly ok: true; readonly result: unknown;}; type WorkerRpcFailure = { readonly ok: false; readonly error: SerializedWorkerError;}; type WorkerRpcResponse = WorkerRpcSuccess | WorkerRpcFailure; // ---------------------------------------------------------------------------// Constants// --------------------------------------------------------------------------- const DEFAULT_TIMEOUT_MS = 5 * 60_000;/** * Extra wall-clock the host waits beyond the effective in-sandbox timeout * before declaring the isolate wedged. The in-sandbox timer should always fire * first for a healthy isolate; this margin covers scheduling jitter and the RPC * unwind so a normally-timing-out execution reports its own descriptive timeout * rather than this backstop. */const HOST_TIMEOUT_GRACE_MS = 30_000;const ENTRY_MODULE = "executor.js"; const normalizeErrorObject = (error: Error) => ({ __type: "Error" as const, name: error.name, message: error.message,}); const isNormalizedErrorObject = ( value: unknown,): value is { readonly __type: "Error"; readonly message: string } => typeof value === "object" && value !== null && "__type" in value && value.__type === "Error" && "message" in value && typeof value.message === "string"; const serializeWorkerErrorValue = (value: unknown): SerializedWorkerErrorValue => { if (value instanceof Error) { return normalizeErrorObject(value); } if ( value === null || typeof value === "string" || typeof value === "number" || typeof value === "boolean" ) { return value; } try { return JSON.parse(JSON.stringify(value)) as SerializedWorkerErrorValue; } catch { return String(value); }}; const renderTransportMessage = (value: unknown): string => { if (typeof value === "string") { return value; } if (isNormalizedErrorObject(value)) { return value.message; } if (value instanceof Error) { return value.message; } if ( typeof value === "object" && value !== null && "message" in value && typeof value.message === "string" ) { return value.message; } if (typeof value === "object" && value !== null) { try { return JSON.stringify(value); } catch { return String(value); } } if (typeof value === "undefined") { return "Unknown error"; } return String(value);}; const serializedErrorName = (value: SerializedWorkerErrorValue): string | null => typeof value === "object" && value !== null && "name" in value && typeof (value as { name?: unknown }).name === "string" ? (value as { name: string }).name : null; /** * Signatures of a compile error in the user's own code. Not all syntax * errors are caught while stripping TypeScript: smart quotes from a * paste, an unbalanced brace, and other plain-JS parse errors slip past * sucrase and only fail when workerd compiles the generated module, * surfacing as "Failed to start Worker: Uncaught SyntaxError: ...". The * bare V8 phrasings ("Unexpected token ...", "Invalid or unexpected * token") are matched too, since some surfaces report the inner message * without the wrapper. */const COMPILE_SIGNATURES = [ "Failed to start Worker", "SyntaxError", "Unexpected token", "Invalid or unexpected token",] as const; /** * Signatures of a sandbox runtime condition that is the user's own * concern (or transient and retryable) rather than an executor defect: a * non-serializable return value, the isolate's CPU or memory limit, and * being momentarily at worker capacity. These are the real categories * seen in production on the `executor.runtime.*` spans, all of which were * being collapsed to an opaque internal error before the model could act * on them. */const RUNTIME_SIGNATURES = [ "could not be cloned", "does not support serialization", "Could not serialize", "exceeded CPU", "exceeded memory", "Too many concurrent dynamic workers",] as const; export type SandboxFailureKind = "compilation" | "runtime" | "internal"; /** * Classify a sandbox rejection so the runtime knows whether to surface * its message descriptively (the user's mistake or a transient, * safe-to-report condition) or collapse it to an opaque internal error (a * genuine, unexpected sandbox defect). Tool-invocation failures never * reach here: the sandbox reports those through its own result envelope, * so this only sees module compile failures, return-value serialization * failures, isolate resource limits, and capacity rejections. Anything * unrecognized stays "internal" and opaque, preserving the host's * failure-channel boundary. */export const classifySandboxFailure = ( serialized: SerializedWorkerErrorValue, message: string,): SandboxFailureKind => { const name = serializedErrorName(serialized); if ( name === "SyntaxError" || COMPILE_SIGNATURES.some((signature) => message.includes(signature)) ) { return "compilation"; } if ( name === "DataCloneError" || RUNTIME_SIGNATURES.some((signature) => message.includes(signature)) ) { return "runtime"; } return "internal";}; /** * Map a raw sandbox rejection (a thrown value from the worker loader or * the `evaluate` RPC) to the typed error its classification calls for. * Compilation and runtime conditions carry the verbatim message through * the descriptive channel; unrecognized defects stay opaque. */const toSandboxFailure = ( cause: unknown,): CodeCompilationError | SandboxRuntimeError | DynamicWorkerExecutionError => { const serialized = serializeWorkerErrorValue(cause); const message = renderTransportMessage(serialized); switch (classifySandboxFailure(serialized, message)) { case "compilation": return new CodeCompilationError({ runtime: "dynamic-worker", message, cause }); case "runtime": return new SandboxRuntimeError({ runtime: "dynamic-worker", message, cause }); default: return new DynamicWorkerExecutionError({ message }); }}; export const serializeWorkerCause = (cause: Cause.Cause<unknown>): SerializedWorkerError => { const failures = cause.reasons .filter(Cause.isFailReason) .map((reason) => serializeWorkerErrorValue(reason.error)); const defects = cause.reasons .filter(Cause.isDieReason) .map((reason) => serializeWorkerErrorValue(reason.defect)); const interrupted = cause.reasons.some(Cause.isInterruptReason); const primary = failures[0] ?? defects[0] ?? null; const kind = failures.length > 0 && defects.length > 0 ? "mixed" : failures.length > 0 ? "fail" : defects.length > 0 ? "die" : interrupted ? "interrupt" : "unknown"; return { kind, message: primary !== null ? renderTransportMessage(primary) : interrupted ? "Interrupted" : "Unknown error", primary, failures, defects, interrupted, };}; export const renderWorkerError = (error: SerializedWorkerError): string => { if (isNormalizedErrorObject(error.primary)) { return error.primary.message; } if (typeof error.primary === "string") { return error.primary; } if ( typeof error.primary === "object" && error.primary !== null && "message" in error.primary && typeof error.primary.message === "string" ) { return error.primary.message; } if (typeof error.primary === "object" && error.primary !== null) { try { return JSON.stringify(error.primary); } catch { return error.message; } } return error.message;}; export type { WorkerRpcResponse }; // ---------------------------------------------------------------------------// Blob/File codec (both directions across the dispatcher boundary)//// Workers RPC's structured-clone allow-list excludes `Blob` / `File`, so// we encode them to a tagged ArrayBuffer envelope and rehydrate on the// far side. Symmetric in both directions: sandbox encodes args + host// rehydrates them; host encodes result + sandbox rehydrates it. The// matching encoder lives inside `module-template.ts` because it runs in// the dynamic Worker isolate. `ArrayBuffer` / typed arrays / primitives// cross structured clone natively.// --------------------------------------------------------------------------- type BinaryEnvelope = { readonly __executorBinary: 1; readonly kind: "blob" | "file"; readonly type: string; readonly name?: string; readonly lastModified?: number; readonly buffer: ArrayBuffer;}; const isBinaryEnvelope = (value: unknown): value is BinaryEnvelope => typeof value === "object" && value !== null && (value as { __executorBinary?: unknown }).__executorBinary === 1 && (value as { buffer?: unknown }).buffer instanceof ArrayBuffer && typeof (value as { type?: unknown }).type === "string"; const isPlainObject = (value: object): boolean => { const proto = Object.getPrototypeOf(value); return proto === Object.prototype || proto === null;}; const rehydrateBinary = (value: unknown, seen = new WeakSet<object>()): unknown => { if (value === null || typeof value !== "object") return value; if (value instanceof ArrayBuffer || ArrayBuffer.isView(value)) return value; if (seen.has(value)) { throw new Error("Tool RPC payload contains a circular reference"); } seen.add(value); if (isBinaryEnvelope(value)) { seen.delete(value); if (value.kind === "file" && typeof value.name === "string") { return new File([value.buffer], value.name, { type: value.type, ...(typeof value.lastModified === "number" ? { lastModified: value.lastModified } : {}), }); } return new Blob([value.buffer], { type: value.type }); } if (Array.isArray(value)) { const out = value.map((item) => rehydrateBinary(item, seen)); seen.delete(value); return out; } if (!isPlainObject(value)) { seen.delete(value); return value; } const out: Record<string, unknown> = {}; for (const [k, v] of Object.entries(value as Record<string, unknown>)) { out[k] = rehydrateBinary(v, seen); } seen.delete(value); return out;}; // Async because `Blob.arrayBuffer()` is async. Used on tool results before// the dispatcher hands them back to the sandbox.const encodeBinary = async (value: unknown, seen = new WeakSet<object>()): Promise<unknown> => { if (value === null || typeof value !== "object") return value; if (value instanceof ArrayBuffer || ArrayBuffer.isView(value)) return value; if (seen.has(value)) { throw new Error("Tool RPC payload contains a circular reference"); } seen.add(value); if (typeof File !== "undefined" && value instanceof File) { const out = { __executorBinary: 1 as const, kind: "file" as const, type: value.type, name: value.name, lastModified: value.lastModified, buffer: await value.arrayBuffer(), }; seen.delete(value); return out; } if (value instanceof Blob) { const out = { __executorBinary: 1 as const, kind: "blob" as const, type: value.type, buffer: await value.arrayBuffer(), }; seen.delete(value); return out; } if (Array.isArray(value)) { const out = await Promise.all(value.map((item) => encodeBinary(item, seen))); seen.delete(value); return out; } if (!isPlainObject(value)) { seen.delete(value); return value; } const out: Record<string, unknown> = {}; for (const [k, v] of Object.entries(value as Record<string, unknown>)) { out[k] = await encodeBinary(v, seen); } seen.delete(value); return out;}; // ---------------------------------------------------------------------------// ToolDispatcher — bridges RPC calls back to SandboxToolInvoker// --------------------------------------------------------------------------- /** * An `RpcTarget` passed to the dynamic Worker so that sandboxed code can * invoke tools on the host. The dynamic worker calls * `__dispatcher.call(path, args)` over Workers RPC. `Uint8Array` / * `ArrayBuffer` cross structured clone natively; `Blob` / `File` are * encoded sandbox-side as a tagged envelope and rehydrated here via * `rehydrateBinary` before the invoker sees them. JSON serialization on * this hop would replace those values with `"{}"` or numeric-keyed * objects, which is what broke `multipart/form-data` uploads. * * Each call is wrapped in an `executor.tool.rpc_dispatch` span so the * tool-invocation shell (Workers RPC roundtrip → local invoker → * serialize result) is visible in the trace. Tool-level attributes * like `mcp.tool.name` already come from the inner * `mcp.tool.dispatch` span that `tool-invoker.ts` wraps around * `executor.tools.invoke`. */export type RunPromise = <A, E>(effect: Effect.Effect<A, E>) => Promise<A>; export class ToolDispatcher extends RpcTarget { readonly #invoker: SandboxToolInvoker; readonly #runPromise: RunPromise; // Number of tool round-trips currently in flight. The host-side execution // timeout is suspended while this is > 0: whenever the sandbox is blocked // here it is demonstrably not wedged (the host owns the continuation and // will resume it), so neither a slow tool call nor a minutes-long approval // pause (which manifests as an in-flight dispatch awaiting an elicitation // response) should count against the unresponsive-sandbox backstop. #inFlight = 0; #lastReturnedAt = Date.now();Lines 1–500 of 843. Load more
/** * DynamicWorkerExecutor — runs sandboxed code in an isolated Cloudflare * Worker via the WorkerLoader binding. * * Tool calls are dispatched over Workers RPC: the host creates a * `ToolDispatcher` (an `RpcTarget`) that bridges back to the * `SandboxToolInvoker` from codemode-core, and passes it to the * dynamic worker's `evaluate()` entrypoint. */ import { RpcTarget } from "cloudflare:workers"; import * as Cause from "effect/Cause"; import * as Data from "effect/Data"; import * as Effect from "effect/Effect"; import { CodeCompilationError, recoverExecutionBody, SandboxHostTimeoutError, SandboxRuntimeError, stripTypeScript, type CodeExecutor, type ExecuteOutputItem, type ExecuteResult, type SandboxToolInvoker, } from "@executor-js/codemode-core"; import { buildExecutorModule } from "./module-template"; // --------------------------------------------------------------------------- // Errors // --------------------------------------------------------------------------- export class DynamicWorkerExecutionError extends Data.TaggedError("DynamicWorkerExecutionError")<{ readonly message: string; }> {} // --------------------------------------------------------------------------- // Options // --------------------------------------------------------------------------- export type DynamicWorkerExecutorOptions = { readonly loader: WorkerLoader; /** * Timeout in milliseconds for code execution. Defaults to 5 minutes. */ readonly timeoutMs?: number; /** * Extra wall-clock the host waits beyond `timeoutMs` before declaring the * isolate wedged and delivering a `SandboxHostTimeoutError`. Defaults to 30s. * The in-sandbox timer should always fire first for a healthy isolate; this * margin only covers the case where that timer is defeated by a wedged RPC. * Exposed primarily so tests can shrink the backstop; production leaves it * unset. */ readonly hostTimeoutGraceMs?: number; /** * Poll interval for the host-timeout watchdog, in milliseconds. Defaults to * 1s. Exposed for tests; production leaves it unset. */ readonly hostTimeoutPollMs?: number; /** * Controls outbound network access from sandboxed code. * - `null` (default): `fetch()` and `connect()` throw — fully isolated. * - `undefined`: inherits parent Worker's network access. * - A `Fetcher`: all outbound requests route through this handler. */ readonly globalOutbound?: Fetcher | null; /** * Additional modules to make available in the sandbox. * Keys are module specifiers, values are module source code. * The key `"executor.js"` is reserved. */ readonly modules?: Record<string, string>; }; export type SerializedWorkerErrorValue = unknown; export type SerializedWorkerError = { readonly kind: "fail" | "die" | "interrupt" | "mixed" | "unknown"; readonly message: string; readonly primary: SerializedWorkerErrorValue | null; readonly failures: ReadonlyArray<SerializedWorkerErrorValue>; readonly defects: ReadonlyArray<SerializedWorkerErrorValue>; readonly interrupted: boolean; }; type WorkerRpcSuccess = { readonly ok: true; readonly result: unknown; }; type WorkerRpcFailure = { readonly ok: false; readonly error: SerializedWorkerError; }; type WorkerRpcResponse = WorkerRpcSuccess | WorkerRpcFailure; // --------------------------------------------------------------------------- // Constants // --------------------------------------------------------------------------- const DEFAULT_TIMEOUT_MS = 5 * 60_000; /** * Extra wall-clock the host waits beyond the effective in-sandbox timeout * before declaring the isolate wedged. The in-sandbox timer should always fire * first for a healthy isolate; this margin covers scheduling jitter and the RPC * unwind so a normally-timing-out execution reports its own descriptive timeout * rather than this backstop. */ const HOST_TIMEOUT_GRACE_MS = 30_000; const ENTRY_MODULE = "executor.js"; const normalizeErrorObject = (error: Error) => ({ __type: "Error" as const, name: error.name, message: error.message, }); const isNormalizedErrorObject = ( value: unknown, ): value is { readonly __type: "Error"; readonly message: string } => typeof value === "object" && value !== null && "__type" in value && value.__type === "Error" && "message" in value && typeof value.message === "string"; const serializeWorkerErrorValue = (value: unknown): SerializedWorkerErrorValue => { if (value instanceof Error) { return normalizeErrorObject(value); } if ( value === null || typeof value === "string" || typeof value === "number" || typeof value === "boolean" ) { return value; } try { return JSON.parse(JSON.stringify(value)) as SerializedWorkerErrorValue; } catch { return String(value); } }; const renderTransportMessage = (value: unknown): string => { if (typeof value === "string") { return value; } if (isNormalizedErrorObject(value)) { return value.message; } if (value instanceof Error) { return value.message; } if ( typeof value === "object" && value !== null && "message" in value && typeof value.message === "string" ) { return value.message; } if (typeof value === "object" && value !== null) { try { return JSON.stringify(value); } catch { return String(value); } } if (typeof value === "undefined") { return "Unknown error"; } return String(value); }; const serializedErrorName = (value: SerializedWorkerErrorValue): string | null => typeof value === "object" && value !== null && "name" in value && typeof (value as { name?: unknown }).name === "string" ? (value as { name: string }).name : null; /** * Signatures of a compile error in the user's own code. Not all syntax * errors are caught while stripping TypeScript: smart quotes from a * paste, an unbalanced brace, and other plain-JS parse errors slip past * sucrase and only fail when workerd compiles the generated module, * surfacing as "Failed to start Worker: Uncaught SyntaxError: ...". The * bare V8 phrasings ("Unexpected token ...", "Invalid or unexpected * token") are matched too, since some surfaces report the inner message * without the wrapper. */ const COMPILE_SIGNATURES = [ "Failed to start Worker", "SyntaxError", "Unexpected token", "Invalid or unexpected token", ] as const; /** * Signatures of a sandbox runtime condition that is the user's own * concern (or transient and retryable) rather than an executor defect: a * non-serializable return value, the isolate's CPU or memory limit, and * being momentarily at worker capacity. These are the real categories * seen in production on the `executor.runtime.*` spans, all of which were * being collapsed to an opaque internal error before the model could act * on them. */ const RUNTIME_SIGNATURES = [ "could not be cloned", "does not support serialization", "Could not serialize", "exceeded CPU", "exceeded memory", "Too many concurrent dynamic workers", ] as const; export type SandboxFailureKind = "compilation" | "runtime" | "internal"; /** * Classify a sandbox rejection so the runtime knows whether to surface * its message descriptively (the user's mistake or a transient, * safe-to-report condition) or collapse it to an opaque internal error (a * genuine, unexpected sandbox defect). Tool-invocation failures never * reach here: the sandbox reports those through its own result envelope, * so this only sees module compile failures, return-value serialization * failures, isolate resource limits, and capacity rejections. Anything * unrecognized stays "internal" and opaque, preserving the host's * failure-channel boundary. */ export const classifySandboxFailure = ( serialized: SerializedWorkerErrorValue, message: string, ): SandboxFailureKind => { const name = serializedErrorName(serialized); if ( name === "SyntaxError" || COMPILE_SIGNATURES.some((signature) => message.includes(signature)) ) { return "compilation"; } if ( name === "DataCloneError" || RUNTIME_SIGNATURES.some((signature) => message.includes(signature)) ) { return "runtime"; } return "internal"; }; /** * Map a raw sandbox rejection (a thrown value from the worker loader or * the `evaluate` RPC) to the typed error its classification calls for. * Compilation and runtime conditions carry the verbatim message through * the descriptive channel; unrecognized defects stay opaque. */ const toSandboxFailure = ( cause: unknown, ): CodeCompilationError | SandboxRuntimeError | DynamicWorkerExecutionError => { const serialized = serializeWorkerErrorValue(cause); const message = renderTransportMessage(serialized); switch (classifySandboxFailure(serialized, message)) { case "compilation": return new CodeCompilationError({ runtime: "dynamic-worker", message, cause }); case "runtime": return new SandboxRuntimeError({ runtime: "dynamic-worker", message, cause }); default: return new DynamicWorkerExecutionError({ message }); } }; export const serializeWorkerCause = (cause: Cause.Cause<unknown>): SerializedWorkerError => { const failures = cause.reasons .filter(Cause.isFailReason) .map((reason) => serializeWorkerErrorValue(reason.error)); const defects = cause.reasons .filter(Cause.isDieReason) .map((reason) => serializeWorkerErrorValue(reason.defect)); const interrupted = cause.reasons.some(Cause.isInterruptReason); const primary = failures[0] ?? defects[0] ?? null; const kind = failures.length > 0 && defects.length > 0 ? "mixed" : failures.length > 0 ? "fail" : defects.length > 0 ? "die" : interrupted ? "interrupt" : "unknown"; return { kind, message: primary !== null ? renderTransportMessage(primary) : interrupted ? "Interrupted" : "Unknown error", primary, failures, defects, interrupted, }; }; export const renderWorkerError = (error: SerializedWorkerError): string => { if (isNormalizedErrorObject(error.primary)) { return error.primary.message; } if (typeof error.primary === "string") { return error.primary; } if ( typeof error.primary === "object" && error.primary !== null && "message" in error.primary && typeof error.primary.message === "string" ) { return error.primary.message; } if (typeof error.primary === "object" && error.primary !== null) { try { return JSON.stringify(error.primary); } catch { return error.message; } } return error.message; }; export type { WorkerRpcResponse }; // --------------------------------------------------------------------------- // Blob/File codec (both directions across the dispatcher boundary) // // Workers RPC's structured-clone allow-list excludes `Blob` / `File`, so // we encode them to a tagged ArrayBuffer envelope and rehydrate on the // far side. Symmetric in both directions: sandbox encodes args + host // rehydrates them; host encodes result + sandbox rehydrates it. The // matching encoder lives inside `module-template.ts` because it runs in // the dynamic Worker isolate. `ArrayBuffer` / typed arrays / primitives // cross structured clone natively. // --------------------------------------------------------------------------- type BinaryEnvelope = { readonly __executorBinary: 1; readonly kind: "blob" | "file"; readonly type: string; readonly name?: string; readonly lastModified?: number; readonly buffer: ArrayBuffer; }; const isBinaryEnvelope = (value: unknown): value is BinaryEnvelope => typeof value === "object" && value !== null && (value as { __executorBinary?: unknown }).__executorBinary === 1 && (value as { buffer?: unknown }).buffer instanceof ArrayBuffer && typeof (value as { type?: unknown }).type === "string"; const isPlainObject = (value: object): boolean => { const proto = Object.getPrototypeOf(value); return proto === Object.prototype || proto === null; }; const rehydrateBinary = (value: unknown, seen = new WeakSet<object>()): unknown => { if (value === null || typeof value !== "object") return value; if (value instanceof ArrayBuffer || ArrayBuffer.isView(value)) return value; if (seen.has(value)) { throw new Error("Tool RPC payload contains a circular reference"); } seen.add(value); if (isBinaryEnvelope(value)) { seen.delete(value); if (value.kind === "file" && typeof value.name === "string") { return new File([value.buffer], value.name, { type: value.type, ...(typeof value.lastModified === "number" ? { lastModified: value.lastModified } : {}), }); } return new Blob([value.buffer], { type: value.type }); } if (Array.isArray(value)) { const out = value.map((item) => rehydrateBinary(item, seen)); seen.delete(value); return out; } if (!isPlainObject(value)) { seen.delete(value); return value; } const out: Record<string, unknown> = {}; for (const [k, v] of Object.entries(value as Record<string, unknown>)) { out[k] = rehydrateBinary(v, seen); } seen.delete(value); return out; }; // Async because `Blob.arrayBuffer()` is async. Used on tool results before // the dispatcher hands them back to the sandbox. const encodeBinary = async (value: unknown, seen = new WeakSet<object>()): Promise<unknown> => { if (value === null || typeof value !== "object") return value; if (value instanceof ArrayBuffer || ArrayBuffer.isView(value)) return value; if (seen.has(value)) { throw new Error("Tool RPC payload contains a circular reference"); } seen.add(value); if (typeof File !== "undefined" && value instanceof File) { const out = { __executorBinary: 1 as const, kind: "file" as const, type: value.type, name: value.name, lastModified: value.lastModified, buffer: await value.arrayBuffer(), }; seen.delete(value); return out; } if (value instanceof Blob) { const out = { __executorBinary: 1 as const, kind: "blob" as const, type: value.type, buffer: await value.arrayBuffer(), }; seen.delete(value); return out; } if (Array.isArray(value)) { const out = await Promise.all(value.map((item) => encodeBinary(item, seen))); seen.delete(value); return out; } if (!isPlainObject(value)) { seen.delete(value); return value; } const out: Record<string, unknown> = {}; for (const [k, v] of Object.entries(value as Record<string, unknown>)) { out[k] = await encodeBinary(v, seen); } seen.delete(value); return out; }; // --------------------------------------------------------------------------- // ToolDispatcher — bridges RPC calls back to SandboxToolInvoker // --------------------------------------------------------------------------- /** * An `RpcTarget` passed to the dynamic Worker so that sandboxed code can * invoke tools on the host. The dynamic worker calls * `__dispatcher.call(path, args)` over Workers RPC. `Uint8Array` / * `ArrayBuffer` cross structured clone natively; `Blob` / `File` are * encoded sandbox-side as a tagged envelope and rehydrated here via * `rehydrateBinary` before the invoker sees them. JSON serialization on * this hop would replace those values with `"{}"` or numeric-keyed * objects, which is what broke `multipart/form-data` uploads. * * Each call is wrapped in an `executor.tool.rpc_dispatch` span so the * tool-invocation shell (Workers RPC roundtrip → local invoker → * serialize result) is visible in the trace. Tool-level attributes * like `mcp.tool.name` already come from the inner * `mcp.tool.dispatch` span that `tool-invoker.ts` wraps around * `executor.tools.invoke`. */ export type RunPromise = <A, E>(effect: Effect.Effect<A, E>) => Promise<A>; export class ToolDispatcher extends RpcTarget { readonly #invoker: SandboxToolInvoker; readonly #runPromise: RunPromise; // Number of tool round-trips currently in flight. The host-side execution // timeout is suspended while this is > 0: whenever the sandbox is blocked // here it is demonstrably not wedged (the host owns the continuation and // will resume it), so neither a slow tool call nor a minutes-long approval // pause (which manifests as an in-flight dispatch awaiting an elicitation // response) should count against the unresponsive-sandbox backstop. #inFlight = 0; #lastReturnedAt = Date.now();