From ef2de530ffd597a1be3efc42ef4b21695effa74b Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Wed, 22 Jul 2026 19:09:27 +0800 Subject: [PATCH] fix(code-runtime): flatten worker JSON transport --- ...-20-code-mode-typed-tool-returns.i18n.yaml | 4 +- ...2026-07-20-code-mode-typed-tool-returns.md | 8 +- ...6-07-20-code-mode-typed-tool-returns.zh.md | 8 +- docs/config-catalog.md | 2 +- .../code-runtime-worker/README.md | 4 +- .../code-runtime-worker/src/bootstrap.ts | 19 +- .../code-runtime-worker/src/index.ts | 17 +- .../code-runtime-worker/src/protocol.ts | 16 +- .../code-runtime-worker/src/worker-json.ts | 181 ++++++++++++++++++ .../tests/bootstrap.spec.ts | 44 +++-- .../code-runtime-worker/tests/runtime.spec.ts | 13 +- .../tests/source-worker.compat.spec.ts | 5 +- .../tests/worker-json.spec.ts | 89 ++++++++- 13 files changed, 352 insertions(+), 58 deletions(-) diff --git a/.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.i18n.yaml b/.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.i18n.yaml index a84e69c18e..9ce72eca05 100644 --- a/.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.i18n.yaml +++ b/.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.i18n.yaml @@ -2,5 +2,5 @@ # side as of the last confirmed-consistent state. Both languages carry equal authority; # after editing either side, bring the other along and re-record with: # pnpm run verify-translation-pairing --write -2026-07-20-code-mode-typed-tool-returns.md: 2f7b39ddaa3d4f2441a9061583bb60dbb4c8e14c -2026-07-20-code-mode-typed-tool-returns.zh.md: 9842897cb372d04b4b09679ed13062511347cdc2 +2026-07-20-code-mode-typed-tool-returns.md: 9c24b2cdf493ffb904de16e6816438379afd4211 +2026-07-20-code-mode-typed-tool-returns.zh.md: 33d89b1014589535379ecff0821b7574c4dfde9e diff --git a/.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.md b/.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.md index 2f7b39ddaa..9c24b2cdf4 100644 --- a/.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.md +++ b/.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.md @@ -53,7 +53,7 @@ Before dispatch the bridge snapshots binding arguments as lossless JSON and make The worker exposes the actual `ToolCallError` constructor used for `tools` binding failures, so `error instanceof ToolCallError` works. The error has the standard `Error` message plus the exact `toolName`; it deliberately omits `ToolFailure.info`, error codes, and Native content. This is an exception contract for control flow, not a failure union for programmatic classification. -Binding arguments and resolutions are revalidated as lossless JSON on both sides of the hostile worker protocol and cross through structured clone with no byte cap. Both snapshot boundaries traverse iteratively, so valid nesting has no JavaScript call-stack depth cap. The dependency-light runtime seam names its structural equivalent `CodeJsonValue` so it need not depend on the session-owned canonical type; the generated SDK and tool API use `JsonValue`. Intermediate values are not prompt-truncated, context-spilled, or persisted. This preserves full acquired search, workflow, task, filesystem, and MCP values for programmatic filtering while leaving provider and executor acquisition limits truthful. +Binding arguments and resolutions are revalidated as lossless JSON on both sides of the hostile worker protocol and have no byte cap. Before crossing through structured clone, each detached value is encoded as a flat pre-order token stream whose transport nesting is bounded; the receiver rebuilds it iteratively. Valid application nesting therefore has neither a JavaScript call-stack depth cap nor a platform-specific nested structured-clone limit. The dependency-light runtime seam names its structural equivalent `CodeJsonValue` so it need not depend on the session-owned canonical type; the generated SDK and tool API use `JsonValue`. Intermediate values are not prompt-truncated, context-spilled, or persisted. This preserves full acquired search, workflow, task, filesystem, and MCP values for programmatic filtering while leaving provider and executor acquisition limits truthful. ### Outer result and output ledger @@ -63,7 +63,7 @@ The runtime accepts an exact lossless JSON completion of any root. Returning `un Logs stream eagerly so a terminated run can retain output already admitted. Native stdout and stderr writes that bypass the worker's patched stream slots use independent pipes, so terminal settlement continues bounded capture until worker termination completes before materializing the result. When the cap is crossed, the runtime returns an explicit bounded failure with the fitting captured prefix. That outer result then traverses the ordinary `run_code` rendering and spill policy, which may save the captured text and expose its configured head/tail preview. The spill layer cannot recover bytes the runtime rejected beyond the hard cap. -Compute time, wall time, worker heap, cancellation, and fresh-worker isolation remain independent limits. The outer ledger never charges intermediate bindings, so structured-clone cost and available process or worker memory are their practical bounds. +Compute time, wall time, worker heap, cancellation, and fresh-worker isolation remain independent limits. The outer ledger never charges intermediate bindings, so snapshotting, flat-wire encoding and decoding, structured-clone cost, and available process or worker memory are their practical bounds. ### Typed handles and lifetime @@ -97,14 +97,14 @@ Keyless real-worker integration tests pin the two handle workflows that prose re Code programs can compose tools through stable values instead of reverse-engineering Native prose. Native and Both Mode retain their existing text and editor presentation, while Code Mode receives output-schema types and exact runtime JSON. Tool authors must treat the canonical value as their programmatic API and put display-only formatting in the renderer. -The worker performs structured cloning and lossless validation but does not make intermediate values cheap or durable. Outer overflow is an explicit failed run, and error handling remains intentionally human-guided rather than a versioned code union. +The worker performs bounded-depth flat-wire transport and lossless validation but does not make intermediate values cheap or durable. Outer overflow is an explicit failed run, and error handling remains intentionally human-guided rather than a versioned code union. ## Known Limitations and Deferred Work - Subagent and workflow caller-defined structured outputs remain object-rooted through consumer-level guards even though tool outputs may use any JSON root. - Post-execute has separate value and presentation projections; replacing content is not a confidentiality mechanism, so policy must block or replace the value to hide it from programmatic callers. - Intermediate canonical values are execution-local and unavailable to replay because durable events persist only presentation and bounded summaries. -- Intermediate values have no byte cap and can exhaust process or worker memory through retention or structured-clone cost. +- Intermediate values have no byte cap and can exhaust process or worker memory through retention, flat-wire copies, or structured-clone cost. - The 64 MiB hard cap applies only to outer output; spill cannot recover bytes rejected beyond that cap. - Provider or executor acquisition limits may already have discarded source data before a canonical value reaches Code Mode. - Unsupported MCP output schemas fall back to `JsonValue`; richer Native multimedia projection is deferred. diff --git a/.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.zh.md b/.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.zh.md index 9842897cb3..33d89b1014 100644 --- a/.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.zh.md +++ b/.agents/notes/implemented/feature/2026-07-20-code-mode-typed-tool-returns.zh.md @@ -53,7 +53,7 @@ declare const tools: { worker 暴露的是真正用于 `tools` 绑定失败的 `ToolCallError` 构造函数，因此 `error instanceof ToolCallError` 能够成立。该错误包含标准的 `Error` 消息和确切的 `toolName`，并有意省略 `ToolFailure.info`、错误代码与 Native 内容。这是一项用于控制流的异常契约，而不是供程序分类的失败联合。 -绑定参数与绑定返回值会在不可信 worker 协议的两端重新校验为无损 JSON，再通过结构化克隆传输，且不设字节上限。两处快照边界均采用迭代方式遍历，因此有效嵌套不受 JavaScript 调用栈深度上限限制。为保持依赖轻量，运行时 seam 将结构等价类型命名为 `CodeJsonValue`，从而无需依赖会话侧拥有的规范类型；生成的 SDK 和工具 API 则使用 `JsonValue`。这些值不会经过提示词截断、上下文输出落盘或持久化。因此，程序可以完整筛选已经采集的搜索、工作流、任务、文件系统与 MCP 值，同时提供方和执行器的采集上限仍会实际生效。 +绑定参数与绑定返回值会在不可信 worker 协议的两端重新校验为无损 JSON，且不设字节上限。每个分离后的值在通过结构化克隆跨越边界前，都会编码为扁平的前序 token 流，其传输结构的嵌套深度有界；接收方再以迭代方式重建该值。因此，有效应用数据的嵌套深度既不受 JavaScript 调用栈深度上限限制，也不受特定平台对嵌套结构化克隆施加的上限限制。为保持依赖轻量，运行时 seam 将结构等价类型命名为 `CodeJsonValue`，从而无需依赖会话侧拥有的规范类型；生成的 SDK 和工具 API 则使用 `JsonValue`。这些值不会经过提示词截断、上下文输出落盘或持久化。因此，程序可以完整筛选已经采集的搜索、工作流、任务、文件系统与 MCP 值，同时提供方和执行器的采集上限仍会实际生效。 ### 外层结果与输出账本 @@ -63,7 +63,7 @@ worker 暴露的是真正用于 `tools` 绑定失败的 `ToolCallError` 构造 日志会在产生时立即流出，因此运行被终止时仍可保留已经纳入额度的输出。绕过 worker 中已改写流写入入口的原生 stdout 和 stderr 写入会经由彼此独立的管道传输，因此运行时在终态结算期间仍会继续在上限内捕获输出，直至 worker 完全终止，然后才组装结果。超过上限后，运行时会返回一个显式的有界失败，并携带可容纳的已捕获前缀。该外层结果随后通过普通的 `run_code` 渲染与输出落盘策略；策略可以保存已捕获的文本，并暴露其配置指定的头尾预览。输出落盘层无法恢复运行时在硬上限之外拒绝的字节。 -计算时间、墙钟时间、worker 堆内存、取消和每次运行使用全新 worker 的隔离仍是互相独立的限制。外层账本从不计入中间绑定值，因此这些值实际受结构化克隆开销以及进程或 worker 可用内存限制。 +计算时间、墙钟时间、worker 堆内存、取消和每次运行使用全新 worker 的隔离仍是互相独立的限制。外层账本从不计入中间绑定值，因此生成快照、扁平协议格式的编码与解码、结构化克隆开销，以及进程或 worker 的可用内存构成了这些值的实际边界。 ### 类型化句柄与生命周期 @@ -97,14 +97,14 @@ worker 暴露的是真正用于 `tools` 绑定失败的 `ToolCallError` 构造 Code Mode 程序可以通过稳定值组合工具，无需逆向解析 Native 自然语言。Native 和 Both Mode 保留现有文本与编辑器展示，Code Mode 则获得输出 schema 类型和精确的运行时 JSON。工具作者必须把规范值视为程序化 API，并将仅用于展示的格式化放入渲染器。 -worker 会执行结构化克隆和无损校验，但不会降低中间值的开销，也不会使其具备持久性。外层输出溢出会显式导致运行失败，错误处理则有意由人类引导，而不是依赖带版本的错误代码联合。 +worker 会以嵌套深度有界的扁平协议格式传输数据并执行无损校验，但不会降低中间值的开销，也不会使其具备持久性。外层输出溢出会显式导致运行失败，错误处理则有意由人类引导，而不是依赖带版本的错误代码联合。 ## 已知限制与延后工作 - 即使工具输出可以采用任意 JSON 根，subagent 和工作流中由调用方定义的结构化输出仍通过消费方级别的门禁保持对象根限制。 - Post-execute 分别提供值投影与展示投影；替换内容不是保密机制，因此策略若需向程序化调用方隐藏内容，就必须阻止调用或替换值。 - 中间规范值仅存在于执行期间，无法用于回放，因为持久事件只存储展示和有界摘要。 -- 中间值没有字节上限，可能因保留成本或结构化克隆开销而耗尽进程或 worker 内存。 +- 中间值没有字节上限，可能因值的保留、扁平协议格式副本或结构化克隆开销而耗尽进程或 worker 内存。 - 64 MiB 硬上限只适用于外层输出；输出落盘无法恢复超出该上限后被拒绝的字节。 - 提供方或执行器的采集上限可能在规范值到达 Code Mode 前就已丢弃部分源数据。 - 不支持的 MCP 输出 schema 会回退为 `JsonValue`；更丰富的 Native 多媒体投影留待后续实现。 diff --git a/docs/config-catalog.md b/docs/config-catalog.md index a9c8e70105..10a4430ff2 100644 --- a/docs/config-catalog.md +++ b/docs/config-catalog.md @@ -301,7 +301,7 @@ export interface Config { } ``` -Source: [`packages/code-runtime/code-runtime-worker/src/index.ts:22`](../packages/code-runtime/code-runtime-worker/src/index.ts) +Source: [`packages/code-runtime/code-runtime-worker/src/index.ts:24`](../packages/code-runtime/code-runtime-worker/src/index.ts) ## `@deepseek-ai/dsh-compact-basic` diff --git a/packages/code-runtime/code-runtime-worker/README.md b/packages/code-runtime/code-runtime-worker/README.md index 3e69d0ae0f..f4e7e6ebda 100644 --- a/packages/code-runtime/code-runtime-worker/README.md +++ b/packages/code-runtime/code-runtime-worker/README.md @@ -22,14 +22,14 @@ Every field is validated and defaulted; `maxOutputBytes` is a safe integer of at - **Type-strip host-side, in execution context** — the program is wrapped in an async-function shell, stripped with `node:module`'s `stripTypeScriptTypes` (erasable syntax only — `enum`/namespaces are rejected as a program `exception` and no worker spawns), and sliced back out byte-positioned; it then executes as the body of an `AsyncFunction`, so top-level `await`/`return` work. - **The port assumes a hostile peer** — model code can reach `parentPort` and forge traffic, so every inbound message is shape-validated and REBUILT before anything reads it (`null`, primitives, junk types, and malformed payloads drop without a throw; forged extra fields never ride along), the host answers each call id at most once, resolves binding names as OWN properties only (a forged `constructor` cannot walk a prototype chain), drops post-settlement replies, and validates every binding resolution and completion as lossless JSON. Forged `log`/`done` messages cannot bypass the outer cap: the host repeats validation and accounts every admitted log plus the completion or diagnostic. Worker-side namespaces are null-prototype with `defineProperty`, so `__proto__`-shaped binding names are ordinary keys. - **Two independent budgets, because the peer is hostile** — `computeMs` meters the worker's MEASURED busy time (`worker.performance.eventLoopUtilization()` polling): a hot loop cannot hide behind a pending decoy dispatch, and a program awaiting a slow tool accrues nothing. `maxWallMs` backstops what busy time cannot see (awaiting a promise nobody resolves). Both funnel into `worker.terminate()`, which ends hot synchronous loops too; heap overflow surfaces as the worker's OOM exit (`kind: 'worker-exit'`). -- **Intermediate binding values are complete JSON** — binding arguments and resolutions cross by structured clone after iterative lossless-JSON validation and have no byte or call-stack depth cap. They never enter the outer-output ledger or model context; provider/executor acquisition bounds and process/worker memory remain the limits. +- **Intermediate binding values are complete JSON** — binding arguments and resolutions undergo iterative lossless-JSON validation, flatten into a bounded-depth pre-order wire value for structured clone, and rebuild iteratively on the other side. They have no byte, JavaScript call-stack, or nested structured-clone depth cap. They never enter the outer-output ledger or model context; provider/executor acquisition bounds and process/worker memory remain the limits. - **Logs stream eagerly into one outer ledger** — console/stdout/stderr text crosses the port in emission order, so a timed-out or killed program still shows what it printed. Native writes that bypass the patched stream slots arrive on pipes independent of the completion port; settlement therefore continues bounded pipe capture until worker termination completes before materializing the result. `maxOutputBytes` accounts the JSON serialization of the outer `logs` array plus the completion value or failure diagnostic. At or below the cap the exact value returns; a lossy completion is `invalid-output`, and a combined overflow is `output-limit` rather than a substituted inspected string. The failure retains the fitting captured prefix and later follows the normal outer `run_code` spill policy. - **Empty environment** — the worker gets `env: {}` and `execArgv: []`: no ambient credentials (stronger than the scrubbed-env rule for spawned commands) and no inherited loader flags. - **Dispose to quiescence** — teardown fails in-flight runs as `abort` and AWAITS each worker's exit before resolving. ## The worker entry, unbuilt and built -Source mode loads erasable-only `src/worker.ts` through Node's native type stripping. Its transitive runtime closure contains only Node built-ins and relative source modules, so a fresh checkout never requires a sibling workspace package's unbuilt `lib/` export. The worker-local JSON snapshotter is parity-tested against the session-owned canonical boundary; the host repeats canonical validation after structured clone. Built mode passes the sibling `lib/worker.cjs` as a filesystem path because pkg's VFS Worker hook expects CommonJS; the same path works under ordinary Node. `tests/built-lib.e2e.ts` pins the real load path required by [docs/testing.md](../../../docs/testing.md). +Source mode loads erasable-only `src/worker.ts` through Node's native type stripping. Its transitive runtime closure contains only Node built-ins and relative source modules, so a fresh checkout never requires a sibling workspace package's unbuilt `lib/` export. The worker-local JSON snapshotter is parity-tested against the session-owned canonical boundary; both sides flatten and rebuild validated values around the message port so application nesting never reaches structured clone. Built mode passes the sibling `lib/worker.cjs` as a filesystem path because pkg's VFS Worker hook expects CommonJS; the same path works under ordinary Node. `tests/built-lib.e2e.ts` pins the real load path required by [docs/testing.md](../../../docs/testing.md). The SDK surface is the default/named `WorkerCodeRuntime` class plus `Config`. The operational `./worker` subpath exists only as the packaged spawn entry; the wire protocol and bootstrap helpers are source-private implementation details. diff --git a/packages/code-runtime/code-runtime-worker/src/bootstrap.ts b/packages/code-runtime/code-runtime-worker/src/bootstrap.ts index 28c0919749..bb021370ca 100644 --- a/packages/code-runtime/code-runtime-worker/src/bootstrap.ts +++ b/packages/code-runtime/code-runtime-worker/src/bootstrap.ts @@ -8,7 +8,7 @@ import { inspect } from 'node:util' import type { DoneMessage, ReplyMessage, WorkerBootData, WorkerToHost } from './protocol.ts' import { jsonValueBytesUpTo } from './output-json.ts' -import { snapshotCodeJsonValue } from './worker-json.ts' +import { decodeWorkerJson, encodeWorkerJson, snapshotCodeJsonValue } from './worker-json.ts' /** The port surface the bootstrap needs — satisfied by `parentPort` and by the tests' fake. */ export interface BootstrapPort { @@ -152,7 +152,7 @@ export function truncateUtf8Bytes(text: string, maxBytes: number): string { * * @param value - the program's completion value. * @param maxOutputBytes - the byte cap for the outer result. - * @returns the done-message fragment: `{}` for `undefined`, else `{ value }`. + * @returns the done-message fragment: `{}` for `undefined`, else a flat wire `{ value }`. */ export function prepareCompletion(value: unknown, maxOutputBytes: number): Omit { if (value === undefined) return {} @@ -168,7 +168,7 @@ export function prepareCompletion(value: unknown, maxOutputBytes: number): Omit< if (jsonValueBytesUpTo(snapshot, maxOutputBytes) === undefined) { return { error: { kind: 'output-limit', message: `outer output exceeded ${maxOutputBytes} bytes` } } } - return { value: snapshot } + return { value: encodeWorkerJson(snapshot) } } /** One awaited binding call's settlement handles, keyed by call id in the pending map. */ @@ -208,8 +208,13 @@ export function wireReplies(port: BootstrapPort, pending: Map => { - let detached: unknown + let detached: ReturnType try { detached = snapshotCodeJsonValue(args) } catch { @@ -254,7 +259,7 @@ export function makeNamespaces( }, }) try { - port.postMessage({ type: 'call', id, global, name, args: detached }) + port.postMessage({ type: 'call', id, global, name, args: encodeWorkerJson(detached) }) } catch (error: unknown) { pending.delete(id) const message = `binding arguments must be structured-cloneable: ${error instanceof Error ? error.message : String(error)}` diff --git a/packages/code-runtime/code-runtime-worker/src/index.ts b/packages/code-runtime/code-runtime-worker/src/index.ts index 568d6558ca..80511754e6 100644 --- a/packages/code-runtime/code-runtime-worker/src/index.ts +++ b/packages/code-runtime/code-runtime-worker/src/index.ts @@ -17,6 +17,8 @@ import type { CodeBindingFunction, CodeJsonValue, CodeRunFailure, CodeRunRequest import { snapshotJsonValue } from '@deepseek-ai/dsh-session' import type { ReplyMessage, WorkerBootData, WorkerToHost } from './protocol.ts' import { jsonStringBytesUpTo, jsonValueBytesUpTo, truncateJsonStringBytes } from './output-json.ts' +import { decodeWorkerJson, encodeWorkerJson } from './worker-json.ts' +import type { WorkerJsonWire } from './worker-json.ts' /** Plugin config: every execution cap, changeable from `cordis.yml` (no hardcoded tunables). */ export interface Config { @@ -142,7 +144,7 @@ function parseWorkerMessage(raw: unknown): WorkerToHost | undefined { switch (m.type) { case 'call': { if (typeof m.id !== 'number' || typeof m.global !== 'string' || typeof m.name !== 'string') return undefined - return { type: 'call', id: m.id, global: m.global, name: m.name, args: m.args } + return { type: 'call', id: m.id, global: m.global, name: m.name, args: m.args as WorkerJsonWire } } case 'log': { if (typeof m.text !== 'string') return undefined @@ -150,7 +152,7 @@ function parseWorkerMessage(raw: unknown): WorkerToHost | undefined { } case 'output-limit': return { type: 'output-limit' } case 'done': { - if (m.error === undefined) return { type: 'done', ...m.value !== undefined ? { value: m.value } : {} } + if (m.error === undefined) return { type: 'done', ...m.value !== undefined ? { value: m.value as WorkerJsonWire } : {} } const error = m.error if (typeof error !== 'object' || error === null) return undefined const { kind, message } = error as Record @@ -409,10 +411,7 @@ export class WorkerCodeRuntime extends CodeRuntime { finish(() => output.success([...logs, ...strayLogs])) return } - // The worker-thread boundary has already structured-cloned this - // hostile value, so accessors and proxies cannot survive to throw - // during the lossless-JSON snapshot. - const value = snapshotJsonValue(message.value) as CodeJsonValue | undefined + const value = decodeWorkerJson(message.value) if (value === undefined) { finish(() => output.failure([...logs, ...strayLogs], { kind: 'invalid-output', message: 'program completion must be lossless JSON' })) } else { @@ -442,9 +441,7 @@ export class WorkerCodeRuntime extends CodeRuntime { reply({ type: 'reply', id: message.id, ok: false, message: `unknown binding ${JSON.stringify(`${message.global}.${message.name}`)}` }) return } - // Structured clone has already removed accessors and proxies, so the - // host can repeat the lossless snapshot without a reflective throw. - const args = snapshotJsonValue(message.args) as CodeJsonValue | undefined + const args = decodeWorkerJson(message.args) if (args === undefined) { reply({ type: 'reply', id: message.id, ok: false, message: 'binding arguments must be lossless JSON' }) return @@ -461,7 +458,7 @@ export class WorkerCodeRuntime extends CodeRuntime { if (value === undefined) { reply({ type: 'reply', id: message.id, ok: false, message: 'binding resolution must be lossless JSON' }) } else { - reply({ type: 'reply', id: message.id, ok: true, value }) + reply({ type: 'reply', id: message.id, ok: true, value: encodeWorkerJson(value) }) } } catch (error: unknown) { reply({ type: 'reply', id: message.id, ok: false, message: messageOf(error) }) diff --git a/packages/code-runtime/code-runtime-worker/src/protocol.ts b/packages/code-runtime/code-runtime-worker/src/protocol.ts index a76515a78c..8d8ec54b60 100644 --- a/packages/code-runtime/code-runtime-worker/src/protocol.ts +++ b/packages/code-runtime/code-runtime-worker/src/protocol.ts @@ -5,6 +5,8 @@ * @module @deepseek-ai/dsh-code-runtime-worker/src/protocol */ +import type { WorkerJsonWire } from './worker-json.ts' + /** What the host hands the worker at spawn, via `workerData`. */ export interface WorkerBootData { /** The type-stripped (plain JS) program body. */ @@ -24,8 +26,8 @@ interface CallMessage { global: string /** The function name within the namespace. */ name: string - /** The single argument, structured-clone-plain. */ - args: unknown + /** The single argument as a flat lossless-JSON wire value. */ + args: WorkerJsonWire } /** Worker → host: captured text, streamed eagerly so output survives a mid-run termination (timeout, abort, OOM). */ @@ -43,13 +45,13 @@ interface OutputLimitMessage { * Worker → host: the program settled. `error` carries a program exception * (the only failure the bootstrap itself can report — budgets, aborts, and * substrate death are observed host-side). `value` is present only on a - * clean completion that produced one (already size-capped and - * clone-safe per the bootstrap's value preparation). Logs are NOT carried - * here — they streamed eagerly as {@link LogMessage}s. + * clean completion that produced one, as a flat wire value already + * size-capped and lossless per the bootstrap. Logs are NOT carried here — + * they streamed eagerly as {@link LogMessage}s. */ export interface DoneMessage { type: 'done' - value?: unknown + value?: WorkerJsonWire error?: { kind: 'exception' | 'invalid-output' | 'output-limit'; message: string } } @@ -58,5 +60,5 @@ export type WorkerToHost = CallMessage | LogMessage | OutputLimitMessage | DoneM /** Host → worker: the answer to one {@link CallMessage}. */ export type ReplyMessage = - | { type: 'reply'; id: number; ok: true; value: unknown } + | { type: 'reply'; id: number; ok: true; value: WorkerJsonWire } | { type: 'reply'; id: number; ok: false; message: string } diff --git a/packages/code-runtime/code-runtime-worker/src/worker-json.ts b/packages/code-runtime/code-runtime-worker/src/worker-json.ts index 105cc2a646..7ef4526321 100644 --- a/packages/code-runtime/code-runtime-worker/src/worker-json.ts +++ b/packages/code-runtime/code-runtime-worker/src/worker-json.ts @@ -150,4 +150,185 @@ export function snapshotCodeJsonValue(value: unknown): CodeJsonValue | undefined } return root } + +interface ArrayWireToken { + kind: 'array' + length: number +} + +interface ObjectWireToken { + kind: 'object' + keys: string[] +} + +type WorkerJsonToken = null | boolean | number | string | ArrayWireToken | ObjectWireToken + +/** + * A pre-order, bounded-depth transport for one lossless JSON value. Container + * markers and scalar leaves share one flat token array, so `worker_threads` + * never has to structured-clone the value's application nesting. + */ +export type WorkerJsonWire = WorkerJsonToken[] + +/** + * Flatten one validated JSON value for the worker-thread message port. + * @param value - the lossless JSON value to transport. + * @returns a pre-order token stream whose own nesting is bounded. + */ +export function encodeWorkerJson(value: CodeJsonValue): WorkerJsonWire { + const wire: WorkerJsonWire = [] + const pending: CodeJsonValue[] = [value] + for (let current = pending.pop(); current !== undefined; current = pending.pop()) { + if (current === null || typeof current === 'boolean' || typeof current === 'number' || typeof current === 'string') { + wire.push(current) + continue + } + if (Array.isArray(current)) { + wire.push({ kind: 'array', length: current.length }) + for (let index = current.length - 1; index >= 0; index--) { + const item = current[index] + if (item === undefined) throw new Error('cannot encode a sparse JSON array') + pending.push(item) + } + continue + } + const keys = Object.keys(current) + wire.push({ kind: 'object', keys }) + for (let index = keys.length - 1; index >= 0; index--) { + const key = keys[index] + /* v8 ignore next -- the loop is bounded by the captured key count. */ + if (key === undefined) throw new Error('cannot encode a missing JSON object key') + const item = current[key] + if (item === undefined) throw new Error('cannot encode an undefined JSON object property') + pending.push(item) + } + } + return wire +} + +type DecodeFrame = + | { kind: 'array'; target: CodeJsonValue[]; length: number; index: number } + | { kind: 'object'; target: Record; keys: string[]; index: number } + +/** Whether an array contains exactly its dense indexed slots and `length`. */ +function isDenseArray(value: unknown[]): boolean { + if (!hasPlainArrayPrototype(value) || Reflect.ownKeys(value).length !== value.length + 1) return false + for (let index = 0; index < value.length; index++) { + if (!Object.hasOwn(value, index)) return false + } + return true +} + +/** Return one exact container marker, or reject any extra/missing fields. */ +function containerToken(value: object): ArrayWireToken | ObjectWireToken | undefined { + if (Array.isArray(value) || !hasPlainObjectPrototype(value)) return undefined + const keys = enumerableStringKeys(value) + if (keys === undefined) return undefined + const token = value as Record + if (token.kind === 'array') { + if (keys.length !== 2 || !keys.includes('kind') || !keys.includes('length')) return undefined + const length = token.length + return typeof length === 'number' && Number.isSafeInteger(length) && length >= 0 + ? { kind: 'array', length } + : undefined + } + if (token.kind === 'object') { + if (keys.length !== 2 || !keys.includes('kind') || !keys.includes('keys')) return undefined + const objectKeys = token.keys + if (!Array.isArray(objectKeys) || !isDenseArray(objectKeys)) return undefined + const unique = new Set() + const normalizedKeys: string[] = [] + for (const key of objectKeys as unknown[]) { + if (typeof key !== 'string' || unique.has(key)) return undefined + unique.add(key) + normalizedKeys.push(key) + } + return { kind: 'object', keys: normalizedKeys } + } + return undefined +} + +/** + * Rebuild one lossless JSON value from the flat worker-thread wire format. + * Malformed or incomplete traffic returns `undefined`; traversal is iterative + * and therefore independent of the transported value's application depth. + * @param input - untrusted message-port payload. + * @returns the detached JSON value, or `undefined` when the wire is invalid. + */ +export function decodeWorkerJson(input: unknown): CodeJsonValue | undefined { + try { + if (!Array.isArray(input) || !isDenseArray(input) || input.length === 0) return undefined + const wire = input as unknown[] + const frames: DecodeFrame[] = [] + let root: CodeJsonValue | undefined + let rootAssigned = false + + const attach = (value: CodeJsonValue): boolean => { + const parent = frames.at(-1) + if (!parent) { + if (rootAssigned) return false + root = value + rootAssigned = true + return true + } + /* v8 ignore next -- completed frames are popped before another token can attach. */ + if (parent.index >= (parent.kind === 'array' ? parent.length : parent.keys.length)) return false + if (parent.kind === 'array') { + parent.target.push(value) + } else { + const key = parent.keys[parent.index] + /* v8 ignore next -- object frames are built from validated keys and their exact length. */ + if (key === undefined) return false + Object.defineProperty(parent.target, key, { + value, + enumerable: true, + configurable: true, + writable: true, + }) + } + parent.index += 1 + return true + } + + for (let tokenIndex = 0; tokenIndex < wire.length; tokenIndex++) { + const token = wire[tokenIndex] + let value: CodeJsonValue + let frame: DecodeFrame | undefined + if (token === null || typeof token === 'boolean' || typeof token === 'string') { + value = token + } else if (typeof token === 'number') { + if (!Number.isFinite(token) || Object.is(token, -0)) return undefined + value = token + } else { + if (typeof token !== 'object') return undefined + const marker = containerToken(token) + if (!marker) return undefined + const remainingTokens = wire.length - tokenIndex - 1 + if (marker.kind === 'array') { + if (marker.length > remainingTokens) return undefined + const target: CodeJsonValue[] = [] + value = target + if (marker.length > 0) frame = { kind: 'array', target, length: marker.length, index: 0 } + } else { + if (marker.keys.length > remainingTokens) return undefined + const target: Record = {} + value = target + if (marker.keys.length > 0) frame = { kind: 'object', target, keys: marker.keys, index: 0 } + } + } + if (!attach(value)) return undefined + if (frame) frames.push(frame) + while (frames.length > 0) { + const current = frames.at(-1) + /* v8 ignore next -- the loop condition guarantees a final frame. */ + if (current === undefined) break + if (current.index < (current.kind === 'array' ? current.length : current.keys.length)) break + frames.pop() + } + } + return frames.length === 0 ? root : undefined + } catch { + return undefined + } +} /* jscpd:ignore-end */ diff --git a/packages/code-runtime/code-runtime-worker/tests/bootstrap.spec.ts b/packages/code-runtime/code-runtime-worker/tests/bootstrap.spec.ts index 300b08d85f..78af789169 100644 --- a/packages/code-runtime/code-runtime-worker/tests/bootstrap.spec.ts +++ b/packages/code-runtime/code-runtime-worker/tests/bootstrap.spec.ts @@ -3,6 +3,7 @@ import { EventEmitter } from 'node:events' import { LogBuffer, makeConsoleShim, makeNamespaces, captureStreamWrites, prepareCompletion, runWorkerMain, ToolCallError, truncateUtf8Bytes, wireReplies } from '../src/bootstrap.ts' import type { BootstrapPort, PatchableStream, PendingCall } from '../src/bootstrap.ts' import type { ReplyMessage, WorkerToHost } from '../src/protocol.ts' +import { decodeWorkerJson, encodeWorkerJson } from '../src/worker-json.ts' /** * An in-process stand-in for the worker's parentPort: the test plays the @@ -37,6 +38,11 @@ class FakePort implements BootstrapPort { done(): WorkerToHost | undefined { return this.sent.find(message => message.type === 'done') } + + doneValue(): unknown { + const done = this.done() + return done?.type === 'done' && done.value !== undefined ? decodeWorkerJson(done.value) : undefined + } } function fakeStreams(): { stdout: PatchableStream; stderr: PatchableStream } { @@ -127,7 +133,7 @@ describe('captureStreamWrites', () => { describe('prepareCompletion', () => { it('omits undefined and passes lossless JSON values exactly', () => { expect(prepareCompletion(undefined, 100)).toEqual({}) - expect(prepareCompletion({ a: [1, 'two'] }, 100)).toEqual({ value: { a: [1, 'two'] } }) + expect(prepareCompletion({ a: [1, 'two'] }, 100)).toEqual({ value: encodeWorkerJson({ a: [1, 'two'] }) }) }) it('turns every lossy completion shape into invalid-output', () => { @@ -149,7 +155,7 @@ describe('prepareCompletion', () => { }) it('measures the exact JSON serialization at and over the boundary', () => { - expect(prepareCompletion('€', 5)).toEqual({ value: '€' }) + expect(prepareCompletion('€', 5)).toEqual({ value: encodeWorkerJson('€') }) expect(prepareCompletion('€', 4)).toEqual({ error: { kind: 'output-limit', message: 'outer output exceeded 4 bytes' }, }) @@ -178,9 +184,20 @@ describe('truncateUtf8Bytes', () => { }) describe('makeNamespaces', () => { + it('rejects a malformed success reply instead of resolving a lossy binding value', async () => { + const port = new FakePort() + const pending = new Map() + wireReplies(port, pending) + const result = new Promise((resolve, reject) => { pending.set(1, { resolve, reject }) }) + port.deliver({ type: 'reply', id: 1, ok: true, value: [undefined] as never }) + await expect(result).rejects.toThrow('binding resolution must be lossless JSON') + }) + it('exposes prototype-colliding names as ordinary own properties', async () => { const port = new FakePort() - port.respond = message => message.type === 'call' ? { type: 'reply', id: message.id, ok: true, value: `${message.name}-ok` } : undefined + port.respond = message => message.type === 'call' + ? { type: 'reply', id: message.id, ok: true, value: encodeWorkerJson(`${message.name}-ok`) } + : undefined const pending = new Map() wireReplies(port, pending) const [tools] = makeNamespaces({ namespaces: [{ global: 'tools', names: ['__proto__', 'constructor', 'toString'] }] }, port, pending, { value: 1 }) as [Record Promise>] @@ -268,14 +285,18 @@ describe('makeNamespaces', () => { describe('runWorkerMain', () => { it('runs a program end-to-end: bindings, console, return value', async () => { const port = new FakePort() - port.respond = message => message.type === 'call' ? { type: 'reply', id: message.id, ok: true, value: (message.args as { n: number }).n * 2 } : undefined + port.respond = (message) => { + if (message.type !== 'call') return undefined + const args = decodeWorkerJson(message.args) as { n: number } + return { type: 'reply', id: message.id, ok: true, value: encodeWorkerJson(args.n * 2) } + } await runWorkerMain(port, { ...BOOT, code: 'const doubled = await tools.double({ n: 21 }); console.log("got", doubled); return { doubled };', namespaces: [{ global: 'tools', names: ['double'] }], }, fakeStreams()) expect(port.logs()).toEqual(['got 42']) - expect(port.done()).toEqual({ type: 'done', value: { doubled: 42 } }) + expect(port.doneValue()).toEqual({ doubled: 42 }) }) it('reports worker-side log capture overflow before completing', async () => { @@ -287,7 +308,7 @@ describe('runWorkerMain', () => { }, fakeStreams()) expect(port.sent).toContainEqual({ type: 'log', text: '1234' }) expect(port.sent).toContainEqual({ type: 'output-limit' }) - expect(port.done()).toEqual({ type: 'done', value: null }) + expect(port.doneValue()).toBeNull() }) it('reports a thrown program error on the done message', async () => { @@ -318,10 +339,7 @@ describe('runWorkerMain', () => { code: 'try { await tools.x({}) } catch (error) { return { caught: error instanceof ToolCallError, name: error.name, toolName: error.toolName, message: error.message } }', namespaces: [{ global: 'tools', names: ['x'] }], }, fakeStreams()) - expect(port.done()).toEqual({ - type: 'done', - value: { caught: true, name: 'ToolCallError', toolName: 'x', message: 'denied by host' }, - }) + expect(port.doneValue()).toEqual({ caught: true, name: 'ToolCallError', toolName: 'x', message: 'denied by host' }) expect(new ToolCallError('x', 'nope')).toMatchObject({ name: 'ToolCallError', toolName: 'x', message: 'nope' }) }) @@ -330,15 +348,15 @@ describe('runWorkerMain', () => { port.respond = (message) => { if (message.type !== 'call') return undefined // Deliver a stray reply first; the real one follows. - port.deliver({ type: 'reply', id: 9_999, ok: true, value: 'stray' }) - return { type: 'reply', id: message.id, ok: true, value: 'real' } + port.deliver({ type: 'reply', id: 9_999, ok: true, value: encodeWorkerJson('stray') }) + return { type: 'reply', id: message.id, ok: true, value: encodeWorkerJson('real') } } await runWorkerMain(port, { ...BOOT, code: 'return await tools.x({})', namespaces: [{ global: 'tools', names: ['x'] }], }, fakeStreams()) - expect(port.done()).toEqual({ type: 'done', value: 'real' }) + expect(port.doneValue()).toBe('real') }) it('captures raw stream writes through the patched process streams', async () => { diff --git a/packages/code-runtime/code-runtime-worker/tests/runtime.spec.ts b/packages/code-runtime/code-runtime-worker/tests/runtime.spec.ts index ff630e51f1..f0c8fcf98f 100644 --- a/packages/code-runtime/code-runtime-worker/tests/runtime.spec.ts +++ b/packages/code-runtime/code-runtime-worker/tests/runtime.spec.ts @@ -92,7 +92,7 @@ describe('WorkerCodeRuntime — programs and bindings (real workers)', () => { cursor = Array.isArray(cursor) ? cursor[0] : undefined } expect(cursor).toBe('leaf') - }, 60_000) + }, 15_000) it('reports non-erasable syntax as an exception without spawning a worker', async () => { const { runtime } = await setup() @@ -371,7 +371,7 @@ describe('WorkerCodeRuntime — budgets and containment (real workers)', () => { const { parentPort } = await import('node:worker_threads'); const write = (text) => Object.getPrototypeOf(process.stdout).write.call(process.stdout, text); write('late-pipe-' + 'x'.repeat(100_000)); - parentPort.postMessage({ type: 'done', value: 'done' }); + parentPort.postMessage({ type: 'done', value: ['done'] }); for (;;) {} `, bindings: [], @@ -438,7 +438,7 @@ describe('WorkerCodeRuntime — hostile programs (real workers)', () => { program: ` const { parentPort } = await import('node:worker_threads'); for (let i = 0; i < 50; i++) parentPort.postMessage({ type: 'log', text: 'F'.repeat(100), forged: true }); - parentPort.postMessage({ type: 'done', value: 'V'.repeat(100000) }); + parentPort.postMessage({ type: 'done', value: ['V'.repeat(100000)] }); for (;;) {} `, bindings: [], @@ -482,8 +482,9 @@ describe('WorkerCodeRuntime — hostile programs (real workers)', () => { const result = await runtime.run({ program: ` const { parentPort } = await import('node:worker_threads'); - let value = null; - for (let depth = 0; depth < 3_000; depth++) value = [value]; + const value = []; + for (let depth = 0; depth < 3_000; depth++) value.push({ kind: 'array', length: 1 }); + value.push(null); setTimeout(() => { parentPort.postMessage({ type: 'done', value }) }, 25); // Prevent bootstrap's normal undefined completion from racing the forged terminal. await new Promise(() => {}); @@ -500,7 +501,7 @@ describe('WorkerCodeRuntime — hostile programs (real workers)', () => { } expect(depth).toBe(3_000) expect(value).toBeNull() - }, 60_000) + }, 15_000) it('turns forged over-limit error text into output-limit at the host', async () => { const { runtime } = await setup({ maxOutputBytes: 64 }) diff --git a/packages/code-runtime/code-runtime-worker/tests/source-worker.compat.spec.ts b/packages/code-runtime/code-runtime-worker/tests/source-worker.compat.spec.ts index ee35b0e990..6ef2775114 100644 --- a/packages/code-runtime/code-runtime-worker/tests/source-worker.compat.spec.ts +++ b/packages/code-runtime/code-runtime-worker/tests/source-worker.compat.spec.ts @@ -3,6 +3,7 @@ import { tmpdir } from 'node:os' import { join } from 'node:path' import { Worker } from 'node:worker_threads' import { expect, it } from 'vitest' +import { decodeWorkerJson } from '../src/worker-json.ts' /** * Prove the unbuilt worker is a self-contained source closure. Copying it out @@ -28,7 +29,9 @@ it('boots the source worker without workspace package outputs', async () => { worker?.once('error', reject) }) - expect(message).toEqual({ type: 'done', value: { answer: 42 } }) + expect(message).toMatchObject({ type: 'done' }) + const value = typeof message === 'object' && message !== null ? (message as { value?: unknown }).value : undefined + expect(decodeWorkerJson(value)).toEqual({ answer: 42 }) } finally { if (worker) await worker.terminate() await rm(directory, { recursive: true, force: true }) diff --git a/packages/code-runtime/code-runtime-worker/tests/worker-json.spec.ts b/packages/code-runtime/code-runtime-worker/tests/worker-json.spec.ts index 8481747c4b..012cc1204a 100644 --- a/packages/code-runtime/code-runtime-worker/tests/worker-json.spec.ts +++ b/packages/code-runtime/code-runtime-worker/tests/worker-json.spec.ts @@ -1,7 +1,7 @@ import { runInNewContext } from 'node:vm' import { describe, expect, it } from 'vitest' import { snapshotJsonValue } from '@deepseek-ai/dsh-session' -import { snapshotCodeJsonValue } from '../src/worker-json.ts' +import { decodeWorkerJson, encodeWorkerJson, snapshotCodeJsonValue } from '../src/worker-json.ts' describe('snapshotCodeJsonValue', () => { it('matches the canonical scalar boundary', () => { @@ -142,3 +142,90 @@ describe('snapshotCodeJsonValue', () => { expect(snapshotCodeJsonValue({ after: true })).toEqual({ after: true }) }) }) + +describe('flat worker JSON wire', () => { + it('round-trips every JSON root while preserving object keys and container order', () => { + const withPrototypeKey = Object.create(null) as Record + withPrototypeKey.__proto__ = { safe: true } + const values = [null, false, true, 1.25, 'text', [], {}, [1, { nested: [2] }], withPrototypeKey] + for (const value of values) { + const snapshot = snapshotCodeJsonValue(value) + expect(snapshot).not.toBeUndefined() + expect(decodeWorkerJson(encodeWorkerJson(snapshot!))).toEqual(snapshot) + } + const decoded = decodeWorkerJson(encodeWorkerJson(snapshotCodeJsonValue(withPrototypeKey)!)) as Record + expect(Object.hasOwn(decoded, '__proto__')).toBe(true) + expect(decoded.__proto__).toEqual({ safe: true }) + }) + + it('round-trips deep values through a bounded-depth token array', () => { + let value: unknown = 'leaf' + for (let depth = 0; depth < 5_000; depth++) value = [value] + const snapshot = snapshotCodeJsonValue(value)! + const wire = encodeWorkerJson(snapshot) + expect(wire).toHaveLength(5_001) + + let cursor = decodeWorkerJson(wire) + for (let depth = 0; depth < 5_000; depth++) { + expect(Array.isArray(cursor)).toBe(true) + cursor = Array.isArray(cursor) ? cursor[0] : undefined + } + expect(cursor).toBe('leaf') + }) + + it('rejects malformed, incomplete, lossy, sparse, decorated, and throwing wire values', () => { + const sparse = new Array(1) + const compensatedSparse = new Array(1) + Object.defineProperty(compensatedSparse, 'extra', { value: true }) + const decorated: unknown[] = [null] + Object.defineProperty(decorated, 'extra', { value: true }) + const throwing: unknown[] = [] + Object.defineProperty(throwing, 0, { enumerable: true, get: () => { throw new Error('wire getter') } }) + const decoratedKeys: unknown[] = ['x'] + Object.defineProperty(decoratedKeys, 'extra', { value: true }) + const foreignMarker: Record = { kind: 'array', length: 0 } + Object.setPrototypeOf(foreignMarker, {}) + const hiddenMarker = Object.defineProperty({ kind: 'array', length: 0 }, 'hidden', { value: true }) + + for (const value of [ + undefined, + null, + {}, + [], + sparse, + compensatedSparse, + decorated, + throwing, + [undefined], + [-0], + [Number.NaN], + [Number.POSITIVE_INFINITY], + [1, 2], + [[]], + [foreignMarker], + [hiddenMarker], + [{ kind: 'unknown' }], + [{ kind: 'array' }], + [{ kind: 'array', length: '1' }], + [{ kind: 'array', length: -1 }], + [{ kind: 'array', length: Number.MAX_SAFE_INTEGER + 1 }], + [{ kind: 'array', length: 1 }], + [{ kind: 'array', length: 2 }, { kind: 'array', length: 1 }, null], + [{ kind: 'array', length: 0, extra: true }], + [{ kind: 'object' }], + [{ kind: 'object', keys: 'x' }], + [{ kind: 'object', keys: decoratedKeys }], + [{ kind: 'object', keys: [1] }], + [{ kind: 'object', keys: ['x', 'x'] }, 1, 2], + [{ kind: 'object', keys: ['x'] }], + [{ kind: 'object', keys: [], extra: true }], + ]) { + expect(decodeWorkerJson(value)).toBeUndefined() + } + }) + + it('rejects invalid values passed through a forged static type', () => { + expect(() => encodeWorkerJson([undefined] as never)).toThrow(/sparse JSON array/) + expect(() => encodeWorkerJson({ value: undefined } as never)).toThrow(/undefined JSON object property/) + }) +})