feat: return typed values from Code Mode
This commit is contained in:
@@ -10,20 +10,20 @@ Worker-thread implementation of the [`@deepseek-ai/dsh-code-runtime`](../code-ru
|
||||
config:
|
||||
computeMs: 60000 # busy-time budget (measured event-loop active time)
|
||||
maxWallMs: 600000 # wall-clock ceiling; never pauses for anything
|
||||
maxLogBytes: 65536 # shared byte budget for captured log text
|
||||
maxValueBytes: 32768 # rendered-completion-value cap
|
||||
maxOutputBytes: 67108864 # combined serialized outer-output cap (64 MiB)
|
||||
maxOldGenerationSizeMb: 512 # worker heap cap (resourceLimits)
|
||||
```
|
||||
|
||||
Every field is validated (positive numbers) and defaulted; there are no other tunables.
|
||||
Every field is validated and defaulted; `maxOutputBytes` is a safe integer of at least four bytes, the remaining fields are positive finite numbers, and there are no other tunables.
|
||||
|
||||
## Design
|
||||
|
||||
- **One fresh worker per run, no pooling** — a program's world dies with its worker: no cross-run state to log, state bleed unrepresentable, runs reconstructable from the session log alone.
|
||||
- **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 converts a non-cloneable binding resolution into an error reply. Forged `log`/`done` messages cannot bypass the caps: one host-side ledger bounds everything that lands in `logs`, and the completion value is re-capped host-side. Worker-side namespaces are null-prototype with `defineProperty`, so `__proto__`-shaped binding names are ordinary keys.
|
||||
- **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'`).
|
||||
- **Logs stream eagerly** — console/stdout/stderr text crosses the port in emission order, so a timed-out or killed program still shows what it printed. ONE shared `maxLogBytes` ledger bounds everything: streamed text, forged port traffic, and pipe bytes that bypass the patched streams (appended after), with the overflow marked in-band once.
|
||||
- **Intermediate binding values are complete JSON** — binding arguments and resolutions cross by structured clone after lossless-JSON validation and have no byte 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. `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.
|
||||
|
||||
@@ -35,7 +35,7 @@ The SDK surface is the default/named `WorkerCodeRuntime` class plus `Config`. Th
|
||||
|
||||
## Model Experience
|
||||
|
||||
Indirectly, through Code Mode in [`dsh-tools`](../../core/tools/README.md), which renders this worker's capped printed or returned data and exact `[dsh-code-runtime-worker] log capture truncated at <maxLogBytes> bytes` and `… [truncated]` markers into a retained `run_code` result. Binding traffic and worker internals stay outside context.
|
||||
Indirectly, through Code Mode in [`dsh-tools`](../../core/tools/README.md), which renders the exact outer value when it fits or an explicit `invalid-output` / `output-limit` failure. Only the outer `run_code` result enters model context and its ordinary spill policy; binding traffic and intermediate values remain execution-local.
|
||||
|
||||
#### KV Cache effect
|
||||
|
||||
@@ -47,4 +47,5 @@ No direct invalidation; the named consumer owns any request-prefix changes.
|
||||
- **Type-strip rides Node's experimental `stripTypeScriptTypes` API** — the relied-on behavior is pinned by unit tests, with amaro/sucrase as named drop-in replacements if it shifts.
|
||||
- **`computeMs` expiry can overshoot by up to one poll interval** — busy time is sampled every 25 ms (an internal constant, deliberately not config).
|
||||
- **Programs get a five-method `console` shim** (`log`/`info`/`warn`/`error`/`debug`) — deliberately not Node's full console surface.
|
||||
- **A non-cloneable or oversize completion value does not cross as a value** — it arrives as a bounded, truncation-marked `util.inspect` rendering in `value`'s place.
|
||||
- **Intermediate binding values have no byte cap** — a program can exhaust process or worker memory with a value that never becomes outer output.
|
||||
- **The 64 MiB default is a rejection boundary, not recoverable storage** — outer spill can save only the bounded logs and diagnostic returned after `output-limit`; bytes rejected beyond the runtime cap never reach the spill layer.
|
||||
@@ -27,6 +27,7 @@
|
||||
"license": "BSD-3-Clause",
|
||||
"peerDependencies": {
|
||||
"@deepseek-ai/dsh-code-runtime": "^0.0.1",
|
||||
"@deepseek-ai/dsh-session": "^0.0.1",
|
||||
"cordis": "^4.0.0-rc.7"
|
||||
},
|
||||
"dependencies": {
|
||||
@@ -34,6 +35,7 @@
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-code-runtime": "workspace:^",
|
||||
"@deepseek-ai/dsh-session": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.7"
|
||||
}
|
||||
}
|
||||
@@ -6,8 +6,7 @@
|
||||
*/
|
||||
|
||||
import { inspect } from 'node:util'
|
||||
import { serialize } from 'node:v8'
|
||||
import { logTruncationMarker } from './protocol.ts'
|
||||
import { snapshotJsonValue } from '@deepseek-ai/dsh-session'
|
||||
import type { DoneMessage, ReplyMessage, WorkerBootData, WorkerToHost } from './protocol.ts'
|
||||
|
||||
/** The port surface the bootstrap needs — satisfied by `parentPort` and by the tests' fake. */
|
||||
@@ -30,9 +29,8 @@ export interface PatchableStream {
|
||||
* Ordered text capture under one shared byte budget, delivered to a sink as
|
||||
* each item lands (the real sink streams text over the port eagerly, so
|
||||
* captured output survives a mid-run termination). Once the budget is
|
||||
* exhausted it emits exactly one in-band marker and silently drops everything
|
||||
* after. The cap is a blast-radius bound, so "how much was lost" intentionally
|
||||
* stays unmeasured.
|
||||
* exhausted it emits the fitting prefix and reports the limit once; the host
|
||||
* turns that condition into an explicit `output-limit` run failure.
|
||||
*/
|
||||
export class LogBuffer {
|
||||
private remaining: number
|
||||
@@ -40,12 +38,12 @@ export class LogBuffer {
|
||||
// Explicit fields, not constructor parameter properties: this module loads
|
||||
// under Node's native strip-only mode, which rejects non-erasable syntax —
|
||||
// and parameter properties are non-erasable.
|
||||
private readonly maxBytes: number
|
||||
private readonly sink: (text: string) => void
|
||||
private readonly onLimit: () => void
|
||||
|
||||
constructor(maxBytes: number, sink: (text: string) => void) {
|
||||
this.maxBytes = maxBytes
|
||||
constructor(maxBytes: number, sink: (text: string) => void, onLimit: () => void = () => {}) {
|
||||
this.sink = sink
|
||||
this.onLimit = onLimit
|
||||
this.remaining = maxBytes
|
||||
}
|
||||
|
||||
@@ -58,7 +56,10 @@ export class LogBuffer {
|
||||
const cost = Buffer.byteLength(text, 'utf8')
|
||||
if (cost > this.remaining) {
|
||||
this.truncated = true
|
||||
this.sink(logTruncationMarker(this.maxBytes))
|
||||
const prefix = truncateUtf8Bytes(text, this.remaining)
|
||||
if (prefix.length > 0) this.sink(prefix)
|
||||
this.remaining = 0
|
||||
this.onLimit()
|
||||
return
|
||||
}
|
||||
this.remaining -= cost
|
||||
@@ -144,37 +145,30 @@ export function truncateUtf8Bytes(text: string, maxBytes: number): string {
|
||||
}
|
||||
|
||||
/**
|
||||
* Prepare the program's completion value for the done message: a value whose MEASURED
|
||||
* cross-boundary size fits `maxValueBytes` crosses raw — exact bytes for a string, the
|
||||
* structured-clone wire size (`v8.serialize`) for everything else, so a huge container whose
|
||||
* bounded inspect rendering happens to be small cannot smuggle itself past the cap. Oversized
|
||||
* or non-cloneable values are replaced by a bounded string rendering with an in-band marker.
|
||||
* Prepare the program's completion value for the done message. Only lossless
|
||||
* JSON crosses, and an individually oversized value reports `output-limit`;
|
||||
* the host revalidates both and accounts for the combined outer envelope.
|
||||
*
|
||||
* @param value - the program's completion value.
|
||||
* @param maxValueBytes - the byte cap for the value.
|
||||
* @param maxOutputBytes - the byte cap for the outer result.
|
||||
* @returns the done-message fragment: `{}` for `undefined`, else `{ value }`.
|
||||
*/
|
||||
export function prepareValue(value: unknown, maxValueBytes: number): { value?: unknown } {
|
||||
export function prepareCompletion(value: unknown, maxOutputBytes: number): Omit<DoneMessage, 'type'> {
|
||||
if (value === undefined) return {}
|
||||
if (typeof value === 'string') {
|
||||
if (Buffer.byteLength(value, 'utf8') <= maxValueBytes) return { value }
|
||||
} else {
|
||||
let size: number | undefined
|
||||
try {
|
||||
size = serialize(value).byteLength
|
||||
} catch {
|
||||
// Only the verdict matters: the value has parts the structured-clone
|
||||
// algorithm rejects (functions, classes, …) and must cross as its
|
||||
// rendering instead.
|
||||
size = undefined
|
||||
}
|
||||
if (size !== undefined && size <= maxValueBytes) return { value }
|
||||
let snapshot: unknown
|
||||
try {
|
||||
snapshot = snapshotJsonValue(value)
|
||||
} catch {
|
||||
snapshot = undefined
|
||||
}
|
||||
const rendered = typeof value === 'string' ? value : inspect(value, INSPECT_OPTIONS)
|
||||
const capped = Buffer.byteLength(rendered, 'utf8') > maxValueBytes
|
||||
? `${truncateUtf8Bytes(rendered, maxValueBytes)}… [truncated]`
|
||||
: rendered
|
||||
return { value: capped }
|
||||
if (snapshot === undefined) {
|
||||
return { error: { kind: 'invalid-output', message: 'program completion must be lossless JSON' } }
|
||||
}
|
||||
const size = Buffer.byteLength(JSON.stringify(snapshot), 'utf8')
|
||||
if (size > maxOutputBytes) {
|
||||
return { error: { kind: 'output-limit', message: `outer output exceeded ${maxOutputBytes} bytes` } }
|
||||
}
|
||||
return { value: snapshot }
|
||||
}
|
||||
|
||||
/** One awaited binding call's settlement handles, keyed by call id in the pending map. */
|
||||
@@ -183,6 +177,17 @@ export interface PendingCall {
|
||||
reject(error: Error): void
|
||||
}
|
||||
|
||||
/** Program-visible typed rejection for a failed member of the `tools` namespace. */
|
||||
export class ToolCallError extends Error {
|
||||
override readonly name = 'ToolCallError'
|
||||
readonly toolName: string
|
||||
|
||||
constructor(toolName: string, message: string) {
|
||||
super(message)
|
||||
this.toolName = toolName
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Route host replies into the pending-call map: each reply settles its call
|
||||
* at most once, and a reply for an unknown id (stray, or a duplicate answer
|
||||
@@ -227,12 +232,18 @@ export function makeNamespaces(
|
||||
enumerable: true,
|
||||
value: (args: unknown): Promise<unknown> => new Promise((resolve, reject) => {
|
||||
const id = nextId.value++
|
||||
pending.set(id, { resolve, reject })
|
||||
pending.set(id, {
|
||||
resolve,
|
||||
reject: (error) => {
|
||||
reject(global === 'tools' ? new ToolCallError(name, error.message) : error)
|
||||
},
|
||||
})
|
||||
try {
|
||||
port.postMessage({ type: 'call', id, global, name, args })
|
||||
} catch (error: unknown) {
|
||||
pending.delete(id)
|
||||
reject(new Error(`binding arguments must be structured-cloneable: ${error instanceof Error ? error.message : String(error)}`))
|
||||
const message = `binding arguments must be structured-cloneable: ${error instanceof Error ? error.message : String(error)}`
|
||||
reject(global === 'tools' ? new ToolCallError(name, message) : new Error(message))
|
||||
}
|
||||
}),
|
||||
})
|
||||
@@ -254,7 +265,11 @@ export async function runWorkerMain(
|
||||
data: WorkerBootData,
|
||||
streams: { stdout: PatchableStream; stderr: PatchableStream },
|
||||
): Promise<void> {
|
||||
const logs = new LogBuffer(data.maxLogBytes, (text) => { port.postMessage({ type: 'log', text }) })
|
||||
const logs = new LogBuffer(
|
||||
data.maxOutputBytes,
|
||||
(text) => { port.postMessage({ type: 'log', text }) },
|
||||
() => { port.postMessage({ type: 'output-limit' }) },
|
||||
)
|
||||
captureStreamWrites(logs, streams.stdout)
|
||||
captureStreamWrites(logs, streams.stderr)
|
||||
|
||||
@@ -271,12 +286,12 @@ export async function runWorkerMain(
|
||||
// `AsyncFunction` is not a global. The program body is strict-mode.
|
||||
/* v8 ignore next -- the arrow exists only to reach the AsyncFunction constructor; it is never invoked. */
|
||||
const AsyncFunction = (async () => {}).constructor as new (...args: string[]) => (...fnArgs: unknown[]) => Promise<unknown>
|
||||
const fn = new AsyncFunction(...data.namespaces.map(namespace => namespace.global), 'console', `'use strict';\n${data.code}`)
|
||||
const value = await fn(...namespaces, consoleShim)
|
||||
done = { type: 'done', ...prepareValue(value, data.maxValueBytes) }
|
||||
const fn = new AsyncFunction(...data.namespaces.map(namespace => namespace.global), 'ToolCallError', 'console', `'use strict';\n${data.code}`)
|
||||
const value = await fn(...namespaces, ToolCallError, consoleShim)
|
||||
done = { type: 'done', ...prepareCompletion(value, data.maxOutputBytes) }
|
||||
} catch (error: unknown) {
|
||||
const message = error instanceof Error ? error.stack ?? error.message : String(error)
|
||||
done = { type: 'done', error: { message } }
|
||||
done = { type: 'done', error: { kind: 'exception', message } }
|
||||
}
|
||||
port.postMessage(done)
|
||||
}
|
||||
@@ -12,9 +12,8 @@ import { fileURLToPath } from 'node:url'
|
||||
import { Context } from 'cordis'
|
||||
import z from 'schemastery'
|
||||
import { CodeRuntime } from '@deepseek-ai/dsh-code-runtime'
|
||||
import type { CodeBindingFunction, CodeRunFailure, CodeRunRequest, CodeRunResult } from '@deepseek-ai/dsh-code-runtime'
|
||||
import { prepareValue, truncateUtf8Bytes } from './bootstrap.ts'
|
||||
import { logTruncationMarker } from './protocol.ts'
|
||||
import type { CodeBindingFunction, CodeJsonValue, CodeRunFailure, CodeRunRequest, CodeRunResult } from '@deepseek-ai/dsh-code-runtime'
|
||||
import { snapshotJsonValue } from '@deepseek-ai/dsh-session'
|
||||
import type { ReplyMessage, WorkerBootData, WorkerToHost } from './protocol.ts'
|
||||
|
||||
/** Plugin config: every execution cap, changeable from `cordis.yml` (no hardcoded tunables). */
|
||||
@@ -35,14 +34,8 @@ export interface Config {
|
||||
* nobody will resolve).
|
||||
*/
|
||||
maxWallMs?: number
|
||||
/** Shared byte budget for captured log text (console + raw stream writes), truncation marked in-band. */
|
||||
maxLogBytes?: number
|
||||
/**
|
||||
* Byte cap for the completion value, measured by its real cross-boundary
|
||||
* size (string bytes, or structured-clone wire size); an oversized or
|
||||
* non-cloneable value crosses as a capped string rendering.
|
||||
*/
|
||||
maxValueBytes?: number
|
||||
/** Hard cap for the combined serialized outer logs, completion value, and failure diagnostic. */
|
||||
maxOutputBytes?: number
|
||||
/** The worker's max old-generation heap in MiB (`resourceLimits`); overflow kills the worker, surfacing as kind `'worker-exit'`. */
|
||||
maxOldGenerationSizeMb?: number
|
||||
}
|
||||
@@ -59,6 +52,9 @@ type ResolvedConfig = Required<Config>
|
||||
*/
|
||||
const ELU_POLL_INTERVAL_MS = 25
|
||||
|
||||
/** Smallest cap that can represent the empty logs array plus an empty JSON failure diagnostic. */
|
||||
const MIN_OUTPUT_BYTES = 4
|
||||
|
||||
/** ECMAScript reserved words that cannot be async-function parameter names — rejected as binding globals. */
|
||||
const RESERVED_WORDS = new Set([
|
||||
'await', 'break', 'case', 'catch', 'class', 'const', 'continue', 'debugger', 'default', 'delete', 'do',
|
||||
@@ -130,25 +126,74 @@ function parseWorkerMessage(raw: unknown): WorkerToHost | undefined {
|
||||
if (typeof m.text !== 'string') return undefined
|
||||
return { type: 'log', text: m.text }
|
||||
}
|
||||
case 'output-limit': return { type: 'output-limit' }
|
||||
case 'done': {
|
||||
if (m.error === undefined) return { type: 'done', ...m.value !== undefined ? { value: m.value } : {} }
|
||||
const error = m.error
|
||||
if (typeof error !== 'object' || error === null) return undefined
|
||||
const message = (error as Record<string, unknown>).message
|
||||
if (typeof message !== 'string') return undefined
|
||||
return { type: 'done', ...m.value !== undefined ? { value: m.value } : {}, error: { message } }
|
||||
const { kind, message } = error as Record<string, unknown>
|
||||
if ((kind !== 'exception' && kind !== 'invalid-output' && kind !== 'output-limit') || typeof message !== 'string') return undefined
|
||||
return { type: 'done', error: { kind, message } }
|
||||
}
|
||||
default: return undefined
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Headroom the host's value re-cap grants over `maxValueBytes`: exactly the
|
||||
* truncation suffix {@link prepareValue} appends, so a value the WORKER
|
||||
* already capped (byte-exact prefix + this marker) passes through unchanged
|
||||
* instead of being marked twice.
|
||||
*/
|
||||
const VALUE_RENDER_SLACK = Buffer.byteLength('… [truncated]', 'utf8')
|
||||
|
||||
/** Serialized byte size of one lossless JSON value. */
|
||||
function jsonBytes(value: CodeJsonValue): number {
|
||||
return Buffer.byteLength(JSON.stringify(value), 'utf8')
|
||||
}
|
||||
|
||||
/** One run's combined outer-output ledger; binding values never enter it. */
|
||||
class OutputLedger {
|
||||
private bytes = 2 // JSON serialization of the empty logs array: []
|
||||
private entries = 0
|
||||
|
||||
constructor(private readonly maxBytes: number) {}
|
||||
|
||||
/** Admit one exact log entry, or report that the hard cap was crossed. */
|
||||
admit(text: string, sink: string[]): boolean {
|
||||
const cost = Buffer.byteLength(JSON.stringify(text), 'utf8') + (this.entries > 0 ? 1 : 0)
|
||||
if (this.bytes + cost > this.maxBytes) return false
|
||||
this.bytes += cost
|
||||
this.entries += 1
|
||||
sink.push(text)
|
||||
return true
|
||||
}
|
||||
|
||||
/** Finalize a successful absent-or-JSON completion against the combined cap. */
|
||||
success(logs: string[], value?: CodeJsonValue): CodeRunResult {
|
||||
if (value !== undefined && this.bytes + jsonBytes(value) > this.maxBytes) return this.limit(logs)
|
||||
return { logs, ...value !== undefined ? { value } : {} }
|
||||
}
|
||||
|
||||
/** Finalize a failure diagnostic, with output-limit taking precedence when combined bytes exceed the cap. */
|
||||
failure(logs: string[], error: CodeRunFailure): CodeRunResult {
|
||||
if (this.bytes + Buffer.byteLength(JSON.stringify(error.message), 'utf8') > this.maxBytes) return this.limit(logs)
|
||||
return { logs, error }
|
||||
}
|
||||
|
||||
/** Build the explicit output-limit failure while retaining the fitting log prefix. */
|
||||
limit(logs: string[]): CodeRunResult {
|
||||
const fullMessage = `outer output exceeded ${this.maxBytes} bytes`
|
||||
let retainedBytes = this.bytes
|
||||
const messageBytes = Buffer.byteLength(JSON.stringify(fullMessage), 'utf8')
|
||||
while (logs.length > 0 && retainedBytes + messageBytes > this.maxBytes) {
|
||||
const removed = logs.pop()
|
||||
/* v8 ignore next -- the while guard proves pop cannot return undefined. */
|
||||
if (removed === undefined) throw new Error('output ledger lost its final log entry')
|
||||
retainedBytes -= Buffer.byteLength(JSON.stringify(removed), 'utf8') + (logs.length > 0 ? 1 : 0)
|
||||
}
|
||||
const availableMessageBytes = this.maxBytes - retainedBytes
|
||||
// This fixed diagnostic is ASCII with no JSON escapes, so two bytes are
|
||||
// the surrounding quotes and every retained character costs one byte.
|
||||
const message = messageBytes <= availableMessageBytes
|
||||
? fullMessage
|
||||
: fullMessage.slice(0, availableMessageBytes - 2)
|
||||
return { logs, error: { kind: 'output-limit', message } }
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* The shipped {@link CodeRuntime} backend (`ctx.codeRuntime`). Registers as
|
||||
@@ -161,8 +206,7 @@ export class WorkerCodeRuntime extends CodeRuntime {
|
||||
static Config: z<Config> = z.object({
|
||||
computeMs: z.number().default(60_000),
|
||||
maxWallMs: z.number().default(600_000),
|
||||
maxLogBytes: z.number().default(65_536),
|
||||
maxValueBytes: z.number().default(32_768),
|
||||
maxOutputBytes: z.number().default(67_108_864),
|
||||
maxOldGenerationSizeMb: z.number().default(512),
|
||||
})
|
||||
|
||||
@@ -181,6 +225,9 @@ export class WorkerCodeRuntime extends CodeRuntime {
|
||||
for (const [key, value] of Object.entries(this.config)) {
|
||||
if (!(Number.isFinite(value) && value > 0)) throw new Error(`dsh-code-runtime-worker: config.${key} must be a positive number, got ${String(value)}`)
|
||||
}
|
||||
if (!Number.isSafeInteger(this.config.maxOutputBytes) || this.config.maxOutputBytes < MIN_OUTPUT_BYTES) {
|
||||
throw new Error(`dsh-code-runtime-worker: config.maxOutputBytes must be a safe integer of at least ${MIN_OUTPUT_BYTES}, got ${String(this.config.maxOutputBytes)}`)
|
||||
}
|
||||
ctx.effect(() => () => this.teardown(), 'worker code-runtime teardown')
|
||||
}
|
||||
|
||||
@@ -232,7 +279,7 @@ export class WorkerCodeRuntime extends CodeRuntime {
|
||||
if (!IDENTIFIER.test(namespace.global) || RESERVED_WORDS.has(namespace.global)) {
|
||||
throw new Error(`dsh-code-runtime-worker: binding global ${JSON.stringify(namespace.global)} is not a usable identifier`)
|
||||
}
|
||||
if (namespace.global === 'console' || bindings.has(namespace.global)) {
|
||||
if (namespace.global === 'console' || namespace.global === 'ToolCallError' || bindings.has(namespace.global)) {
|
||||
throw new Error(`dsh-code-runtime-worker: duplicate binding global ${JSON.stringify(namespace.global)}`)
|
||||
}
|
||||
bindings.set(namespace.global, namespace.functions)
|
||||
@@ -249,8 +296,7 @@ export class WorkerCodeRuntime extends CodeRuntime {
|
||||
const bootData: WorkerBootData = {
|
||||
code,
|
||||
namespaces: [...bindings].map(([global, functions]) => ({ global, names: Object.keys(functions) })),
|
||||
maxLogBytes: this.config.maxLogBytes,
|
||||
maxValueBytes: this.config.maxValueBytes,
|
||||
maxOutputBytes: this.config.maxOutputBytes,
|
||||
}
|
||||
const worker = new Worker(WORKER_PATH, {
|
||||
workerData: bootData,
|
||||
@@ -274,28 +320,13 @@ export class WorkerCodeRuntime extends CodeRuntime {
|
||||
const answered = new Set<number>()
|
||||
const logs: string[] = []
|
||||
const strayLogs: string[] = []
|
||||
|
||||
// One host-side budget covers normal, forged, and stray-pipe log entries. The first
|
||||
// overflow emits the shared in-band marker and drops everything after it.
|
||||
let logBudget = this.config.maxLogBytes
|
||||
let logsTruncated = false
|
||||
const admit = (text: string, sink: string[]): void => {
|
||||
if (logsTruncated) return
|
||||
const cost = Buffer.byteLength(text, 'utf8')
|
||||
if (cost > logBudget) {
|
||||
logsTruncated = true
|
||||
sink.push(logTruncationMarker(this.config.maxLogBytes))
|
||||
return
|
||||
}
|
||||
logBudget -= cost
|
||||
sink.push(text)
|
||||
}
|
||||
const output = new OutputLedger(this.config.maxOutputBytes)
|
||||
|
||||
// No settled guard: `finish` snapshots the arrays when it resolves, so
|
||||
// a chunk flushing after settlement mutates only the discarded buffers,
|
||||
// and the ledger bounds that growth until the pipes close.
|
||||
const captureStray = (chunk: Buffer): void => {
|
||||
admit(chunk.toString('utf8'), strayLogs)
|
||||
if (!settled && !output.admit(chunk.toString('utf8'), strayLogs)) finish(output.limit([...logs, ...strayLogs]))
|
||||
}
|
||||
worker.stdout.on('data', captureStray)
|
||||
worker.stderr.on('data', captureStray)
|
||||
@@ -304,7 +335,7 @@ export class WorkerCodeRuntime extends CodeRuntime {
|
||||
// logs captured before timeout, abort, or failure remain in the result.
|
||||
let finishResolve!: () => void
|
||||
const finished = new Promise<void>((done) => { finishResolve = done })
|
||||
const finish = (result: Omit<CodeRunResult, 'logs'>): void => {
|
||||
const finish = (result: CodeRunResult): void => {
|
||||
if (settled) return
|
||||
settled = true
|
||||
clearInterval(eluTimer)
|
||||
@@ -313,18 +344,28 @@ export class WorkerCodeRuntime extends CodeRuntime {
|
||||
this.live.delete(live)
|
||||
void worker.terminate().then(() => {
|
||||
finishResolve()
|
||||
resolve({ ...result, logs: [...logs, ...strayLogs] })
|
||||
resolve(result)
|
||||
})
|
||||
}
|
||||
|
||||
const onDone = (message: WorkerToHost): void => {
|
||||
if (message.type !== 'done') return
|
||||
// Re-cap forged completion traffic at the hostile boundary. Honest worker-capped values
|
||||
// pass unchanged via VALUE_RENDER_SLACK; error text is bounded too.
|
||||
finish({
|
||||
...prepareValue(message.value, this.config.maxValueBytes + VALUE_RENDER_SLACK),
|
||||
...message.error ? { error: { kind: 'exception' as const, message: truncateUtf8Bytes(message.error.message, this.config.maxValueBytes) } } : {},
|
||||
})
|
||||
const captured = [...logs, ...strayLogs]
|
||||
if (message.error) {
|
||||
finish(output.failure(captured, message.error))
|
||||
return
|
||||
}
|
||||
if (message.value === undefined) {
|
||||
finish(output.success(captured))
|
||||
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
|
||||
finish(value === undefined
|
||||
? output.failure(captured, { kind: 'invalid-output', message: 'program completion must be lossless JSON' })
|
||||
: output.success(captured, value))
|
||||
}
|
||||
|
||||
const onCall = (message: WorkerToHost): void => {
|
||||
@@ -336,13 +377,9 @@ export class WorkerCodeRuntime extends CodeRuntime {
|
||||
answered.add(message.id)
|
||||
const reply = (payload: ReplyMessage): void => {
|
||||
if (settled) return
|
||||
try {
|
||||
worker.postMessage(payload)
|
||||
} catch {
|
||||
// The reply value failed structured clone; renegotiate as an error
|
||||
// reply, which is always clone-plain. Nothing else throws here.
|
||||
worker.postMessage({ type: 'reply', id: message.id, ok: false, message: 'binding resolution is not structured-cloneable' })
|
||||
}
|
||||
// Canonical resolutions were snapshotted as lossless JSON before
|
||||
// this point, so this payload is structured-cloneable by contract.
|
||||
worker.postMessage(payload)
|
||||
}
|
||||
const record = bindings.get(message.global)
|
||||
// Own-property lookup only: a forged name like 'constructor' or
|
||||
@@ -355,7 +392,18 @@ export class WorkerCodeRuntime extends CodeRuntime {
|
||||
}
|
||||
void (async () => {
|
||||
try {
|
||||
reply({ type: 'reply', id: message.id, ok: true, value: await fn(message.args) })
|
||||
const resolved = await fn(message.args)
|
||||
let value: CodeJsonValue | undefined
|
||||
try {
|
||||
value = snapshotJsonValue(resolved)
|
||||
} catch {
|
||||
value = undefined
|
||||
}
|
||||
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 })
|
||||
}
|
||||
} catch (error: unknown) {
|
||||
reply({ type: 'reply', id: message.id, ok: false, message: messageOf(error) })
|
||||
}
|
||||
@@ -367,15 +415,22 @@ export class WorkerCodeRuntime extends CodeRuntime {
|
||||
// this listener would crash the host process. Junk drops silently.
|
||||
const message = parseWorkerMessage(raw)
|
||||
if (!message) return
|
||||
if (message.type === 'log' && !settled) admit(message.text, logs)
|
||||
if (message.type === 'log' && !settled && !output.admit(message.text, logs)) {
|
||||
finish(output.limit([...logs, ...strayLogs]))
|
||||
return
|
||||
}
|
||||
if (message.type === 'output-limit' && !settled) {
|
||||
finish(output.limit([...logs, ...strayLogs]))
|
||||
return
|
||||
}
|
||||
onCall(message)
|
||||
onDone(message)
|
||||
})
|
||||
worker.on('error', (error: Error) => {
|
||||
finish({ error: { kind: 'worker-exit', message: `worker error: ${error.message}` } })
|
||||
finish(output.failure([...logs, ...strayLogs], { kind: 'worker-exit', message: `worker error: ${error.message}` }))
|
||||
})
|
||||
worker.on('exit', (exitCode: number) => {
|
||||
finish({ error: { kind: 'worker-exit', message: `worker exited with code ${exitCode} before completing` } })
|
||||
finish(output.failure([...logs, ...strayLogs], { kind: 'worker-exit', message: `worker exited with code ${exitCode} before completing` }))
|
||||
})
|
||||
|
||||
// The compute budget reads the worker's own measured busy time, so a
|
||||
@@ -384,21 +439,21 @@ export class WorkerCodeRuntime extends CodeRuntime {
|
||||
const eluTimer = setInterval(() => {
|
||||
const elu = worker.performance.eventLoopUtilization()
|
||||
if (elu.active > this.config.computeMs) {
|
||||
finish({ error: { kind: 'timeout', message: `compute budget exhausted (${this.config.computeMs}ms busy)` } })
|
||||
finish(output.failure([...logs, ...strayLogs], { kind: 'timeout', message: `compute budget exhausted (${this.config.computeMs}ms busy)` }))
|
||||
}
|
||||
}, ELU_POLL_INTERVAL_MS)
|
||||
const wallTimer = setTimeout(() => {
|
||||
finish({ error: { kind: 'timeout', message: `wall-clock ceiling reached (${this.config.maxWallMs}ms)` } })
|
||||
finish(output.failure([...logs, ...strayLogs], { kind: 'timeout', message: `wall-clock ceiling reached (${this.config.maxWallMs}ms)` }))
|
||||
}, this.config.maxWallMs)
|
||||
const onAbort = (): void => {
|
||||
finish({ error: { kind: 'abort', message: String(request.signal?.reason) } })
|
||||
finish(output.failure([...logs, ...strayLogs], { kind: 'abort', message: String(request.signal?.reason) }))
|
||||
}
|
||||
request.signal?.addEventListener('abort', onAbort, { once: true })
|
||||
|
||||
const live: LiveRun = {
|
||||
worker,
|
||||
finished,
|
||||
settle: (failure: CodeRunFailure) => { finish({ error: failure }) },
|
||||
settle: (failure: CodeRunFailure) => { finish(output.failure([...logs, ...strayLogs], failure)) },
|
||||
}
|
||||
this.live.add(live)
|
||||
})
|
||||
|
||||
@@ -11,10 +11,8 @@ export interface WorkerBootData {
|
||||
code: string
|
||||
/** Binding namespaces to materialize: the global name plus the function names (functions themselves stay host-side). */
|
||||
namespaces: { global: string; names: string[] }[]
|
||||
/** Shared byte budget for captured log text; exceeding it drops further entries after one in-band marker. */
|
||||
maxLogBytes: number
|
||||
/** Byte cap for the rendered completion value (see the value-preparation contract in bootstrap.ts). */
|
||||
maxValueBytes: number
|
||||
/** Hard cap for the combined serialized outer logs plus completion value or failure diagnostic. */
|
||||
maxOutputBytes: number
|
||||
}
|
||||
|
||||
/** Worker → host: one bridged binding call. */
|
||||
@@ -36,6 +34,11 @@ interface LogMessage {
|
||||
text: string
|
||||
}
|
||||
|
||||
/** Worker → host: worker-side capture or completion measurement exceeded the outer cap. */
|
||||
interface OutputLimitMessage {
|
||||
type: 'output-limit'
|
||||
}
|
||||
|
||||
/**
|
||||
* Worker → host: the program settled. `error` carries a program exception
|
||||
* (the only failure the bootstrap itself can report — budgets, aborts, and
|
||||
@@ -47,26 +50,13 @@ interface LogMessage {
|
||||
export interface DoneMessage {
|
||||
type: 'done'
|
||||
value?: unknown
|
||||
error?: { message: string }
|
||||
error?: { kind: 'exception' | 'invalid-output' | 'output-limit'; message: string }
|
||||
}
|
||||
|
||||
/** Every message the worker sends. */
|
||||
export type WorkerToHost = CallMessage | LogMessage | DoneMessage
|
||||
export type WorkerToHost = CallMessage | LogMessage | OutputLimitMessage | DoneMessage
|
||||
|
||||
/** Host → worker: the answer to one {@link CallMessage}. */
|
||||
export type ReplyMessage =
|
||||
| { type: 'reply'; id: number; ok: true; value: unknown }
|
||||
| { type: 'reply'; id: number; ok: false; message: string }
|
||||
|
||||
/**
|
||||
* The in-band marker entry text announcing that log capture stopped at the
|
||||
* byte budget. Shared wire vocabulary: the worker's LogBuffer emits it when
|
||||
* ITS budget exhausts, and the host emits the identical text when its own
|
||||
* ledger drops an entry first (forged port traffic, stray pipe bytes) — so
|
||||
* a truncated run reads the same however the cap was hit.
|
||||
* @param maxBytes - the configured `maxLogBytes` the marker names.
|
||||
* @returns the marker line.
|
||||
*/
|
||||
export function logTruncationMarker(maxBytes: number): string {
|
||||
return `[dsh-code-runtime-worker] log capture truncated at ${maxBytes} bytes`
|
||||
}
|
||||
@@ -1,6 +1,6 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { EventEmitter } from 'node:events'
|
||||
import { LogBuffer, makeConsoleShim, makeNamespaces, captureStreamWrites, prepareValue, runWorkerMain, truncateUtf8Bytes, wireReplies } from '../src/bootstrap.ts'
|
||||
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'
|
||||
|
||||
@@ -43,19 +43,34 @@ function fakeStreams(): { stdout: PatchableStream; stderr: PatchableStream } {
|
||||
return { stdout: { write: () => true }, stderr: { write: () => true } }
|
||||
}
|
||||
|
||||
const BOOT = { maxLogBytes: 65_536, maxValueBytes: 32_768 }
|
||||
/** Capture one promise rejection without Vitest's intentionally `any` matcher channel. */
|
||||
async function rejectionOf(promise: Promise<unknown>): Promise<unknown> {
|
||||
try {
|
||||
await promise
|
||||
return undefined
|
||||
} catch (error: unknown) {
|
||||
return error
|
||||
}
|
||||
}
|
||||
|
||||
const BOOT = { maxOutputBytes: 65_536 }
|
||||
|
||||
describe('LogBuffer', () => {
|
||||
it('streams entries to the sink until the byte budget, then emits one marker and drops the rest', () => {
|
||||
it('streams entries to the sink until the byte budget, then emits one fitting prefix and reports the limit once', () => {
|
||||
const seen: string[] = []
|
||||
const buffer = new LogBuffer(10, text => seen.push(text))
|
||||
let limits = 0
|
||||
const buffer = new LogBuffer(10, text => seen.push(text), () => { limits += 1 })
|
||||
buffer.push('12345')
|
||||
buffer.push('123456')
|
||||
buffer.push('dropped')
|
||||
expect(seen).toEqual([
|
||||
'12345',
|
||||
'[dsh-code-runtime-worker] log capture truncated at 10 bytes',
|
||||
])
|
||||
expect(seen).toEqual(['12345', '12345'])
|
||||
expect(limits).toBe(1)
|
||||
|
||||
const exactlyFull: string[] = []
|
||||
const fullBuffer = new LogBuffer(4, text => exactlyFull.push(text))
|
||||
fullBuffer.push('1234')
|
||||
fullBuffer.push('no-prefix-fits')
|
||||
expect(exactlyFull).toEqual(['1234'])
|
||||
})
|
||||
})
|
||||
|
||||
@@ -109,45 +124,42 @@ describe('captureStreamWrites', () => {
|
||||
})
|
||||
})
|
||||
|
||||
describe('prepareValue', () => {
|
||||
it('omits undefined, passes small cloneable values raw', () => {
|
||||
expect(prepareValue(undefined, 100)).toEqual({})
|
||||
expect(prepareValue({ a: [1, 'two'] }, 100)).toEqual({ value: { a: [1, 'two'] } })
|
||||
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'] } })
|
||||
})
|
||||
|
||||
it('replaces a non-cloneable value with its rendering', () => {
|
||||
const { value } = prepareValue({ fn: () => 1 }, 1_000)
|
||||
expect(typeof value).toBe('string')
|
||||
expect(value).toContain('fn')
|
||||
it('turns every lossy completion shape into invalid-output', () => {
|
||||
const cyclic: Record<string, unknown> = {}
|
||||
cyclic.self = cyclic
|
||||
const sparse = Array(2)
|
||||
class Exotic { readonly marker = true }
|
||||
for (const value of [{ fn: () => 1 }, -0, Number.POSITIVE_INFINITY, sparse, cyclic, new Exotic()]) {
|
||||
expect(prepareCompletion(value, 1_000)).toEqual({
|
||||
error: { kind: 'invalid-output', message: 'program completion must be lossless JSON' },
|
||||
})
|
||||
}
|
||||
})
|
||||
|
||||
it('replaces an oversized value with a truncation-marked capped rendering', () => {
|
||||
const { value } = prepareValue('x'.repeat(50), 10)
|
||||
expect(value).toBe(`${'x'.repeat(10)}… [truncated]`)
|
||||
it('reports an oversized value instead of substituting rendered text', () => {
|
||||
expect(prepareCompletion('x'.repeat(50), 10)).toEqual({
|
||||
error: { kind: 'output-limit', message: 'outer output exceeded 10 bytes' },
|
||||
})
|
||||
})
|
||||
|
||||
it('measures a container by its structured-clone wire size, not its bounded rendering', () => {
|
||||
// The bounded inspect rendering of a huge array is tiny ("... N more
|
||||
// items"), but its real cross-boundary size is not — the cap must catch
|
||||
// it, replacing the value with that bounded rendering.
|
||||
const huge = new Array(50_000).fill(7)
|
||||
const { value } = prepareValue(huge, 1_000)
|
||||
expect(typeof value).toBe('string')
|
||||
expect(value).toContain('more items')
|
||||
it('measures the exact JSON serialization at and over the boundary', () => {
|
||||
expect(prepareCompletion('€', 5)).toEqual({ value: '€' })
|
||||
expect(prepareCompletion('€', 4)).toEqual({
|
||||
error: { kind: 'output-limit', message: 'outer output exceeded 4 bytes' },
|
||||
})
|
||||
})
|
||||
|
||||
it('caps a multibyte string by UTF-8 bytes, not UTF-16 length', () => {
|
||||
// 4 code units but 12 UTF-8 bytes: a length-counting cap would pass the
|
||||
// full string through untruncated.
|
||||
expect(prepareValue('€€€€', 4)).toEqual({ value: '€… [truncated]' })
|
||||
})
|
||||
|
||||
it('caps a multibyte rendering by UTF-8 bytes too', () => {
|
||||
// Wire size (24-byte string inside an array) exceeds the cap, so the
|
||||
// value crosses as its rendering — whose truncation must also be
|
||||
// byte-exact: "[ '" (3 bytes) + two € (6 bytes) = 9; a third € would
|
||||
// overflow the 10-byte budget.
|
||||
expect(prepareValue(['€€€€€€€€'], 10)).toEqual({ value: "[ '€€… [truncated]" })
|
||||
it('contains a getter failure as invalid-output', () => {
|
||||
const value = Object.defineProperty({}, 'x', { enumerable: true, get() { throw new Error('getter exploded') } })
|
||||
expect(prepareCompletion(value, 1_000)).toEqual({
|
||||
error: { kind: 'invalid-output', message: 'program completion must be lossless JSON' },
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@@ -191,10 +203,35 @@ describe('makeNamespaces', () => {
|
||||
}
|
||||
const pending = new Map<number, PendingCall>()
|
||||
const [tools] = makeNamespaces({ namespaces: [{ global: 'tools', names: ['x'] }] }, throwingPort, pending, { value: 1 }) as [Record<string, (args: unknown) => Promise<unknown>>]
|
||||
await expect(tools.x?.(() => 1)).rejects.toThrow(/structured-cloneable: DataCloneError-ish/)
|
||||
await expect(tools.x?.(() => 1)).rejects.toThrow(/structured-cloneable: raw-clone-failure/)
|
||||
const first = await rejectionOf(tools.x?.(() => 1) ?? Promise.resolve())
|
||||
const second = await rejectionOf(tools.x?.(() => 1) ?? Promise.resolve())
|
||||
expect(first).toMatchObject({ name: 'ToolCallError', toolName: 'x' })
|
||||
expect(second).toMatchObject({ name: 'ToolCallError', toolName: 'x' })
|
||||
expect(first).toBeInstanceOf(ToolCallError)
|
||||
expect(second).toBeInstanceOf(ToolCallError)
|
||||
expect((first as Error).message).toMatch(/DataCloneError-ish/)
|
||||
expect((second as Error).message).toMatch(/raw-clone-failure/)
|
||||
expect(pending.size).toBe(0)
|
||||
})
|
||||
|
||||
it('uses ordinary Error for non-tools namespace failures', async () => {
|
||||
const deniedPort = new FakePort()
|
||||
deniedPort.respond = message => message.type === 'call'
|
||||
? { type: 'reply', id: message.id, ok: false, message: 'helper denied' }
|
||||
: undefined
|
||||
const deniedPending = new Map<number, PendingCall>()
|
||||
wireReplies(deniedPort, deniedPending)
|
||||
const [helpers] = makeNamespaces({ namespaces: [{ global: 'helpers', names: ['x'] }] }, deniedPort, deniedPending, { value: 1 }) as [Record<string, (args: unknown) => Promise<unknown>>]
|
||||
const denied = await rejectionOf(helpers.x?.({}) ?? Promise.resolve())
|
||||
expect(denied).toBeInstanceOf(Error)
|
||||
expect(denied).not.toBeInstanceOf(ToolCallError)
|
||||
|
||||
const clonePort: BootstrapPort = { postMessage: () => { throw new Error('clone failed') }, on: () => {} }
|
||||
const [cloneHelpers] = makeNamespaces({ namespaces: [{ global: 'helpers', names: ['x'] }] }, clonePort, new Map(), { value: 1 }) as [Record<string, (args: unknown) => Promise<unknown>>]
|
||||
const cloneFailure = await rejectionOf(cloneHelpers.x?.(() => 1) ?? Promise.resolve())
|
||||
expect(cloneFailure).toBeInstanceOf(Error)
|
||||
expect(cloneFailure).not.toBeInstanceOf(ToolCallError)
|
||||
})
|
||||
})
|
||||
|
||||
describe('runWorkerMain', () => {
|
||||
@@ -210,11 +247,24 @@ describe('runWorkerMain', () => {
|
||||
expect(port.done()).toEqual({ type: 'done', value: { doubled: 42 } })
|
||||
})
|
||||
|
||||
it('reports worker-side log capture overflow before completing', async () => {
|
||||
const port = new FakePort()
|
||||
await runWorkerMain(port, {
|
||||
maxOutputBytes: 4,
|
||||
code: 'console.log("12345"); return null',
|
||||
namespaces: [],
|
||||
}, fakeStreams())
|
||||
expect(port.sent).toContainEqual({ type: 'log', text: '1234' })
|
||||
expect(port.sent).toContainEqual({ type: 'output-limit' })
|
||||
expect(port.done()).toEqual({ type: 'done', value: null })
|
||||
})
|
||||
|
||||
it('reports a thrown program error on the done message', async () => {
|
||||
const port = new FakePort()
|
||||
await runWorkerMain(port, { ...BOOT, code: 'throw new Error("boom")', namespaces: [] }, fakeStreams())
|
||||
const done = port.done()
|
||||
expect(done?.type).toBe('done')
|
||||
expect(done?.type === 'done' ? done.error?.kind : undefined).toBe('exception')
|
||||
expect(done?.type === 'done' ? done.error?.message : undefined).toContain('boom')
|
||||
expect(done?.type === 'done' ? done.value : undefined).toBeUndefined()
|
||||
})
|
||||
@@ -222,11 +272,11 @@ describe('runWorkerMain', () => {
|
||||
it('renders non-Error throws and stack-less Errors on the done message', async () => {
|
||||
const rawPort = new FakePort()
|
||||
await runWorkerMain(rawPort, { ...BOOT, code: 'throw "raw-throw"', namespaces: [] }, fakeStreams())
|
||||
expect(rawPort.done()).toEqual({ type: 'done', error: { message: 'raw-throw' } })
|
||||
expect(rawPort.done()).toEqual({ type: 'done', error: { kind: 'exception', message: 'raw-throw' } })
|
||||
|
||||
const barePort = new FakePort()
|
||||
await runWorkerMain(barePort, { ...BOOT, code: 'const e = new Error("bare"); e.stack = undefined; throw e', namespaces: [] }, fakeStreams())
|
||||
expect(barePort.done()).toEqual({ type: 'done', error: { message: 'bare' } })
|
||||
expect(barePort.done()).toEqual({ type: 'done', error: { kind: 'exception', message: 'bare' } })
|
||||
})
|
||||
|
||||
it('surfaces a host failure reply as a program-side rejection it can catch', async () => {
|
||||
@@ -234,10 +284,14 @@ describe('runWorkerMain', () => {
|
||||
port.respond = message => message.type === 'call' ? { type: 'reply', id: message.id, ok: false, message: 'denied by host' } : undefined
|
||||
await runWorkerMain(port, {
|
||||
...BOOT,
|
||||
code: 'try { await tools.x({}) } catch (error) { return `caught: ${error.message}` }',
|
||||
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: denied by host' })
|
||||
expect(port.done()).toEqual({
|
||||
type: 'done',
|
||||
value: { caught: true, name: 'ToolCallError', toolName: 'x', message: 'denied by host' },
|
||||
})
|
||||
expect(new ToolCallError('x', 'nope')).toMatchObject({ name: 'ToolCallError', toolName: 'x', message: 'nope' })
|
||||
})
|
||||
|
||||
it('ignores replies for unknown pending ids', async () => {
|
||||
|
||||
@@ -2,7 +2,7 @@ import { describe, expect, it } from 'vitest'
|
||||
import { Context } from 'cordis'
|
||||
import { WorkerCodeRuntime } from '@deepseek-ai/dsh-code-runtime-worker'
|
||||
import type { Config } from '@deepseek-ai/dsh-code-runtime-worker'
|
||||
import type { CodeRunResult } from '@deepseek-ai/dsh-code-runtime'
|
||||
import type { CodeBindingFunction, CodeBindingNamespace, CodeRunResult } from '@deepseek-ai/dsh-code-runtime'
|
||||
|
||||
/**
|
||||
* Integration suite over REAL worker threads (no mocks — workers are cheap
|
||||
@@ -17,8 +17,8 @@ async function setup(config: Config = {}) {
|
||||
}
|
||||
|
||||
/** Convenience: one namespace `tools` with the given functions. */
|
||||
function tools(functions: Record<string, (args: unknown) => Promise<unknown>>) {
|
||||
return [{ global: 'tools', functions }]
|
||||
function tools(functions: Record<string, (args: unknown) => Promise<unknown>>): CodeBindingNamespace[] {
|
||||
return [{ global: 'tools', functions: functions as Record<string, CodeBindingFunction> }]
|
||||
}
|
||||
|
||||
describe('WorkerCodeRuntime — programs and bindings (real workers)', () => {
|
||||
@@ -52,10 +52,10 @@ describe('WorkerCodeRuntime — programs and bindings (real workers)', () => {
|
||||
const result = await runtime.run({
|
||||
program: `
|
||||
const first = await tools.echo({ n: 1 });
|
||||
let caught = '';
|
||||
try { await tools.fail({}) } catch (error) { caught = error.message }
|
||||
let caughtRaw = '';
|
||||
try { await tools.failRaw({}) } catch (error) { caughtRaw = error.message }
|
||||
let caught = {};
|
||||
try { await tools.fail({}) } catch (error) { caught = { isTyped: error instanceof ToolCallError, name: error.name, toolName: error.toolName, message: error.message } }
|
||||
let caughtRaw = {};
|
||||
try { await tools.failRaw({}) } catch (error) { caughtRaw = { name: error.name, toolName: error.toolName, message: error.message } }
|
||||
return { first, caught, caughtRaw };
|
||||
`,
|
||||
bindings: tools({
|
||||
@@ -66,7 +66,11 @@ describe('WorkerCodeRuntime — programs and bindings (real workers)', () => {
|
||||
}),
|
||||
})
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.value).toEqual({ first: { echoed: { n: 1 } }, caught: 'nope', caughtRaw: 'raw-nope' })
|
||||
expect(result.value).toEqual({
|
||||
first: { echoed: { n: 1 } },
|
||||
caught: { isTyped: true, name: 'ToolCallError', toolName: 'fail', message: 'nope' },
|
||||
caughtRaw: { name: 'ToolCallError', toolName: 'failRaw', message: 'raw-nope' },
|
||||
})
|
||||
expect(calls).toEqual([{ n: 1 }])
|
||||
})
|
||||
|
||||
@@ -90,10 +94,11 @@ describe('WorkerCodeRuntime — programs and bindings (real workers)', () => {
|
||||
expect(result.value).toBe('{}')
|
||||
})
|
||||
|
||||
it('replaces a non-cloneable return value with a string rendering', async () => {
|
||||
it('rejects a non-lossless completion instead of replacing it with rendered text', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({ program: 'return { f: () => 1 }', bindings: [] })
|
||||
expect(typeof result.value).toBe('string')
|
||||
expect(result.value).toBeUndefined()
|
||||
expect(result.error).toEqual({ kind: 'invalid-output', message: 'program completion must be lossless JSON' })
|
||||
})
|
||||
|
||||
it('completes a program that returns nothing with no value at all', async () => {
|
||||
@@ -201,30 +206,47 @@ describe('WorkerCodeRuntime — budgets and containment (real workers)', () => {
|
||||
expect(after.value).toBe('alive')
|
||||
}, 30_000)
|
||||
|
||||
it('truncates runaway log output at the byte budget with an in-band marker', async () => {
|
||||
const { runtime } = await setup({ maxLogBytes: 300 })
|
||||
it('fails runaway log output explicitly while retaining a bounded prefix', async () => {
|
||||
const { runtime } = await setup({ maxOutputBytes: 300 })
|
||||
const result = await runtime.run({
|
||||
program: 'for (let i = 0; i < 1000; i++) console.log("spam line", i); return 1',
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.logs.at(-1)).toContain('truncated at 300 bytes')
|
||||
const total = result.logs.reduce((sum, text) => sum + Buffer.byteLength(text, 'utf8'), 0)
|
||||
expect(total).toBeLessThan(1_000)
|
||||
expect(result.error).toEqual({ kind: 'output-limit', message: 'outer output exceeded 300 bytes' })
|
||||
expect(result.value).toBeUndefined()
|
||||
expect(result.logs.length).toBeGreaterThan(0)
|
||||
expect(Buffer.byteLength(JSON.stringify(result.logs), 'utf8')).toBeLessThan(300)
|
||||
})
|
||||
|
||||
it('caps an oversized return value with a truncation marker', async () => {
|
||||
const { runtime } = await setup({ maxValueBytes: 64 })
|
||||
it('fails an oversized return value without substituting a string', async () => {
|
||||
const { runtime } = await setup({ maxOutputBytes: 64 })
|
||||
const result = await runtime.run({ program: 'return "y".repeat(10_000)', bindings: [] })
|
||||
expect(result.value).toBe(`${'y'.repeat(64)}… [truncated]`)
|
||||
expect(result.value).toBeUndefined()
|
||||
expect(result.error).toEqual({ kind: 'output-limit', message: 'outer output exceeded 64 bytes' })
|
||||
})
|
||||
|
||||
it('caps a multibyte return value by UTF-8 bytes, not string length', async () => {
|
||||
// 4 code units, 12 UTF-8 bytes: a length-counting cap would let the full
|
||||
// string cross. The worker's byte-exact capped rendering then passes the
|
||||
// host re-cap unchanged (cap + marker is exactly the granted slack).
|
||||
const { runtime } = await setup({ maxValueBytes: 4 })
|
||||
const result = await runtime.run({ program: 'return "€€€€"', bindings: [] })
|
||||
expect(result.value).toBe('€… [truncated]')
|
||||
it('uses UTF-8 serialized bytes at the exact completion boundary', async () => {
|
||||
const exact = await setup({ maxOutputBytes: 7 })
|
||||
const exactResult = await exact.runtime.run({ program: 'return "€"', bindings: [] })
|
||||
// [] costs two bytes and JSON serialization of "€" costs five.
|
||||
expect(exactResult).toEqual({ logs: [], value: '€' })
|
||||
|
||||
const over = await setup({ maxOutputBytes: 6 })
|
||||
const overResult = await over.runtime.run({ program: 'return "€"', bindings: [] })
|
||||
expect(overResult.error?.kind).toBe('output-limit')
|
||||
})
|
||||
|
||||
it('accounts logs and completion in one exact combined ledger', async () => {
|
||||
// JSON(["abc"]) is seven bytes and JSON("xy") is four.
|
||||
const exact = await setup({ maxOutputBytes: 11 })
|
||||
expect(await exact.runtime.run({ program: 'console.log("abc"); return "xy"', bindings: [] }))
|
||||
.toEqual({ logs: ['abc'], value: 'xy' })
|
||||
|
||||
const over = await setup({ maxOutputBytes: 10 })
|
||||
const result = await over.runtime.run({ program: 'console.log("abc"); return "xy"', bindings: [] })
|
||||
expect(result.value).toBeUndefined()
|
||||
expect(result.error?.kind).toBe('output-limit')
|
||||
expect(Buffer.byteLength(JSON.stringify(result.logs), 'utf8') + Buffer.byteLength(JSON.stringify(result.error?.message), 'utf8')).toBeLessThanOrEqual(10)
|
||||
})
|
||||
|
||||
it('completes a program that awaits its write callback, capturing the chunk', async () => {
|
||||
@@ -241,32 +263,48 @@ describe('WorkerCodeRuntime — budgets and containment (real workers)', () => {
|
||||
expect(result.logs).toContain('flushed')
|
||||
})
|
||||
|
||||
it('caps a huge container whose bounded rendering is small (wire size, not rendering, is what counts)', async () => {
|
||||
it('returns a large JSON container exactly when the outer cap permits it', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({ program: 'return new Array(50_000).fill(7)', bindings: [] })
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(typeof result.value).toBe('string')
|
||||
expect(result.value).toContain('more items')
|
||||
expect(result.value).toEqual(new Array(50_000).fill(7))
|
||||
})
|
||||
|
||||
it('captures pipe writes that bypass the patched write slot as stray logs, capped by the same budget', async () => {
|
||||
const { runtime } = await setup({ maxLogBytes: 4 })
|
||||
it('returns an exact completion at the default 64 MiB combined boundary', async () => {
|
||||
const { runtime } = await setup()
|
||||
// [] costs two bytes and the JSON string contributes two quotes, leaving
|
||||
// exactly this many payload bytes under the 67_108_864-byte default.
|
||||
const result = await runtime.run({ program: 'return "x".repeat(67_108_860)', bindings: [] })
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.logs).toEqual([])
|
||||
expect(result.value).toHaveLength(67_108_860)
|
||||
}, 60_000)
|
||||
|
||||
it('fails one byte over the default 64 MiB combined boundary', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({ program: 'return "x".repeat(67_108_861)', bindings: [] })
|
||||
expect(result.value).toBeUndefined()
|
||||
expect(result.error).toEqual({ kind: 'output-limit', message: 'outer output exceeded 67108864 bytes' })
|
||||
}, 60_000)
|
||||
|
||||
it('accounts pipe writes that bypass the patched write slot in the same outer ledger', async () => {
|
||||
const { runtime } = await setup({ maxOutputBytes: 80 })
|
||||
const result = await runtime.run({
|
||||
// The prototype write bypasses the patched instance and reaches the real pipe. Pauses keep
|
||||
// writes in separate chunks and let both reach the host before settlement.
|
||||
program: `
|
||||
const write = (text) => Object.getPrototypeOf(process.stdout).write.call(process.stdout, text);
|
||||
write('abcd');
|
||||
write('a'.repeat(20));
|
||||
await new Promise(resolve => setTimeout(resolve, 150));
|
||||
write('ef');
|
||||
write('b'.repeat(100));
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
return 1;
|
||||
`,
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.logs).toContain('abcd')
|
||||
expect(result.logs).not.toContain('ef')
|
||||
expect(result.error?.kind).toBe('output-limit')
|
||||
expect(result.logs).toContain('a'.repeat(20))
|
||||
expect(result.logs).not.toContain('b'.repeat(100))
|
||||
}, 15_000)
|
||||
})
|
||||
|
||||
@@ -305,7 +343,8 @@ describe('WorkerCodeRuntime — hostile programs (real workers)', () => {
|
||||
{ type: 'log', text: 7 },
|
||||
{ type: 'log', text: {} },
|
||||
{ type: 'done', error: 5 },
|
||||
{ type: 'done', error: { message: 5 } },
|
||||
{ type: 'done', error: { kind: 'exception', message: 5 } },
|
||||
{ type: 'done', error: { kind: 'invented', message: 'bad kind' } },
|
||||
]) parentPort.postMessage(junk);
|
||||
return await tools.real({});
|
||||
`,
|
||||
@@ -316,10 +355,10 @@ describe('WorkerCodeRuntime — hostile programs (real workers)', () => {
|
||||
expect(result.logs).toEqual([])
|
||||
})
|
||||
|
||||
it('caps forged log floods and forged done values at the configured budgets, dropping forged extra fields', async () => {
|
||||
const { runtime } = await setup({ maxLogBytes: 200, maxValueBytes: 64 })
|
||||
it('fails forged log floods and forged done values through the same outer cap', async () => {
|
||||
const { runtime } = await setup({ maxOutputBytes: 200 })
|
||||
const result = await runtime.run({
|
||||
// Forged messages bypass the worker-side LogBuffer and prepareValue
|
||||
// Forged messages bypass the worker-side LogBuffer and completion check
|
||||
// entirely — only the host-side ledger and re-cap stand between model
|
||||
// code and an unbounded result.
|
||||
program: `
|
||||
@@ -330,53 +369,79 @@ describe('WorkerCodeRuntime — hostile programs (real workers)', () => {
|
||||
`,
|
||||
bindings: [],
|
||||
})
|
||||
expect(typeof result.value).toBe('string')
|
||||
const value = result.value as string
|
||||
expect(value.startsWith('V'.repeat(64))).toBe(true)
|
||||
expect(value.endsWith('… [truncated]')).toBe(true)
|
||||
expect(value.length).toBeLessThan(120)
|
||||
const marker = '[dsh-code-runtime-worker] log capture truncated at 200 bytes'
|
||||
const total = result.logs.reduce((sum, text) => sum + Buffer.byteLength(text, 'utf8'), 0)
|
||||
expect(total).toBeLessThanOrEqual(200 + Buffer.byteLength(marker, 'utf8'))
|
||||
expect(result.logs.at(-1)).toBe(marker)
|
||||
expect(result.value).toBeUndefined()
|
||||
expect(result.error).toEqual({ kind: 'output-limit', message: 'outer output exceeded 200 bytes' })
|
||||
expect(Buffer.byteLength(JSON.stringify(result.logs), 'utf8')).toBeLessThan(200)
|
||||
})
|
||||
|
||||
it('accepts a forged done carrying both value and error (self-sabotage, contained)', async () => {
|
||||
it('drops a malformed forged done carrying both value and error', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: `
|
||||
const { parentPort } = await import('node:worker_threads');
|
||||
parentPort.postMessage({ type: 'done', value: 'lied', error: { message: 'fake failure' } });
|
||||
for (;;) {}
|
||||
parentPort.postMessage({ type: 'done', value: 'lied', error: { kind: 'exception', message: 'fake failure' } });
|
||||
return 'honest';
|
||||
`,
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.value).toBe('lied')
|
||||
expect(result.error).toEqual({ kind: 'exception', message: 'fake failure' })
|
||||
expect(result).toEqual({ logs: [], error: { kind: 'exception', message: 'fake failure' } })
|
||||
})
|
||||
|
||||
it('byte-bounds forged multibyte error text at the host', async () => {
|
||||
// Forged error text bypasses the worker entirely; the host bound is a
|
||||
// BYTE bound (two € = 6 bytes fit an 8-byte cap, a third would not).
|
||||
const { runtime } = await setup({ maxValueBytes: 8 })
|
||||
it('turns forged over-limit error text into output-limit at the host', async () => {
|
||||
const { runtime } = await setup({ maxOutputBytes: 64 })
|
||||
const result = await runtime.run({
|
||||
program: `
|
||||
const { parentPort } = await import('node:worker_threads');
|
||||
parentPort.postMessage({ type: 'done', error: { message: '€'.repeat(1000) } });
|
||||
parentPort.postMessage({ type: 'done', error: { kind: 'exception', message: '€'.repeat(1000) } });
|
||||
for (;;) {}
|
||||
`,
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.error).toEqual({ kind: 'exception', message: '€€' })
|
||||
expect(result.error).toEqual({ kind: 'output-limit', message: 'outer output exceeded 64 bytes' })
|
||||
})
|
||||
|
||||
it('answers a binding whose resolution cannot be cloned with a failure reply', async () => {
|
||||
it('answers a binding whose resolution is not lossless JSON with a typed failure reply', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: 'try { await tools.bad({}) } catch (error) { return error.message }',
|
||||
program: 'try { await tools.bad({}) } catch (error) { return { name: error.name, toolName: error.toolName, message: error.message } }',
|
||||
bindings: tools({ bad: async () => (() => 1) }),
|
||||
})
|
||||
expect(result.value).toContain('not structured-cloneable')
|
||||
expect(result.value).toEqual({ name: 'ToolCallError', toolName: 'bad', message: 'binding resolution must be lossless JSON' })
|
||||
})
|
||||
|
||||
it('contains throwing getters while snapshotting binding resolutions', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: 'try { await tools.bad({}) } catch (error) { return { name: error.name, toolName: error.toolName, message: error.message } }',
|
||||
bindings: tools({ bad: async () => Object.defineProperty({}, 'bad', { enumerable: true, get() { throw new Error('getter exploded') } }) }),
|
||||
})
|
||||
expect(result.value).toEqual({ name: 'ToolCallError', toolName: 'bad', message: 'binding resolution must be lossless JSON' })
|
||||
})
|
||||
|
||||
it('revalidates a forged lossy completion at the host boundary', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: `
|
||||
const { parentPort } = await import('node:worker_threads');
|
||||
parentPort.postMessage({ type: 'done', value: -0 });
|
||||
for (;;) {}
|
||||
`,
|
||||
bindings: [],
|
||||
})
|
||||
expect(result).toEqual({ logs: [], error: { kind: 'invalid-output', message: 'program completion must be lossless JSON' } })
|
||||
})
|
||||
|
||||
it('honors a forged worker-side output-limit signal', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: `
|
||||
const { parentPort } = await import('node:worker_threads');
|
||||
parentPort.postMessage({ type: 'output-limit' });
|
||||
for (;;) {}
|
||||
`,
|
||||
bindings: [],
|
||||
})
|
||||
expect(result).toEqual({ logs: [], error: { kind: 'output-limit', message: 'outer output exceeded 67108864 bytes' } })
|
||||
})
|
||||
|
||||
it('exposes binding names that collide with Object.prototype as ordinary functions', async () => {
|
||||
@@ -392,12 +457,13 @@ describe('WorkerCodeRuntime — hostile programs (real workers)', () => {
|
||||
})
|
||||
|
||||
describe('WorkerCodeRuntime — seam misuse and lifecycle', () => {
|
||||
it('rejects invalid binding globals loudly (identifier, reserved word, duplicate, console)', async () => {
|
||||
it('rejects invalid binding globals loudly (identifier, reserved word, duplicate, reserved injected globals)', async () => {
|
||||
const { runtime } = await setup()
|
||||
const cases: [string, RegExp][] = [
|
||||
['not valid!', /not a usable identifier/],
|
||||
['await', /not a usable identifier/],
|
||||
['console', /duplicate binding global/],
|
||||
['ToolCallError', /duplicate binding global/],
|
||||
]
|
||||
for (const [global, message] of cases) {
|
||||
await expect(runtime.run({ program: 'return 1', bindings: [{ global, functions: {} }] })).rejects.toThrow(message)
|
||||
@@ -413,6 +479,12 @@ describe('WorkerCodeRuntime — seam misuse and lifecycle', () => {
|
||||
await expect(ctx.plugin(WorkerCodeRuntime, { computeMs: -1 })).rejects.toThrow(/positive number/)
|
||||
})
|
||||
|
||||
it('requires maxOutputBytes to fit the smallest outer failure envelope', async () => {
|
||||
const ctx = new Context()
|
||||
await expect(ctx.plugin(WorkerCodeRuntime, { maxOutputBytes: 3 })).rejects.toThrow(/safe integer of at least 4/)
|
||||
await expect(ctx.plugin(WorkerCodeRuntime, { maxOutputBytes: 4.5 })).rejects.toThrow(/safe integer of at least 4/)
|
||||
})
|
||||
|
||||
it('keeps runs isolated: no state survives from one run to the next', async () => {
|
||||
const { runtime } = await setup()
|
||||
await runtime.run({ program: 'globalThis.leak = "value"; return 1', bindings: [] })
|
||||
|
||||
@@ -8,6 +8,9 @@
|
||||
"src"
|
||||
],
|
||||
"references": [
|
||||
{
|
||||
"path": "../../core/session"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/cosmokit"
|
||||
},
|
||||
|
||||
Reference in New Issue
Block a user