Merge remote-tracking branch 'origin/master' into workflow-vm-to-workerthread
This commit is contained in:
imccyu
@@ -0,0 +1,32 @@
|
||||
# @deepseek-ai/dsh-code-runtime-worker
|
||||
|
||||
Worker-thread implementation of the [`@deepseek-ai/dsh-code-runtime`](../code-runtime/README.md) seam: `WorkerCodeRuntime` runs each program in ONE fresh Node `worker_threads.Worker` — TypeScript in, type-stripped host-side, bindings bridged over the message port, `{ value, logs, error? }` out. **Containment, not a security boundary**: trust posture is bash-equivalent by design (the [Code Mode RFC](../../../docs/rfc/proposed/feature/2026-06-15-code-mode.md) § Trust posture), with containment bash does not have — separate isolate, empty environment, heap cap, hard termination.
|
||||
|
||||
## Config
|
||||
|
||||
```yaml
|
||||
- id: code-runtime
|
||||
name: '@deepseek-ai/dsh-code-runtime-worker'
|
||||
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
|
||||
maxOldGenerationSizeMb: 512 # worker heap cap (resourceLimits)
|
||||
```
|
||||
|
||||
Every field is validated (positive numbers) and defaulted; 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.
|
||||
- **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 entries cross the port as they happen, so a timed-out or killed program still shows what it printed. ONE shared `maxLogBytes` ledger bounds everything: streamed entries, forged port traffic, and pipe bytes that bypass the patched streams (appended after), with the overflow marked in-band once.
|
||||
- **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
|
||||
|
||||
`worker.ts` is deliberately erasable-only TypeScript with type-only cross-package imports: unbuilt (vitest/tsx), the host spawns `src/worker.ts` directly and Node's native type stripping loads it; built, the entry ships as the sibling bundle `lib/worker.js` (its own tsdown entry). The built path is pinned by `tests/built-lib.e2e.ts`, the real-load-path guard from [docs/testing.md](../../../docs/testing.md).
|
||||
@@ -0,0 +1,40 @@
|
||||
{
|
||||
"name": "@deepseek-ai/dsh-code-runtime-worker",
|
||||
"description": "Worker-thread implementation of the DeepSeek Harness code-execution seam",
|
||||
"version": "0.0.1",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"main": "lib/index.js",
|
||||
"types": "lib/types/index.d.ts",
|
||||
"exports": {
|
||||
".": {
|
||||
"types": "./lib/types/index.d.ts",
|
||||
"default": "./lib/index.js"
|
||||
},
|
||||
"./worker": {
|
||||
"types": "./lib/types/worker.d.ts",
|
||||
"default": "./lib/worker.js"
|
||||
},
|
||||
"./src/*": "./src/*",
|
||||
"./package.json": "./package.json"
|
||||
},
|
||||
"files": [
|
||||
"lib/index.js",
|
||||
"lib/worker.js",
|
||||
"lib/types/**/*.d.ts",
|
||||
"lib/types/**/*.d.ts.map",
|
||||
"src"
|
||||
],
|
||||
"license": "BSD-3-Clause",
|
||||
"peerDependencies": {
|
||||
"@deepseek-ai/dsh-code-runtime": "^0.0.1",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
},
|
||||
"dependencies": {
|
||||
"schemastery": "^3.18.0"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@deepseek-ai/dsh-code-runtime": "workspace:^",
|
||||
"cordis": "^4.0.0-rc.6"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,301 @@
|
||||
/**
|
||||
* Worker-side execution logic, written as plain functions over an injected
|
||||
* port so the unit suite can run every line IN-PROCESS against a fake port
|
||||
* (a real worker thread is a separate V8 isolate the coverage provider
|
||||
* cannot observe). The real worker entry (`worker.ts`) is a thin
|
||||
* self-executing glue file over {@link runWorkerMain}, excluded from
|
||||
* coverage the same way `bin.ts` entrypoints are, and exercised end-to-end
|
||||
* by the integration tests that spawn real workers.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-code-runtime-worker/src/bootstrap
|
||||
*/
|
||||
|
||||
import { inspect } from 'node:util'
|
||||
import { serialize } from 'node:v8'
|
||||
import type { CodeLogEntry } from '@deepseek-ai/dsh-code-runtime'
|
||||
import { logTruncationMarker } from './protocol.ts'
|
||||
import type { DoneMessage, ReplyMessage, WorkerBootData, WorkerToHost } from './protocol.ts'
|
||||
|
||||
/** The port surface the bootstrap needs — satisfied by `parentPort` and by the tests' fake. */
|
||||
export interface BootstrapPort {
|
||||
postMessage(message: WorkerToHost): void
|
||||
on(event: 'message', listener: (message: ReplyMessage) => void): void
|
||||
}
|
||||
|
||||
/**
|
||||
* A writable stream's `write` slot, as the bootstrap patches it (see
|
||||
* {@link captureStreamWrites}). Method-typed so the real
|
||||
* `process.stdout`/`process.stderr` (narrower chunk parameters) remain
|
||||
* assignable.
|
||||
*/
|
||||
export interface PatchableStream {
|
||||
write(chunk: unknown, ...rest: unknown[]): boolean
|
||||
}
|
||||
|
||||
/**
|
||||
* Ordered log capture under one shared byte budget, delivered to a sink as
|
||||
* each entry lands (the real sink streams entries over the port eagerly, so
|
||||
* captured output survives a mid-run termination). Once the budget is
|
||||
* exhausted it emits exactly one in-band marker entry (on the `stderr`
|
||||
* diagnostics channel) and silently drops everything after — the cap is a
|
||||
* blast-radius bound, so "how much was lost" intentionally stays unmeasured.
|
||||
*/
|
||||
export class LogBuffer {
|
||||
private remaining: number
|
||||
private truncated = false
|
||||
// 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: (entry: CodeLogEntry) => void
|
||||
|
||||
constructor(maxBytes: number, sink: (entry: CodeLogEntry) => void) {
|
||||
this.maxBytes = maxBytes
|
||||
this.sink = sink
|
||||
this.remaining = maxBytes
|
||||
}
|
||||
|
||||
/**
|
||||
* Emit one entry to the sink, charging its text against the budget (drops + marks once exhausted).
|
||||
* @param entry - the log entry to deliver.
|
||||
*/
|
||||
push(entry: CodeLogEntry): void {
|
||||
if (this.truncated) return
|
||||
const cost = Buffer.byteLength(entry.text, 'utf8')
|
||||
if (cost > this.remaining) {
|
||||
this.truncated = true
|
||||
this.sink({ source: 'stderr', text: logTruncationMarker(this.maxBytes) })
|
||||
return
|
||||
}
|
||||
this.remaining -= cost
|
||||
this.sink(entry)
|
||||
}
|
||||
}
|
||||
|
||||
/** The five console methods the shim captures, in the seam's level vocabulary. */
|
||||
const CONSOLE_LEVELS = ['log', 'info', 'warn', 'error', 'debug'] as const
|
||||
|
||||
/**
|
||||
* A `console` replacement whose five leveled methods render their arguments
|
||||
* `util.inspect`-style (matching real console formatting closely enough for
|
||||
* a model to recognize its own output) into the buffer. Only these five
|
||||
* exist — the program gets a deliberately small console, not Node's full
|
||||
* surface.
|
||||
* @param logs - the buffer every rendered line is pushed into.
|
||||
* @returns the five-method console object handed to the program.
|
||||
*/
|
||||
export function makeConsoleShim(logs: LogBuffer): Record<(typeof CONSOLE_LEVELS)[number], (...args: unknown[]) => void> {
|
||||
const render = (args: unknown[]): string =>
|
||||
args.map(arg => typeof arg === 'string' ? arg : inspect(arg, INSPECT_OPTIONS)).join(' ')
|
||||
const shim = Object.create(null) as Record<(typeof CONSOLE_LEVELS)[number], (...args: unknown[]) => void>
|
||||
for (const level of CONSOLE_LEVELS) {
|
||||
shim[level] = (...args: unknown[]) => { logs.push({ source: 'console', level, text: render(args) }) }
|
||||
}
|
||||
return shim
|
||||
}
|
||||
|
||||
/**
|
||||
* Redirect a stream's `write` into the log buffer (the program-visible
|
||||
* `process.stdout`/`process.stderr` in the real worker), so raw writes land
|
||||
* in emission order alongside console output instead of racing down a pipe.
|
||||
* The shim keeps Node's `write(chunk[, encoding][, callback])` contract: the
|
||||
* callback fires asynchronously once the chunk is admitted (a program
|
||||
* awaiting flush completion must complete, not sit until the wall timeout),
|
||||
* even for writes the exhausted budget drops.
|
||||
* @param logs - the buffer captured writes are pushed into.
|
||||
* @param stream - the stream whose `write` slot is patched.
|
||||
* @param source - the log source the captured writes are attributed to.
|
||||
* @returns the restore function (the in-process tests un-patch; the real
|
||||
* worker never needs to).
|
||||
*/
|
||||
export function captureStreamWrites(logs: LogBuffer, stream: PatchableStream, source: 'stdout' | 'stderr'): () => void {
|
||||
// The slot's VALUE is stored for restore and reassigned — never invoked
|
||||
// detached, so the unbound-method concern does not apply.
|
||||
// eslint-disable-next-line @typescript-eslint/unbound-method
|
||||
const original = stream.write
|
||||
stream.write = (chunk: unknown, ...rest: unknown[]): boolean => {
|
||||
logs.push({ source, text: typeof chunk === 'string' ? chunk : String(chunk) })
|
||||
// Node's optional-encoding shape: the callback is whichever of the next
|
||||
// two positions holds a function (a non-function there is the encoding).
|
||||
const callback = [rest[0], rest[1]].find(
|
||||
(arg): arg is (error?: Error | null) => void => typeof arg === 'function',
|
||||
)
|
||||
if (callback) queueMicrotask(() => { callback(null) })
|
||||
return true
|
||||
}
|
||||
return () => { stream.write = original }
|
||||
}
|
||||
|
||||
/** Bounded inspect options: deep enough to be useful, bounded so a pathological value cannot explode the rendering. */
|
||||
const INSPECT_OPTIONS = { depth: 4, maxArrayLength: 100, maxStringLength: 10_000 } as const
|
||||
|
||||
/**
|
||||
* The longest prefix of `text` whose UTF-8 encoding fits `maxBytes`, cut at
|
||||
* a code-point boundary (never mid-surrogate-pair). The byte caps are BYTE
|
||||
* caps — `String.prototype.slice` counts UTF-16 code units, up to 3× smaller
|
||||
* than what a multibyte string actually costs across the boundary.
|
||||
* @param text - the string to bound.
|
||||
* @param maxBytes - the UTF-8 byte budget the prefix must fit.
|
||||
* @returns the prefix (all of `text` when it already fits).
|
||||
*/
|
||||
export function truncateUtf8Bytes(text: string, maxBytes: number): string {
|
||||
if (Buffer.byteLength(text, 'utf8') <= maxBytes) return text
|
||||
let bytes = 0
|
||||
let end = 0
|
||||
for (const char of text) {
|
||||
const cost = Buffer.byteLength(char, 'utf8')
|
||||
if (bytes + cost > maxBytes) break
|
||||
bytes += cost
|
||||
end += char.length
|
||||
}
|
||||
return text.slice(0, end)
|
||||
}
|
||||
|
||||
/**
|
||||
* 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. Anything else
|
||||
* (non-cloneable, or oversized) is REPLACED by its bounded `util.inspect`
|
||||
* rendering, byte-truncated ({@link truncateUtf8Bytes}) with an in-band
|
||||
* marker — the seam contract's "a non-transferable value is replaced by a
|
||||
* string rendering", extended to oversized ones so a huge return cannot
|
||||
* flood the host.
|
||||
* @param value - the program's completion value.
|
||||
* @param maxValueBytes - the byte cap for the value.
|
||||
* @returns the done-message fragment: `{}` for `undefined`, else `{ value }`.
|
||||
*/
|
||||
export function prepareValue(value: unknown, maxValueBytes: number): { value?: unknown } {
|
||||
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 }
|
||||
}
|
||||
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 }
|
||||
}
|
||||
|
||||
/** One awaited binding call's settlement handles, keyed by call id in the pending map. */
|
||||
export interface PendingCall {
|
||||
resolve(value: unknown): void
|
||||
reject(error: Error): void
|
||||
}
|
||||
|
||||
/**
|
||||
* 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
|
||||
* to an id already settled) is ignored. Shared wiring between
|
||||
* {@link runWorkerMain} and the tests that exercise {@link makeNamespaces}
|
||||
* standalone.
|
||||
* @param port - the port whose `message` events carry the replies.
|
||||
* @param pending - the id-keyed map of unsettled binding calls.
|
||||
*/
|
||||
export function wireReplies(port: BootstrapPort, pending: Map<number, PendingCall>): void {
|
||||
port.on('message', (message: ReplyMessage) => {
|
||||
const entry = pending.get(message.id)
|
||||
if (!entry) return
|
||||
pending.delete(message.id)
|
||||
if (message.ok) entry.resolve(message.value)
|
||||
else entry.reject(new Error(message.message))
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Build the binding namespace objects the program sees: one null-prototype
|
||||
* global per namespace, each declared name an own enumerable async function
|
||||
* that bridges over the port (`__proto__`/`constructor`/`toString` are
|
||||
* ordinary keys, never prototype collisions). A non-cloneable argument
|
||||
* rejects that one call with a descriptive error; the host's reply (`ok`
|
||||
* false) rejects it likewise, so a failed tool call surfaces in the program
|
||||
* as an ordinary promise rejection.
|
||||
* @param data - the boot payload's namespace declarations (globals + names).
|
||||
* @param port - the port binding calls are posted to.
|
||||
* @param pending - the id-keyed map each posted call parks its handles in.
|
||||
* @param nextId - the shared mutable id counter (worker-issued correlation ids).
|
||||
* @returns one namespace object per declaration, in declaration order.
|
||||
*/
|
||||
export function makeNamespaces(
|
||||
data: Pick<WorkerBootData, 'namespaces'>,
|
||||
port: BootstrapPort,
|
||||
pending: Map<number, PendingCall>,
|
||||
nextId: { value: number },
|
||||
): Record<string, unknown>[] {
|
||||
return data.namespaces.map(({ global, names }) => {
|
||||
const namespace = Object.create(null) as Record<string, unknown>
|
||||
for (const name of names) {
|
||||
Object.defineProperty(namespace, name, {
|
||||
enumerable: true,
|
||||
value: (args: unknown): Promise<unknown> => new Promise((resolve, reject) => {
|
||||
const id = nextId.value++
|
||||
pending.set(id, { resolve, reject })
|
||||
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)}`))
|
||||
}
|
||||
}),
|
||||
})
|
||||
}
|
||||
return namespace
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Run one program to settlement and post the {@link DoneMessage}: wires the
|
||||
* reply handler, materializes the namespaces and console shim, compiles the
|
||||
* type-stripped body as an async function (top-level `await`/`return`
|
||||
* work), and reports a thrown program error as the done message's `error`
|
||||
* field. Exactly one done message is ever posted.
|
||||
* @param port - the message port to the host (the real `parentPort`, or the tests' fake).
|
||||
* @param data - the boot payload the host sent.
|
||||
* @param streams - the stream objects whose `write` is captured (the real
|
||||
* `process.stdout`/`process.stderr` in the worker; fakes in tests).
|
||||
* @returns resolves after the done message is posted (the tests await it;
|
||||
* the real entry lets the worker exit naturally).
|
||||
*/
|
||||
export async function runWorkerMain(
|
||||
port: BootstrapPort,
|
||||
data: WorkerBootData,
|
||||
streams: { stdout: PatchableStream; stderr: PatchableStream },
|
||||
): Promise<void> {
|
||||
const logs = new LogBuffer(data.maxLogBytes, (entry) => { port.postMessage({ type: 'log', entry }) })
|
||||
captureStreamWrites(logs, streams.stdout, 'stdout')
|
||||
captureStreamWrites(logs, streams.stderr, 'stderr')
|
||||
|
||||
const pending = new Map<number, PendingCall>()
|
||||
wireReplies(port, pending)
|
||||
|
||||
const nextId = { value: 1 }
|
||||
const namespaces = makeNamespaces(data, port, pending, nextId)
|
||||
const consoleShim = makeConsoleShim(logs)
|
||||
|
||||
let done: DoneMessage
|
||||
try {
|
||||
// The async function constructor, reached through an instance because
|
||||
// `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) }
|
||||
} catch (error: unknown) {
|
||||
const message = error instanceof Error ? error.stack ?? error.message : String(error)
|
||||
done = { type: 'done', error: { message } }
|
||||
}
|
||||
port.postMessage(done)
|
||||
}
|
||||
@@ -0,0 +1,441 @@
|
||||
/**
|
||||
* Worker-thread implementation of the code-execution seam: one fresh Node
|
||||
* worker per run, executing the model's TypeScript after a host-side
|
||||
* type-strip, with bindings bridged over the message port. Containment, not
|
||||
* a security boundary (bash-equivalent trust — see the Code Mode RFC's
|
||||
* trust-posture section): the worker gets an EMPTY environment, a heap cap,
|
||||
* and two independent budgets — `computeMs` metered on the worker's
|
||||
* measured event-loop busy time (a hot loop cannot hide behind a pending
|
||||
* binding call) and a never-pausing `maxWallMs` ceiling — all funneling
|
||||
* into `worker.terminate()`, which ends hot synchronous loops too.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-code-runtime-worker
|
||||
*/
|
||||
|
||||
import { Worker } from 'node:worker_threads'
|
||||
import { stripTypeScriptTypes } from 'node:module'
|
||||
import { Context } from 'cordis'
|
||||
import z from 'schemastery'
|
||||
import { CodeRuntime } from '@deepseek-ai/dsh-code-runtime'
|
||||
import type { CodeBindingFunction, CodeLogEntry, CodeRunFailure, CodeRunRequest, CodeRunResult } from '@deepseek-ai/dsh-code-runtime'
|
||||
import { prepareValue, truncateUtf8Bytes } from './bootstrap.ts'
|
||||
import { logTruncationMarker } from './protocol.ts'
|
||||
import type { ReplyMessage, WorkerBootData, WorkerToHost } from './protocol.ts'
|
||||
|
||||
export type { BootstrapPort, PatchableStream } from './bootstrap.ts'
|
||||
export type { CallMessage, DoneMessage, ReplyMessage, WorkerBootData, WorkerToHost } from './protocol.ts'
|
||||
|
||||
/** Plugin config: every execution cap, changeable from `cordis.yml` (no hardcoded tunables). */
|
||||
export interface Config {
|
||||
/**
|
||||
* Busy-time budget in milliseconds: the run fails with kind `'timeout'`
|
||||
* once the worker's MEASURED event-loop active time
|
||||
* (`worker.performance.eventLoopUtilization()`) exceeds this. Metering
|
||||
* measured busy time — not wall time, not host-side pending-call
|
||||
* bookkeeping — is what makes the budget both fair (a program awaiting a
|
||||
* slow tool accrues nothing) and ungameable (a hot loop accrues whether
|
||||
* or not a decoy dispatch is in flight).
|
||||
*/
|
||||
computeMs?: number
|
||||
/**
|
||||
* Wall-clock ceiling in milliseconds; never pauses for anything. The
|
||||
* backstop for what busy-time cannot see (a program awaiting a promise
|
||||
* 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
|
||||
/** The worker's max old-generation heap in MiB (`resourceLimits`); overflow kills the worker, surfacing as kind `'worker-exit'`. */
|
||||
maxOldGenerationSizeMb?: number
|
||||
}
|
||||
|
||||
/** {@link Config} after schemastery fills the defaults (every field present). */
|
||||
type ResolvedConfig = Required<Config>
|
||||
|
||||
/**
|
||||
* How often the host samples the worker's event-loop utilization for the
|
||||
* `computeMs` budget. An internal cadence, not config: the only effect of
|
||||
* the interval is budget-expiry granularity (a run can overshoot by up to
|
||||
* one interval), and nothing a deployment could tune here improves that
|
||||
* without burning host CPU.
|
||||
*/
|
||||
const ELU_POLL_INTERVAL_MS = 25
|
||||
|
||||
/** 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',
|
||||
'else', 'enum', 'export', 'extends', 'false', 'finally', 'for', 'function', 'if', 'import', 'in',
|
||||
'instanceof', 'new', 'null', 'return', 'super', 'switch', 'this', 'throw', 'true', 'try', 'typeof',
|
||||
'var', 'void', 'while', 'with', 'yield', 'let', 'static', 'implements', 'interface', 'package',
|
||||
'private', 'protected', 'public', 'arguments', 'eval',
|
||||
])
|
||||
|
||||
/** Valid async-function parameter name (the binding global becomes one). */
|
||||
const IDENTIFIER = /^[A-Za-z_$][A-Za-z0-9_$]*$/
|
||||
|
||||
/**
|
||||
* The shell a program is wrapped in for the type-strip, matching the
|
||||
* grammatical context it will execute in (an async function body, where
|
||||
* top-level `return` and `await` are legal — a bare module parse would
|
||||
* reject the `return`). Strip mode is position-preserving (removed syntax
|
||||
* becomes whitespace, nothing shifts), so the wrapper survives the strip
|
||||
* byte-identical and the body slices back out with the model's own
|
||||
* line/column positions intact.
|
||||
*/
|
||||
const STRIP_WRAP = { prefix: 'async function __dsh_program__() {\n', suffix: '\n}' } as const
|
||||
|
||||
/** One in-flight run's host-side state, tracked for disposal. */
|
||||
interface LiveRun {
|
||||
worker: Worker
|
||||
settle(failure: CodeRunFailure): void
|
||||
finished: Promise<void>
|
||||
}
|
||||
|
||||
/**
|
||||
* The worker entry module. Source runs unbuilt (`src/worker.ts`, loadable
|
||||
* directly on this repo's Node range via native type stripping — the file
|
||||
* is erasable-only with type-only relative imports); the built package
|
||||
* ships it as a sibling bundle (`lib/worker.js`, its own tsdown entry).
|
||||
* The URL *pathname*'s extension says which world this module is in —
|
||||
* pathname, because dev-time module runners (vitest) may suffix
|
||||
* `import.meta.url` with a query string; relative resolution drops it.
|
||||
*/
|
||||
/* v8 ignore next -- the './worker.js' arm is the built-lib world, unreachable unbuilt by construction; the built-lib e2e pins it. */
|
||||
const WORKER_URL = new URL(new URL(import.meta.url).pathname.endsWith('.ts') ? './worker.ts' : './worker.js', import.meta.url)
|
||||
|
||||
/** Render an unknown thrown value as a message, `Error` or not. */
|
||||
function messageOf(error: unknown): string {
|
||||
return error instanceof Error ? error.message : String(error)
|
||||
}
|
||||
|
||||
/** The log sources / console levels the seam vocabulary admits, as runtime sets for inbound-message validation. */
|
||||
const LOG_SOURCES = new Set<string>(['console', 'stdout', 'stderr'])
|
||||
const LOG_LEVELS = new Set<string>(['log', 'info', 'warn', 'error', 'debug'])
|
||||
|
||||
/**
|
||||
* Runtime shape gate for inbound port traffic. The peer runs MODEL CODE and
|
||||
* can post anything — `null`, primitives, objects with poisoned fields — so
|
||||
* the compile-time `WorkerToHost` type means nothing here: everything is
|
||||
* re-validated and REBUILT field by field (a forged extra field never rides
|
||||
* along; a non-number call id can never be echoed into a reply). Junk returns
|
||||
* `undefined` and is dropped — a throw in the host's `message` listener would
|
||||
* crash the host process.
|
||||
*/
|
||||
function parseWorkerMessage(raw: unknown): WorkerToHost | undefined {
|
||||
if (typeof raw !== 'object' || raw === null) return undefined
|
||||
const m = raw as Record<string, unknown>
|
||||
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 }
|
||||
}
|
||||
case 'log': {
|
||||
const entry = m.entry
|
||||
if (typeof entry !== 'object' || entry === null) return undefined
|
||||
const e = entry as Record<string, unknown>
|
||||
if (typeof e.text !== 'string') return undefined
|
||||
if (typeof e.source !== 'string' || !LOG_SOURCES.has(e.source)) return undefined
|
||||
if (e.level !== undefined && (typeof e.level !== 'string' || !LOG_LEVELS.has(e.level))) return undefined
|
||||
return {
|
||||
type: 'log',
|
||||
entry: {
|
||||
source: e.source as CodeLogEntry['source'],
|
||||
...e.level !== undefined ? { level: e.level as Exclude<CodeLogEntry['level'], undefined> } : {},
|
||||
text: e.text,
|
||||
},
|
||||
}
|
||||
}
|
||||
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 } }
|
||||
}
|
||||
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')
|
||||
|
||||
/**
|
||||
* The shipped {@link CodeRuntime} backend (`ctx.codeRuntime`). Registers as
|
||||
* the `codeRuntime` service; every cap comes from validated config. See the
|
||||
* module doc for the containment model and the class JSDoc on the seam for
|
||||
* the contract this implements (error-as-field, hostile-peer port,
|
||||
* no cross-run state, dispose to quiescence).
|
||||
*/
|
||||
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),
|
||||
maxOldGenerationSizeMb: z.number().default(512),
|
||||
})
|
||||
|
||||
readonly language = 'typescript'
|
||||
readonly isolation = 'worker-thread'
|
||||
|
||||
private readonly config: ResolvedConfig
|
||||
private readonly live = new Set<LiveRun>()
|
||||
private disposed = false
|
||||
|
||||
constructor(ctx: Context, config: Config) {
|
||||
super(ctx)
|
||||
// Schemastery filled the defaults; the cast records that. Positivity is a
|
||||
// semantic check the schema's plain number type does not carry.
|
||||
this.config = config as ResolvedConfig
|
||||
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)}`)
|
||||
}
|
||||
ctx.effect(() => () => this.teardown(), 'worker code-runtime teardown')
|
||||
}
|
||||
|
||||
/**
|
||||
* Dispose to quiescence: mark the service unusable, fail every in-flight
|
||||
* run as aborted, and AWAIT each worker's exit so no worker outlives the
|
||||
* fiber.
|
||||
*/
|
||||
private async teardown(): Promise<void> {
|
||||
this.disposed = true
|
||||
const runs = [...this.live]
|
||||
for (const run of runs) run.settle({ kind: 'abort', message: 'runtime disposed' })
|
||||
await Promise.all(runs.map(run => run.finished))
|
||||
}
|
||||
|
||||
/**
|
||||
* Execute one program in a fresh worker. Program outcomes — including a
|
||||
* type-strip syntax error, which never spawns a worker — resolve with
|
||||
* `result.error`; the method rejects only for seam misuse (a disposed
|
||||
* runtime, an invalid binding namespace).
|
||||
* @param request - the program, its bindings, and the abort signal.
|
||||
* @returns the run's outcome per the seam contract.
|
||||
*/
|
||||
async run(request: CodeRunRequest): Promise<CodeRunResult> {
|
||||
if (this.disposed) throw new Error('dsh-code-runtime-worker: run() after disposal')
|
||||
const bindings = this.validateBindings(request)
|
||||
if (request.signal?.aborted) {
|
||||
return { logs: [], error: { kind: 'abort', message: String(request.signal.reason) } }
|
||||
}
|
||||
|
||||
let code: string
|
||||
try {
|
||||
const stripped = stripTypeScriptTypes(STRIP_WRAP.prefix + request.program + STRIP_WRAP.suffix)
|
||||
code = stripped.slice(STRIP_WRAP.prefix.length, stripped.length - STRIP_WRAP.suffix.length)
|
||||
} catch (error: unknown) {
|
||||
// A program that does not survive the type-strip (syntax error,
|
||||
// non-erasable syntax like `enum`) is a program failure, reported the
|
||||
// same way a thrown exception would be — and no worker ever spawns.
|
||||
return { logs: [], error: { kind: 'exception', message: messageOf(error) } }
|
||||
}
|
||||
|
||||
return await this.execute(request, code, bindings)
|
||||
}
|
||||
|
||||
/** Reject (seam misuse) malformed binding namespaces: non-identifier or reserved globals, duplicates, and the `console` collision. */
|
||||
private validateBindings(request: CodeRunRequest): Map<string, Record<string, CodeBindingFunction>> {
|
||||
const bindings = new Map<string, Record<string, CodeBindingFunction>>()
|
||||
for (const namespace of request.bindings) {
|
||||
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)) {
|
||||
throw new Error(`dsh-code-runtime-worker: duplicate binding global ${JSON.stringify(namespace.global)}`)
|
||||
}
|
||||
bindings.set(namespace.global, namespace.functions)
|
||||
}
|
||||
return bindings
|
||||
}
|
||||
|
||||
/** Spawn the worker for one validated, type-stripped run and drive it to settlement. */
|
||||
private execute(
|
||||
request: CodeRunRequest,
|
||||
code: string,
|
||||
bindings: Map<string, Record<string, CodeBindingFunction>>,
|
||||
): Promise<CodeRunResult> {
|
||||
const bootData: WorkerBootData = {
|
||||
code,
|
||||
namespaces: [...bindings].map(([global, functions]) => ({ global, names: Object.keys(functions) })),
|
||||
maxLogBytes: this.config.maxLogBytes,
|
||||
maxValueBytes: this.config.maxValueBytes,
|
||||
}
|
||||
const worker = new Worker(WORKER_URL, {
|
||||
workerData: bootData,
|
||||
// Model code gets NO ambient environment — stronger than the scrubbed
|
||||
// env the defensive-patterns rule requires for spawned commands.
|
||||
env: {},
|
||||
// Hermetic flags too: without this the worker inherits the host
|
||||
// process's execArgv (a test runner's or tsx's loader hooks), which a
|
||||
// bare isolate with an empty environment cannot satisfy. The entry
|
||||
// needs nothing beyond native type stripping, on this repo's whole
|
||||
// Node range.
|
||||
execArgv: [],
|
||||
resourceLimits: { maxOldGenerationSizeMb: this.config.maxOldGenerationSizeMb },
|
||||
// Backstop capture: the bootstrap patches JS-level writes into its own
|
||||
// ordered buffer, so these pipes normally stay silent; anything that
|
||||
// still arrives (native-level writes) is appended after the done logs.
|
||||
stdout: true,
|
||||
stderr: true,
|
||||
})
|
||||
|
||||
return new Promise<CodeRunResult>((resolve) => {
|
||||
let settled = false
|
||||
const answered = new Set<number>()
|
||||
const logs: CodeLogEntry[] = []
|
||||
const strayLogs: CodeLogEntry[] = []
|
||||
|
||||
// ONE host-side ledger for everything that lands in `logs`/`strayLogs`,
|
||||
// whatever the path: honest port entries, FORGED port entries (model
|
||||
// code posting `log` messages directly, bypassing the worker-side
|
||||
// LogBuffer), and stray pipe bytes. On the first overflow it emits the
|
||||
// same in-band marker the worker's LogBuffer would and drops the rest,
|
||||
// so the documented cap is one shared `maxLogBytes` however it is hit.
|
||||
let logBudget = this.config.maxLogBytes
|
||||
let logsTruncated = false
|
||||
const admit = (entry: CodeLogEntry, sink: CodeLogEntry[]): void => {
|
||||
if (logsTruncated) return
|
||||
const cost = Buffer.byteLength(entry.text, 'utf8')
|
||||
if (cost > logBudget) {
|
||||
logsTruncated = true
|
||||
sink.push({ source: 'stderr', text: logTruncationMarker(this.config.maxLogBytes) })
|
||||
return
|
||||
}
|
||||
logBudget -= cost
|
||||
sink.push(entry)
|
||||
}
|
||||
|
||||
// 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 = (source: 'stdout' | 'stderr') => (chunk: Buffer) => {
|
||||
admit({ source, text: chunk.toString('utf8') }, strayLogs)
|
||||
}
|
||||
worker.stdout.on('data', captureStray('stdout'))
|
||||
worker.stderr.on('data', captureStray('stderr'))
|
||||
|
||||
// Settlement: exactly one outcome wins; every path funnels through
|
||||
// here, cleans up the timers/listeners, terminates the worker, and
|
||||
// resolves only after the worker actually exited (quiescence). Logs
|
||||
// streamed eagerly before the settlement are kept — a timed-out or
|
||||
// killed program still shows the model what it printed.
|
||||
let finishResolve!: () => void
|
||||
const finished = new Promise<void>((done) => { finishResolve = done })
|
||||
const finish = (result: Omit<CodeRunResult, 'logs'>): void => {
|
||||
if (settled) return
|
||||
settled = true
|
||||
clearInterval(eluTimer)
|
||||
clearTimeout(wallTimer)
|
||||
request.signal?.removeEventListener('abort', onAbort)
|
||||
this.live.delete(live)
|
||||
void worker.terminate().then(() => {
|
||||
finishResolve()
|
||||
resolve({ ...result, logs: [...logs, ...strayLogs] })
|
||||
})
|
||||
}
|
||||
|
||||
const onDone = (message: WorkerToHost): void => {
|
||||
if (message.type !== 'done') return
|
||||
// Re-cap the completion value HOST-side: the honest path already
|
||||
// capped it in the worker (prepareValue there), but a forged done
|
||||
// message bypasses the bootstrap entirely — without this, model code
|
||||
// could flood the host past maxValueBytes. Honest values pass
|
||||
// unchanged (see VALUE_RENDER_SLACK); the 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 onCall = (message: WorkerToHost): void => {
|
||||
if (message.type !== 'call' || settled) return
|
||||
// Hostile-peer rules: a duplicate id is ignored, an unknown name is
|
||||
// answered with a failure, and a binding throw/reject becomes the
|
||||
// program-side rejection — contained here, never a host crash.
|
||||
if (answered.has(message.id)) return
|
||||
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' })
|
||||
}
|
||||
}
|
||||
const record = bindings.get(message.global)
|
||||
// Own-property lookup only: a forged name like 'constructor' or
|
||||
// 'hasOwnProperty' must not walk the record's prototype chain and
|
||||
// reach a callable the consumer never declared.
|
||||
const fn = record && Object.hasOwn(record, message.name) ? record[message.name] : undefined
|
||||
if (typeof fn !== 'function') {
|
||||
reply({ type: 'reply', id: message.id, ok: false, message: `unknown binding ${JSON.stringify(`${message.global}.${message.name}`)}` })
|
||||
return
|
||||
}
|
||||
void (async () => {
|
||||
try {
|
||||
reply({ type: 'reply', id: message.id, ok: true, value: await fn(message.args) })
|
||||
} catch (error: unknown) {
|
||||
reply({ type: 'reply', id: message.id, ok: false, message: messageOf(error) })
|
||||
}
|
||||
})()
|
||||
}
|
||||
|
||||
worker.on('message', (raw: unknown) => {
|
||||
// Parse before touching: the peer can post ANY shape, and a throw in
|
||||
// this listener would crash the host process. Junk drops silently.
|
||||
const message = parseWorkerMessage(raw)
|
||||
if (!message) return
|
||||
if (message.type === 'log' && !settled) admit(message.entry, logs)
|
||||
onCall(message)
|
||||
onDone(message)
|
||||
})
|
||||
worker.on('error', (error: Error) => {
|
||||
finish({ error: { 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` } })
|
||||
})
|
||||
|
||||
// The compute budget reads the worker's own measured busy time, so a
|
||||
// hot loop expires it no matter what dispatches are in flight, while a
|
||||
// program idling on a slow binding accrues nothing.
|
||||
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)` } })
|
||||
}
|
||||
}, ELU_POLL_INTERVAL_MS)
|
||||
const wallTimer = setTimeout(() => {
|
||||
finish({ error: { 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) } })
|
||||
}
|
||||
request.signal?.addEventListener('abort', onAbort, { once: true })
|
||||
|
||||
const live: LiveRun = {
|
||||
worker,
|
||||
finished,
|
||||
settle: (failure: CodeRunFailure) => { finish({ error: failure }) },
|
||||
}
|
||||
this.live.add(live)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
export default WorkerCodeRuntime
|
||||
@@ -0,0 +1,78 @@
|
||||
/**
|
||||
* Wire protocol between the host runtime and the worker bootstrap. Everything
|
||||
* crossing the message port is structured-clone-plain and versionless — both
|
||||
* ends ship in this package, always at the same version. The host treats
|
||||
* inbound traffic as HOSTILE (the worker runs model code, which can reach
|
||||
* `parentPort` via `import('node:worker_threads')` and forge any of these
|
||||
* shapes); the worker treats inbound traffic as trusted.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-code-runtime-worker/src/protocol
|
||||
*/
|
||||
|
||||
import type { CodeLogEntry } from '@deepseek-ai/dsh-code-runtime'
|
||||
|
||||
/** What the host hands the worker at spawn, via `workerData`. */
|
||||
export interface WorkerBootData {
|
||||
/** The type-stripped (plain JS) program body. */
|
||||
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
|
||||
}
|
||||
|
||||
/** Worker → host: one bridged binding call. */
|
||||
export interface CallMessage {
|
||||
type: 'call'
|
||||
/** Worker-issued correlation id; the host answers each id at most once and ignores duplicates. */
|
||||
id: number
|
||||
/** The namespace global the call targets. */
|
||||
global: string
|
||||
/** The function name within the namespace. */
|
||||
name: string
|
||||
/** The single argument, structured-clone-plain. */
|
||||
args: unknown
|
||||
}
|
||||
|
||||
/** Worker → host: one captured log entry, streamed eagerly so output survives a mid-run termination (timeout, abort, OOM). */
|
||||
export interface LogMessage {
|
||||
type: 'log'
|
||||
entry: CodeLogEntry
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
*/
|
||||
export interface DoneMessage {
|
||||
type: 'done'
|
||||
value?: unknown
|
||||
error?: { message: string }
|
||||
}
|
||||
|
||||
/** Every message the worker sends. */
|
||||
export type WorkerToHost = CallMessage | LogMessage | 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`
|
||||
}
|
||||
@@ -0,0 +1,20 @@
|
||||
/**
|
||||
* The worker-thread entrypoint: self-executing glue over
|
||||
* `bootstrap.ts`'s {@link runWorkerMain}, kept to the spawn wiring alone.
|
||||
* Like `bin.ts` CLI entrypoints, this file executes only inside a spawned
|
||||
* worker isolate — a place the coverage provider cannot observe — so it is
|
||||
* excluded from the coverage gate while every line of actual logic lives in
|
||||
* `bootstrap.ts`, unit-tested in-process; the real spawn path is pinned by
|
||||
* the integration tests that run genuine workers.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-code-runtime-worker/src/worker
|
||||
*/
|
||||
|
||||
import { parentPort, workerData } from 'node:worker_threads'
|
||||
import { runWorkerMain } from './bootstrap.ts'
|
||||
import type { WorkerBootData } from './protocol.ts'
|
||||
|
||||
// A worker always has a parent port; guard loudly rather than run detached.
|
||||
if (!parentPort) throw new Error('dsh-code-runtime-worker: worker entry loaded outside a worker thread')
|
||||
|
||||
await runWorkerMain(parentPort, workerData as WorkerBootData, { stdout: process.stdout, stderr: process.stderr })
|
||||
@@ -0,0 +1,273 @@
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { EventEmitter } from 'node:events'
|
||||
import { LogBuffer, makeConsoleShim, makeNamespaces, captureStreamWrites, prepareValue, runWorkerMain, truncateUtf8Bytes, wireReplies } from '@deepseek-ai/dsh-code-runtime-worker/src/bootstrap.ts'
|
||||
import type { BootstrapPort, PatchableStream, PendingCall } from '@deepseek-ai/dsh-code-runtime-worker/src/bootstrap.ts'
|
||||
import type { ReplyMessage, WorkerToHost } from '@deepseek-ai/dsh-code-runtime-worker/src/protocol.ts'
|
||||
import type { CodeLogEntry } from '@deepseek-ai/dsh-code-runtime'
|
||||
|
||||
/**
|
||||
* An in-process stand-in for the worker's parentPort: the test plays the
|
||||
* HOST side — inspect what the bootstrap posted, feed replies back — so
|
||||
* every line of worker-side logic runs under coverage without spawning an
|
||||
* isolate (real-worker behavior is pinned by runtime.spec.ts).
|
||||
*/
|
||||
class FakePort implements BootstrapPort {
|
||||
sent: WorkerToHost[] = []
|
||||
private readonly emitter = new EventEmitter()
|
||||
/** Host-scripted responder; return undefined to leave the call pending. */
|
||||
respond: (message: WorkerToHost) => ReplyMessage | undefined = () => undefined
|
||||
|
||||
postMessage(message: WorkerToHost): void {
|
||||
this.sent.push(message)
|
||||
const reply = this.respond(message)
|
||||
if (reply) queueMicrotask(() => this.emitter.emit('message', reply))
|
||||
}
|
||||
|
||||
on(event: 'message', listener: (message: ReplyMessage) => void): void {
|
||||
this.emitter.on(event, listener)
|
||||
}
|
||||
|
||||
deliver(message: ReplyMessage): void {
|
||||
this.emitter.emit('message', message)
|
||||
}
|
||||
|
||||
logs(): CodeLogEntry[] {
|
||||
return this.sent.filter(message => message.type === 'log').map(message => message.entry)
|
||||
}
|
||||
|
||||
done(): WorkerToHost | undefined {
|
||||
return this.sent.find(message => message.type === 'done')
|
||||
}
|
||||
}
|
||||
|
||||
function fakeStreams(): { stdout: PatchableStream; stderr: PatchableStream } {
|
||||
return { stdout: { write: () => true }, stderr: { write: () => true } }
|
||||
}
|
||||
|
||||
const BOOT = { maxLogBytes: 65_536, maxValueBytes: 32_768 }
|
||||
|
||||
describe('LogBuffer', () => {
|
||||
it('streams entries to the sink until the byte budget, then emits one marker and drops the rest', () => {
|
||||
const seen: CodeLogEntry[] = []
|
||||
const buffer = new LogBuffer(10, entry => seen.push(entry))
|
||||
buffer.push({ source: 'console', level: 'log', text: '12345' })
|
||||
buffer.push({ source: 'console', level: 'log', text: '123456' })
|
||||
buffer.push({ source: 'console', level: 'log', text: 'dropped' })
|
||||
expect(seen.map(entry => entry.text)).toEqual([
|
||||
'12345',
|
||||
'[dsh-code-runtime-worker] log capture truncated at 10 bytes',
|
||||
])
|
||||
})
|
||||
})
|
||||
|
||||
describe('makeConsoleShim', () => {
|
||||
it('captures the five levels and renders non-strings inspect-style', () => {
|
||||
const seen: CodeLogEntry[] = []
|
||||
const shim = makeConsoleShim(new LogBuffer(1_000, entry => seen.push(entry)))
|
||||
shim.log('plain', { a: 1 })
|
||||
shim.info('i')
|
||||
shim.warn('w')
|
||||
shim.error('e')
|
||||
shim.debug('d')
|
||||
expect(seen.map(entry => entry.level)).toEqual(['log', 'info', 'warn', 'error', 'debug'])
|
||||
expect(seen[0]?.text).toBe('plain { a: 1 }')
|
||||
expect(seen.every(entry => entry.source === 'console')).toBe(true)
|
||||
})
|
||||
})
|
||||
|
||||
describe('captureStreamWrites', () => {
|
||||
it('redirects writes into the buffer and restores on request', () => {
|
||||
const seen: CodeLogEntry[] = []
|
||||
const buffer = new LogBuffer(1_000, entry => seen.push(entry))
|
||||
let underlying = ''
|
||||
const stream: PatchableStream = { write: (chunk: unknown) => { underlying += String(chunk); return true } }
|
||||
const restore = captureStreamWrites(buffer, stream, 'stdout')
|
||||
stream.write('captured', 'utf8')
|
||||
stream.write(Buffer.from('bytes'))
|
||||
restore()
|
||||
stream.write('after')
|
||||
expect(seen.map(entry => entry.text)).toEqual(['captured', 'bytes'])
|
||||
expect(seen[0]).toMatchObject({ source: 'stdout' })
|
||||
expect(underlying).toBe('after')
|
||||
})
|
||||
|
||||
it('invokes the write callback asynchronously, in both optional-encoding shapes', async () => {
|
||||
const buffer = new LogBuffer(1_000, () => {})
|
||||
const stream: PatchableStream = { write: () => true }
|
||||
captureStreamWrites(buffer, stream, 'stdout')
|
||||
const calls: (Error | null | undefined)[] = []
|
||||
stream.write('two-arg', (error?: Error | null) => calls.push(error))
|
||||
stream.write('three-arg', 'utf8', (error?: Error | null) => calls.push(error))
|
||||
// Node's contract: the callback fires after the write call returns.
|
||||
expect(calls).toEqual([])
|
||||
await new Promise<void>(resolve => stream.write('awaited flush', resolve))
|
||||
expect(calls).toEqual([null, null])
|
||||
})
|
||||
|
||||
it('still fires the callback for a write the exhausted budget drops', async () => {
|
||||
const buffer = new LogBuffer(4, () => {})
|
||||
const stream: PatchableStream = { write: () => true }
|
||||
captureStreamWrites(buffer, stream, 'stdout')
|
||||
stream.write('this write overflows the budget and is dropped')
|
||||
await new Promise<void>(resolve => stream.write('also dropped', resolve))
|
||||
})
|
||||
})
|
||||
|
||||
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'] } })
|
||||
})
|
||||
|
||||
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('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('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('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]" })
|
||||
})
|
||||
})
|
||||
|
||||
describe('truncateUtf8Bytes', () => {
|
||||
it('returns a fitting string whole', () => {
|
||||
expect(truncateUtf8Bytes('fits', 4)).toBe('fits')
|
||||
})
|
||||
|
||||
it('cuts at a code-point boundary, never mid-surrogate-pair', () => {
|
||||
// Each 😀 is one code point, two code units, four UTF-8 bytes: a 5-byte
|
||||
// budget fits exactly one — and never leaves a lone surrogate behind.
|
||||
const cut = truncateUtf8Bytes('😀😀', 5)
|
||||
expect(cut).toBe('😀')
|
||||
expect(Buffer.byteLength(truncateUtf8Bytes('😀😀', 3), 'utf8')).toBe(0)
|
||||
})
|
||||
})
|
||||
|
||||
describe('makeNamespaces', () => {
|
||||
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
|
||||
const pending = new Map<number, PendingCall>()
|
||||
wireReplies(port, pending)
|
||||
const [tools] = makeNamespaces({ namespaces: [{ global: 'tools', names: ['__proto__', 'constructor', 'toString'] }] }, port, pending, { value: 1 }) as [Record<string, (args: unknown) => Promise<unknown>>]
|
||||
expect(Object.getPrototypeOf(tools)).toBeNull()
|
||||
await expect(tools['__proto__']?.({})).resolves.toBe('__proto__-ok')
|
||||
await expect(tools['constructor']?.({})).resolves.toBe('constructor-ok')
|
||||
await expect(tools['toString']?.({})).resolves.toBe('toString-ok')
|
||||
})
|
||||
|
||||
it('rejects a non-cloneable argument without leaking the pending entry', async () => {
|
||||
let firstCall = true
|
||||
const throwingPort: BootstrapPort = {
|
||||
// First call throws an Error (the real DataCloneError shape), the
|
||||
// second a bare string — the rejection renders both.
|
||||
postMessage: () => {
|
||||
if (firstCall) { firstCall = false; throw new Error('DataCloneError-ish') }
|
||||
throw 'raw-clone-failure'
|
||||
},
|
||||
on: () => {},
|
||||
}
|
||||
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/)
|
||||
expect(pending.size).toBe(0)
|
||||
})
|
||||
})
|
||||
|
||||
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
|
||||
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([{ source: 'console', level: 'log', text: 'got 42' }])
|
||||
expect(port.done()).toEqual({ type: 'done', value: { doubled: 42 } })
|
||||
})
|
||||
|
||||
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?.message : undefined).toContain('boom')
|
||||
expect(done?.type === 'done' ? done.value : undefined).toBeUndefined()
|
||||
})
|
||||
|
||||
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' } })
|
||||
|
||||
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' } })
|
||||
})
|
||||
|
||||
it('surfaces a host failure reply as a program-side rejection it can catch', async () => {
|
||||
const port = new FakePort()
|
||||
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}` }',
|
||||
namespaces: [{ global: 'tools', names: ['x'] }],
|
||||
}, fakeStreams())
|
||||
expect(port.done()).toEqual({ type: 'done', value: 'caught: denied by host' })
|
||||
})
|
||||
|
||||
it('ignores replies for unknown pending ids', async () => {
|
||||
const port = new FakePort()
|
||||
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' }
|
||||
}
|
||||
await runWorkerMain(port, {
|
||||
...BOOT,
|
||||
code: 'return await tools.x({})',
|
||||
namespaces: [{ global: 'tools', names: ['x'] }],
|
||||
}, fakeStreams())
|
||||
expect(port.done()).toEqual({ type: 'done', value: 'real' })
|
||||
})
|
||||
|
||||
it('captures raw stream writes through the patched process streams', async () => {
|
||||
const port = new FakePort()
|
||||
const streams = fakeStreams()
|
||||
await runWorkerMain(port, { ...BOOT, code: 'return 1', namespaces: [] }, streams)
|
||||
streams.stdout.write('never seen — already restored? no: patch persists in worker')
|
||||
// The patch stays installed for the worker's lifetime; writes during the
|
||||
// program landed in order. Here the program wrote nothing via streams, so
|
||||
// only the post-run write above went through the patched slot.
|
||||
expect(port.logs().at(-1)).toMatchObject({ source: 'stdout' })
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,55 @@
|
||||
import { spawn } from 'node:child_process'
|
||||
import { existsSync } from 'node:fs'
|
||||
import { join } from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
|
||||
/**
|
||||
* BUILT-ARTIFACT smoke for the published package (the real-load-path guard
|
||||
* from docs/testing.md): the unit suite runs `src/` under vitest, where the
|
||||
* worker entry resolves to `src/worker.ts` — a consumer runs `lib/index.js`
|
||||
* under plain `node`, where it must resolve the sibling `lib/worker.js`
|
||||
* bundle instead. This spawns plain `node` (NOT tsx) from inside the package
|
||||
* directory and imports the package BY NAME, so resolution flows through the
|
||||
* real `exports` map exactly as it would from a downstream install; the
|
||||
* program exercises the type-strip, the worker spawn, the binding bridge,
|
||||
* and log capture end-to-end through the built bundles.
|
||||
*
|
||||
* It build-gates: SKIPS when the built artifacts are absent (suite run
|
||||
* without `pnpm run build`); CI runs it after the build step. KEYLESS — no
|
||||
* model is involved.
|
||||
*/
|
||||
|
||||
const pkgDir = fileURLToPath(new URL('..', import.meta.url))
|
||||
const built = ['lib/index.js', 'lib/worker.js'].every(file => existsSync(join(pkgDir, file)))
|
||||
&& existsSync(join(pkgDir, '../code-runtime/lib/index.js'))
|
||||
|
||||
describe.skipIf(!built)('built lib real load path (plain node)', () => {
|
||||
it('runs a TypeScript program with a binding through lib/index.js and its lib/worker.js entry', async () => {
|
||||
const script = `
|
||||
const { Context } = await import('cordis')
|
||||
const { WorkerCodeRuntime } = await import('@deepseek-ai/dsh-code-runtime-worker')
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(WorkerCodeRuntime, {})
|
||||
const result = await ctx.codeRuntime.run({
|
||||
program: 'const doubled: number = await tools.double({ n: 21 }); console.log("halfway", doubled); return doubled;',
|
||||
bindings: [{ global: 'tools', functions: { double: async args => args.n * 2 } }],
|
||||
})
|
||||
console.log(JSON.stringify(result))
|
||||
process.exit(0)
|
||||
`
|
||||
const child = spawn(process.execPath, ['--input-type=module', '-e', script], { cwd: pkgDir, stdio: ['ignore', 'pipe', 'pipe'] })
|
||||
let stdout = ''
|
||||
let stderr = ''
|
||||
child.stdout.on('data', (chunk: Buffer) => { stdout += chunk.toString('utf8') })
|
||||
child.stderr.on('data', (chunk: Buffer) => { stderr += chunk.toString('utf8') })
|
||||
const exitCode = await new Promise<number | null>(resolve => child.on('close', resolve))
|
||||
|
||||
expect(exitCode, `stderr:\n${stderr}`).toBe(0)
|
||||
const lastLine = stdout.trim().split('\n').at(-1) ?? ''
|
||||
const result = JSON.parse(lastLine) as { value?: unknown; logs: { source: string; level?: string; text: string }[]; error?: unknown }
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.value).toBe(42)
|
||||
expect(result.logs).toContainEqual({ source: 'console', level: 'log', text: 'halfway 42' })
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,451 @@
|
||||
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'
|
||||
|
||||
/**
|
||||
* Integration suite over REAL worker threads (no mocks — workers are cheap
|
||||
* and local, per docs/testing.md's real-over-mock policy). Each test builds
|
||||
* a fresh context so budgets can be tuned per case.
|
||||
*/
|
||||
async function setup(config: Config = {}) {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(WorkerCodeRuntime, config)
|
||||
const runtime = ctx.codeRuntime as WorkerCodeRuntime
|
||||
return { ctx, runtime }
|
||||
}
|
||||
|
||||
/** Convenience: one namespace `tools` with the given functions. */
|
||||
function tools(functions: Record<string, (args: unknown) => Promise<unknown>>) {
|
||||
return [{ global: 'tools', functions }]
|
||||
}
|
||||
|
||||
describe('WorkerCodeRuntime — programs and bindings (real workers)', () => {
|
||||
it('registers with the seam descriptors', async () => {
|
||||
const { runtime } = await setup()
|
||||
expect(runtime.language).toBe('typescript')
|
||||
expect(runtime.isolation).toBe('worker-thread')
|
||||
})
|
||||
|
||||
it('runs TypeScript (erasable syntax), captures console/stdout in order, returns the value', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: `
|
||||
interface Point { x: number; y: number }
|
||||
const p: Point = { x: 1, y: 2 } as Point;
|
||||
console.log('point', p);
|
||||
process.stdout.write('raw-out\\n');
|
||||
console.warn('careful');
|
||||
return p.x + p.y;
|
||||
`,
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.value).toBe(3)
|
||||
expect(result.logs.map(entry => [entry.source, entry.level ?? null])).toEqual([
|
||||
['console', 'log'],
|
||||
['stdout', null],
|
||||
['console', 'warn'],
|
||||
])
|
||||
expect(result.logs[0]?.text).toBe('point { x: 1, y: 2 }')
|
||||
})
|
||||
|
||||
it('bridges binding calls both ways and rejects the program-side call on a host rejection', async () => {
|
||||
const { runtime } = await setup()
|
||||
const calls: unknown[] = []
|
||||
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 }
|
||||
return { first, caught, caughtRaw };
|
||||
`,
|
||||
bindings: tools({
|
||||
echo: async (args) => { calls.push(args); return { echoed: args } },
|
||||
fail: async () => { throw new Error('nope') },
|
||||
// A non-Error throw: the host renders it, the program still catches.
|
||||
failRaw: async () => { throw 'raw-nope' },
|
||||
}),
|
||||
})
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.value).toEqual({ first: { echoed: { n: 1 } }, caught: 'nope', caughtRaw: 'raw-nope' })
|
||||
expect(calls).toEqual([{ n: 1 }])
|
||||
})
|
||||
|
||||
it('reports non-erasable syntax as an exception without spawning a worker', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({ program: 'enum E { A }\nreturn 1', bindings: [] })
|
||||
expect(result.error?.kind).toBe('exception')
|
||||
expect(result.error?.message).toMatch(/enum|strip/i)
|
||||
})
|
||||
|
||||
it('reports a runtime throw as an exception with the message', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({ program: 'throw new Error("kaboom")', bindings: [] })
|
||||
expect(result.error?.kind).toBe('exception')
|
||||
expect(result.error?.message).toContain('kaboom')
|
||||
})
|
||||
|
||||
it('gives the program an EMPTY environment', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({ program: 'return JSON.stringify(process.env)', bindings: [] })
|
||||
expect(result.value).toBe('{}')
|
||||
})
|
||||
|
||||
it('replaces a non-cloneable return value with a string rendering', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({ program: 'return { f: () => 1 }', bindings: [] })
|
||||
expect(typeof result.value).toBe('string')
|
||||
})
|
||||
|
||||
it('completes a program that returns nothing with no value at all', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({ program: 'const x = 1', bindings: [] })
|
||||
expect(result.error).toBeUndefined()
|
||||
expect('value' in result).toBe(false)
|
||||
})
|
||||
|
||||
it('keeps logs streamed before a failure', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: 'console.log("before"); throw new Error("after-log")',
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.error?.kind).toBe('exception')
|
||||
expect(result.logs.map(entry => entry.text)).toContain('before')
|
||||
})
|
||||
})
|
||||
|
||||
describe('WorkerCodeRuntime — budgets and containment (real workers)', () => {
|
||||
it('ends a hot loop at the compute budget — including behind a pending decoy dispatch', async () => {
|
||||
const { runtime } = await setup({ computeMs: 300, maxWallMs: 30_000 })
|
||||
const result = await runtime.run({
|
||||
// The decoy: fire a call at a never-resolving binding WITHOUT awaiting,
|
||||
// then spin. Host-side pending-call bookkeeping would pause a naive
|
||||
// budget here; measured busy time cannot be fooled.
|
||||
program: 'void tools.slow({}); for (;;) {}',
|
||||
bindings: tools({ slow: () => new Promise(() => {}) }),
|
||||
})
|
||||
expect(result.error?.kind).toBe('timeout')
|
||||
expect(result.error?.message).toContain('compute budget')
|
||||
}, 15_000)
|
||||
|
||||
it('does not charge time spent awaiting a slow binding against the compute budget', async () => {
|
||||
const { runtime } = await setup({ computeMs: 250, maxWallMs: 30_000 })
|
||||
const result = await runtime.run({
|
||||
program: 'return await tools.slow({})',
|
||||
bindings: tools({ slow: () => new Promise(resolve => setTimeout(() => { resolve('slow-done') }, 700)) }),
|
||||
})
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.value).toBe('slow-done')
|
||||
}, 15_000)
|
||||
|
||||
it('ends an idle-forever run at the wall-clock ceiling', async () => {
|
||||
const { runtime } = await setup({ computeMs: 30_000, maxWallMs: 400 })
|
||||
const result = await runtime.run({
|
||||
program: 'await tools.never({}); return 1',
|
||||
bindings: tools({ never: () => new Promise(() => {}) }),
|
||||
})
|
||||
expect(result.error?.kind).toBe('timeout')
|
||||
expect(result.error?.message).toContain('wall-clock ceiling')
|
||||
}, 15_000)
|
||||
|
||||
it('reports an abort mid-run and stops the worker', async () => {
|
||||
const { runtime } = await setup()
|
||||
const controller = new AbortController()
|
||||
setTimeout(() => { controller.abort('user-cancel') }, 150)
|
||||
const result = await runtime.run({ program: 'for (;;) {}', bindings: [], signal: controller.signal })
|
||||
expect(result.error).toEqual({ kind: 'abort', message: 'user-cancel' })
|
||||
}, 15_000)
|
||||
|
||||
it('reports a pre-aborted signal without spawning', async () => {
|
||||
const { runtime } = await setup()
|
||||
const controller = new AbortController()
|
||||
controller.abort('too-late')
|
||||
const result = await runtime.run({ program: 'return 1', bindings: [], signal: controller.signal })
|
||||
expect(result.error).toEqual({ kind: 'abort', message: 'too-late' })
|
||||
})
|
||||
|
||||
it('drops a binding resolution that lands after the run settled', async () => {
|
||||
const { runtime } = await setup()
|
||||
const controller = new AbortController()
|
||||
let replyDelivered!: Promise<void>
|
||||
const result = await runtime.run({
|
||||
program: 'void tools.late({}); for (;;) {}',
|
||||
bindings: tools({
|
||||
// Anchored on invocation: abort 100ms after the call reaches the
|
||||
// host, resolve 400ms after — by then the run has settled, so the
|
||||
// resolution's reply hits the post-settlement drop.
|
||||
late: () => new Promise((resolve) => {
|
||||
setTimeout(() => { controller.abort('cancel-now') }, 100)
|
||||
replyDelivered = new Promise(done => setTimeout(() => { resolve('too-late'); done() }, 400))
|
||||
}),
|
||||
}),
|
||||
signal: controller.signal,
|
||||
})
|
||||
expect(result.error).toEqual({ kind: 'abort', message: 'cancel-now' })
|
||||
// Let the late resolution actually fire so its reply executes instead of
|
||||
// being cancelled with the test.
|
||||
await replyDelivered
|
||||
}, 15_000)
|
||||
|
||||
it('contains an OOM under resourceLimits as worker-exit, host process healthy', async () => {
|
||||
const { runtime } = await setup({ maxOldGenerationSizeMb: 32 })
|
||||
const result = await runtime.run({
|
||||
program: 'const hog = []; for (;;) hog.push(new Array(1e6).fill(1));',
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.error?.kind).toBe('worker-exit')
|
||||
// And the host is fine: run something else.
|
||||
const after = await runtime.run({ program: 'return "alive"', bindings: [] })
|
||||
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 })
|
||||
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)?.text).toContain('truncated at 300 bytes')
|
||||
const total = result.logs.reduce((sum, entry) => sum + Buffer.byteLength(entry.text, 'utf8'), 0)
|
||||
expect(total).toBeLessThan(1_000)
|
||||
})
|
||||
|
||||
it('caps an oversized return value with a truncation marker', async () => {
|
||||
const { runtime } = await setup({ maxValueBytes: 64 })
|
||||
const result = await runtime.run({ program: 'return "y".repeat(10_000)', bindings: [] })
|
||||
expect(result.value).toBe(`${'y'.repeat(64)}… [truncated]`)
|
||||
})
|
||||
|
||||
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('completes a program that awaits its write callback, capturing the chunk', async () => {
|
||||
// Node's write(chunk[, encoding][, callback]) contract: dropping the
|
||||
// callback would leave this promise pending until the wall ceiling and
|
||||
// misreport a completed program as a timeout.
|
||||
const { runtime } = await setup({ maxWallMs: 2_000 })
|
||||
const result = await runtime.run({
|
||||
program: 'await new Promise(resolve => process.stdout.write("flushed", resolve)); return "done"',
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.value).toBe('done')
|
||||
expect(result.logs).toContainEqual({ source: 'stdout', text: 'flushed' })
|
||||
})
|
||||
|
||||
it('caps a huge container whose bounded rendering is small (wire size, not rendering, is what counts)', 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')
|
||||
})
|
||||
|
||||
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 })
|
||||
const result = await runtime.run({
|
||||
// The bootstrap patches the stream instance's own `write`; going
|
||||
// through the prototype's slot reaches the real pipe underneath, so
|
||||
// the bytes arrive host-side as stray data. The pauses keep the two
|
||||
// writes in separate pipe chunks and let them land before settlement.
|
||||
program: `
|
||||
const write = (text) => Object.getPrototypeOf(process.stdout).write.call(process.stdout, text);
|
||||
write('abcd');
|
||||
await new Promise(resolve => setTimeout(resolve, 150));
|
||||
write('ef');
|
||||
await new Promise(resolve => setTimeout(resolve, 100));
|
||||
return 1;
|
||||
`,
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.logs).toContainEqual({ source: 'stdout', text: 'abcd' })
|
||||
expect(result.logs.map(entry => entry.text)).not.toContain('ef')
|
||||
}, 15_000)
|
||||
})
|
||||
|
||||
describe('WorkerCodeRuntime — hostile programs (real workers)', () => {
|
||||
it('survives forged port traffic: unknown binding names, duplicate ids, junk shapes', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: `
|
||||
const { parentPort } = await import('node:worker_threads');
|
||||
parentPort.postMessage({ type: 'call', id: 7777, global: 'tools', name: 'missing', args: {} });
|
||||
parentPort.postMessage({ type: 'call', id: 7777, global: 'tools', name: 'missing', args: {} });
|
||||
parentPort.postMessage({ type: 'call', id: 7778, global: 'tools', name: 'constructor', args: {} });
|
||||
parentPort.postMessage({ type: 'junk' });
|
||||
return await tools.real({});
|
||||
`,
|
||||
bindings: tools({ real: async () => 'still-works' }),
|
||||
})
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.value).toBe('still-works')
|
||||
})
|
||||
|
||||
it('survives arbitrary junk on the port: non-objects, junk types, malformed calls, logs, and dones', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: `
|
||||
const { parentPort } = await import('node:worker_threads');
|
||||
for (const junk of [
|
||||
null, 42, 'junk', [],
|
||||
{ type: 'nope' },
|
||||
{ type: 'call' },
|
||||
{ type: 'call', id: 'x', global: 'tools', name: 'real', args: {} },
|
||||
{ type: 'call', id: 1e9, global: 7, name: 'real', args: {} },
|
||||
{ type: 'call', id: 1e9, global: 'tools', name: 7, args: {} },
|
||||
{ type: 'log' },
|
||||
{ type: 'log', entry: null },
|
||||
{ type: 'log', entry: { source: 'stdout', text: 7 } },
|
||||
{ type: 'log', entry: { source: 'nope', text: 'x' } },
|
||||
{ type: 'log', entry: { source: 'console', level: 'nope', text: 'x' } },
|
||||
{ type: 'log', entry: { source: 'console', level: 7, text: 'x' } },
|
||||
{ type: 'done', error: 5 },
|
||||
{ type: 'done', error: { message: 5 } },
|
||||
]) parentPort.postMessage(junk);
|
||||
return await tools.real({});
|
||||
`,
|
||||
bindings: tools({ real: async () => 'still-works' }),
|
||||
})
|
||||
expect(result.error).toBeUndefined()
|
||||
expect(result.value).toBe('still-works')
|
||||
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 })
|
||||
const result = await runtime.run({
|
||||
// Forged messages bypass the worker-side LogBuffer and prepareValue
|
||||
// entirely — only the host-side ledger and re-cap stand between model
|
||||
// code and an unbounded result.
|
||||
program: `
|
||||
const { parentPort } = await import('node:worker_threads');
|
||||
for (let i = 0; i < 50; i++) parentPort.postMessage({ type: 'log', entry: { source: 'stdout', text: 'F'.repeat(100), forged: true } });
|
||||
parentPort.postMessage({ type: 'done', value: 'V'.repeat(100000) });
|
||||
for (;;) {}
|
||||
`,
|
||||
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, entry) => sum + Buffer.byteLength(entry.text, 'utf8'), 0)
|
||||
expect(total).toBeLessThanOrEqual(200 + Buffer.byteLength(marker, 'utf8'))
|
||||
expect(result.logs.at(-1)?.text).toBe(marker)
|
||||
expect(result.logs.every(entry => !('forged' in entry))).toBe(true)
|
||||
})
|
||||
|
||||
it('accepts a forged done carrying both value and error (self-sabotage, contained)', 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 (;;) {}
|
||||
`,
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.value).toBe('lied')
|
||||
expect(result.error).toEqual({ 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 })
|
||||
const result = await runtime.run({
|
||||
program: `
|
||||
const { parentPort } = await import('node:worker_threads');
|
||||
parentPort.postMessage({ type: 'done', error: { message: '€'.repeat(1000) } });
|
||||
for (;;) {}
|
||||
`,
|
||||
bindings: [],
|
||||
})
|
||||
expect(result.error).toEqual({ kind: 'exception', message: '€€' })
|
||||
})
|
||||
|
||||
it('answers a binding whose resolution cannot be cloned with a failure reply', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: 'try { await tools.bad({}) } catch (error) { return error.message }',
|
||||
bindings: tools({ bad: async () => (() => 1) }),
|
||||
})
|
||||
expect(result.value).toContain('not structured-cloneable')
|
||||
})
|
||||
|
||||
it('exposes binding names that collide with Object.prototype as ordinary functions', async () => {
|
||||
const { runtime } = await setup()
|
||||
const result = await runtime.run({
|
||||
program: 'return [await tools["__proto__"]({}), await tools["constructor"]({}), typeof tools["hasOwnProperty"]]',
|
||||
// Computed keys: a literal `'__proto__': …` entry would SET the record's
|
||||
// prototype instead of declaring a binding of that name.
|
||||
bindings: tools({ ['__proto__']: async () => 'proto-ok', ['constructor']: async () => 'ctor-ok' }),
|
||||
})
|
||||
expect(result.value).toEqual(['proto-ok', 'ctor-ok', 'undefined'])
|
||||
})
|
||||
})
|
||||
|
||||
describe('WorkerCodeRuntime — seam misuse and lifecycle', () => {
|
||||
it('rejects invalid binding globals loudly (identifier, reserved word, duplicate, console)', async () => {
|
||||
const { runtime } = await setup()
|
||||
const cases: [string, RegExp][] = [
|
||||
['not valid!', /not a usable identifier/],
|
||||
['await', /not a usable identifier/],
|
||||
['console', /duplicate binding global/],
|
||||
]
|
||||
for (const [global, message] of cases) {
|
||||
await expect(runtime.run({ program: 'return 1', bindings: [{ global, functions: {} }] })).rejects.toThrow(message)
|
||||
}
|
||||
await expect(runtime.run({
|
||||
program: 'return 1',
|
||||
bindings: [{ global: 'tools', functions: {} }, { global: 'tools', functions: {} }],
|
||||
})).rejects.toThrow(/duplicate binding global/)
|
||||
})
|
||||
|
||||
it('rejects config values that are not positive numbers', async () => {
|
||||
const ctx = new Context()
|
||||
await expect(ctx.plugin(WorkerCodeRuntime, { computeMs: -1 })).rejects.toThrow(/positive number/)
|
||||
})
|
||||
|
||||
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: [] })
|
||||
const second = await runtime.run({ program: 'return typeof globalThis.leak', bindings: [] })
|
||||
expect(second.value).toBe('undefined')
|
||||
})
|
||||
|
||||
it('disposal aborts in-flight runs, awaits worker exit, and rejects later runs', async () => {
|
||||
const ctx = new Context()
|
||||
const fiber = await ctx.plugin(WorkerCodeRuntime)
|
||||
const runtime = ctx.codeRuntime as WorkerCodeRuntime
|
||||
const inflight: Promise<CodeRunResult> = runtime.run({ program: 'for (;;) {}', bindings: [] })
|
||||
// Give the worker a moment to actually start spinning.
|
||||
await new Promise(resolve => setTimeout(resolve, 200))
|
||||
await fiber.dispose()
|
||||
const result = await inflight
|
||||
expect(result.error).toEqual({ kind: 'abort', message: 'runtime disposed' })
|
||||
await expect(runtime.run({ program: 'return 1', bindings: [] })).rejects.toThrow(/after disposal/)
|
||||
}, 15_000)
|
||||
|
||||
it('removes ctx.codeRuntime when the providing fiber disposes (HMR safety)', async () => {
|
||||
const ctx = new Context()
|
||||
const fiber = await ctx.plugin(WorkerCodeRuntime)
|
||||
expect(ctx.get('codeRuntime')).toBeInstanceOf(WorkerCodeRuntime)
|
||||
await fiber.dispose()
|
||||
expect(ctx.get('codeRuntime')).toBeUndefined()
|
||||
})
|
||||
})
|
||||
@@ -0,0 +1,24 @@
|
||||
{
|
||||
"extends": "../../../tsconfig.base.json",
|
||||
"compilerOptions": {
|
||||
"rootDir": "src",
|
||||
"outDir": "lib/types"
|
||||
},
|
||||
"include": [
|
||||
"src"
|
||||
],
|
||||
"references": [
|
||||
{
|
||||
"path": "../../../vendor/cosmokit"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/cordis"
|
||||
},
|
||||
{
|
||||
"path": "../../../vendor/schemastery"
|
||||
},
|
||||
{
|
||||
"path": "../code-runtime"
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
import { defineConfig } from 'tsdown'
|
||||
|
||||
/**
|
||||
* Package-shape override (see the root tsdown.config.ts): besides the
|
||||
* default lib/index.js bundle, the worker BOOTSTRAP ships as its own
|
||||
* sibling entry — `new Worker(new URL('./worker.js', import.meta.url))`
|
||||
* loads it as a file, so it cannot be part of the index bundle. TWO
|
||||
* single-entry builds, not one two-entry build: a multi-entry build emits
|
||||
* the shared bootstrap module as a `lib/bootstrap-*.js` chunk both bundles
|
||||
* import, which the package.json `files` whitelist (deliberately exact)
|
||||
* would omit from the packed artifact — each single-entry build inlines its
|
||||
* own bootstrap copy instead, keeping every shipped file self-contained.
|
||||
*/
|
||||
export default defineConfig([
|
||||
{
|
||||
entry: ['lib/types/index.js'],
|
||||
outDir: 'lib',
|
||||
format: ['esm'],
|
||||
platform: 'node',
|
||||
target: 'es2024',
|
||||
fixedExtension: false,
|
||||
dts: false,
|
||||
clean: false,
|
||||
},
|
||||
{
|
||||
entry: ['lib/types/worker.js'],
|
||||
outDir: 'lib',
|
||||
format: ['esm'],
|
||||
platform: 'node',
|
||||
target: 'es2024',
|
||||
fixedExtension: false,
|
||||
dts: false,
|
||||
clean: false,
|
||||
},
|
||||
])
|
||||
Reference in New Issue
Block a user