2026-07-08 11:00:06 +08:00
/**
2026-07-13 23:27:00 +08:00
* Worker-thread code runtime: a fresh worker runs each host-type-stripped TypeScript program
* and bridges bindings over its message port. This is containment, not a security boundary:
* model code has bash-equivalent trust despite an empty environment, a heap cap, measured
* event-loop busy-time and wall-time budgets, and termination that also stops synchronous loops.
2026-07-08 11:00:06 +08:00
* @module @deepseek-ai/dsh-code-runtime-worker
*/
import { Worker } from 'node:worker_threads'
import { stripTypeScriptTypes } from 'node:module'
2026-07-22 01:27:03 +08:00
import type { Readable } from 'node:stream'
2026-07-13 20:55:47 +08:00
import { fileURLToPath } from 'node:url'
2026-08-10 22:04:06 +08:00
import { Context } from '@deepseek-ai/cordis'
import z from '@deepseek-ai/schemastery'
2026-07-27 14:01:15 +08:00
import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
2026-07-31 18:10:33 +08:00
import { CodeRuntime , DUNDER_MEMBER , PORTABLE_RESERVED_WORDS , RESERVED_BINDING_GLOBALS , RESERVED_ERROR_MEMBERS } from '@deepseek-ai/dsh-code-runtime'
2026-07-23 01:32:17 +08:00
import type { CodeBindingNamespace , CodeJsonValue , CodeRunFailure , CodeRunRequest , CodeRunResult } from '@deepseek-ai/dsh-code-runtime'
2026-07-21 04:34:14 +08:00
import { snapshotJsonValue } from '@deepseek-ai/dsh-session'
2026-07-08 11:00:06 +08:00
import type { ReplyMessage , WorkerBootData , WorkerToHost } from './protocol.ts'
2026-07-21 21:56:38 +08:00
import { jsonStringBytesUpTo , jsonValueBytesUpTo , truncateJsonStringBytes } from './output-json.ts'
2026-07-22 19:09:27 +08:00
import { decodeWorkerJson , encodeWorkerJson } from './worker-json.ts'
import type { WorkerJsonWire } from './worker-json.ts'
2026-07-08 11:00:06 +08:00
/** 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
2026-07-27 19:53:57 +08:00
* nobody will resolve). At most `2_147_483_647` (Node's maximum
* `setTimeout` delay, about 24.9 days): a longer value is rejected at load
* because `setTimeout` would clamp it to 1 ms.
2026-07-08 11:00:06 +08:00
*/
maxWallMs? : number
2026-07-22 19:57:44 +08:00
/**
* Hard cap for serialized log-array, completion-value, and failure-message payloads;
* fixed result-envelope syntax is excluded.
*/
2026-07-21 04:34:14 +08:00
maxOutputBytes? : number
2026-07-08 11:00:06 +08:00
/** 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
2026-07-22 19:57:44 +08:00
/** Smallest cap that can represent the counted payloads: an empty logs array plus an empty JSON failure message. */
2026-07-21 04:34:14 +08:00
const MIN_OUTPUT_BYTES = 4
2026-07-31 18:10:33 +08:00
/**
* The seam's language-portable identifier subset (see
* `CodeBindingNamespace.global`): no `$`, which is JS-only spelling — the same
* namespace list must be usable against every backend regardless of language.
*/
const IDENTIFIER = /^[A-Za-z_][A-Za-z0-9_]*$/
2026-07-08 11:00:06 +08:00
/**
* 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 >
}
/**
2026-07-13 20:55:47 +08:00
* The worker entry path. Source runs unbuilt (`src/worker.ts`, loadable
2026-07-08 11:00:06 +08:00
* directly on this repo's Node range via native type stripping — the file
* is erasable-only with type-only relative imports); the built package
2026-07-13 20:55:47 +08:00
* ships it as a sibling CommonJS bundle (`lib/worker.cjs`, its own tsdown
* entry) because pkg's VFS Worker hook compiles string-path entries as
* CommonJS.
2026-07-08 11:00:06 +08:00
* The URL *pathname*'s extension says which world this module is in —
* pathname, because dev-time module runners (vitest) may suffix
2026-07-13 20:55:47 +08:00
* `import.meta.url` with a query string; relative resolution drops it. Worker
* receives a filesystem string so pkg's VFS Worker hook can resolve it.
2026-07-08 11:00:06 +08:00
*/
2026-07-13 20:55:47 +08:00
/* v8 ignore next -- the './worker.cjs' arm is the built-lib world, unreachable unbuilt by construction; the built-lib e2e pins it. */
const WORKER_PATH = fileURLToPath ( new URL ( new URL ( import . meta . url ) . pathname . endsWith ( '.ts' ) ? './worker.ts' : './worker.cjs' , import . meta . url ) )
2026-07-08 11:00:06 +08:00
/** Render an unknown thrown value as a message, `Error` or not. */
function messageOf ( error : unknown ) : string {
return error instanceof Error ? error.message : String ( error )
}
2026-07-22 01:27:03 +08:00
/** Resolve after a worker pipe emits all queued data, or closes/errors during termination. */
function waitForPipeDrain ( stream : Readable ) : Promise < void > {
if ( stream . readableEnded || stream . destroyed ) return Promise . resolve ( )
return new Promise ( ( resolve ) = > {
const done = ( ) : void = > {
stream . off ( 'end' , done )
stream . off ( 'close' , done )
stream . off ( 'error' , done )
resolve ( )
}
stream . once ( 'end' , done )
stream . once ( 'close' , done )
stream . once ( 'error' , done )
// Close the event-registration race if termination finished between the
// initial state check and the listeners above.
2026-07-22 01:42:41 +08:00
/* v8 ignore next -- this race cannot be scheduled deterministically between the adjacent state check and listener registration. */
2026-07-22 01:27:03 +08:00
if ( stream . readableEnded || stream . destroyed ) done ( )
} )
}
2026-07-08 11:39:57 +08:00
/**
* 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
2026-07-22 19:09:27 +08:00
return { type : 'call' , id : m.id , global : m . global , name : m.name , args : m.args as WorkerJsonWire }
2026-07-08 11:39:57 +08:00
}
case 'log' : {
2026-07-14 03:07:41 +08:00
if ( typeof m . text !== 'string' ) return undefined
return { type : 'log' , text : m.text }
2026-07-08 11:39:57 +08:00
}
2026-07-21 04:34:14 +08:00
case 'output-limit' : return { type : 'output-limit' }
2026-07-08 11:39:57 +08:00
case 'done' : {
2026-07-22 19:09:27 +08:00
if ( m . error === undefined ) return { type : 'done' , . . . m . value !== undefined ? { value : m.value as WorkerJsonWire } : { } }
2026-07-08 11:39:57 +08:00
const error = m . error
if ( typeof error !== 'object' || error === null ) return undefined
2026-07-21 04:34:14 +08:00
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 } }
2026-07-08 11:39:57 +08:00
}
default : return undefined
}
}
2026-07-21 04:34:14 +08:00
/** 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 {
2026-07-21 21:56:38 +08:00
const separatorBytes = this . entries > 0 ? 1 : 0
const stringBytes = jsonStringBytesUpTo ( text , this . maxBytes - this . bytes - separatorBytes )
if ( stringBytes === undefined ) return false
this . bytes += stringBytes + separatorBytes
2026-07-21 04:34:14 +08:00
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 {
2026-07-21 21:56:38 +08:00
if ( value !== undefined && jsonValueBytesUpTo ( value , this . maxBytes - this . bytes ) === undefined ) return this . limit ( logs )
2026-07-21 04:34:14 +08:00
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 {
2026-07-21 21:56:38 +08:00
if ( jsonStringBytesUpTo ( error . message , this . maxBytes - this . bytes ) === undefined ) return this . limit ( logs )
2026-07-21 04:34:14 +08:00
return { logs , error }
}
2026-07-21 18:19:07 +08:00
/** Build the explicit output-limit failure while retaining a fitting prefix of the final log. */
2026-07-21 04:34:14 +08:00
limit ( logs : string [ ] ) : CodeRunResult {
const fullMessage = ` outer output exceeded ${ this . maxBytes } bytes `
2026-07-21 21:56:38 +08:00
// The fixed diagnostic is ASCII, so every character is one byte plus the quotes.
const messageBytes = fullMessage . length + 2
const retained : string [ ] = [ ]
let retainedBytes = 2
2026-07-21 18:19:07 +08:00
const logBudget = this . maxBytes - messageBytes
2026-07-21 21:56:38 +08:00
for ( const text of logs ) {
2026-07-21 18:19:07 +08:00
const separatorBytes = retained . length > 0 ? 1 : 0
2026-07-21 21:56:38 +08:00
const availableBytes = logBudget - retainedBytes - separatorBytes
const stringBytes = jsonStringBytesUpTo ( text , availableBytes )
if ( stringBytes !== undefined ) {
retained . push ( text )
retainedBytes += stringBytes + separatorBytes
continue
}
const prefix = truncateJsonStringBytes ( text , availableBytes )
2026-07-21 18:19:07 +08:00
if ( prefix . length > 0 ) {
2026-07-21 21:56:38 +08:00
const prefixBytes = jsonStringBytesUpTo ( prefix , availableBytes )
/* v8 ignore next -- truncateJsonStringBytes guarantees its returned prefix fits the same budget. */
if ( prefixBytes === undefined ) throw new Error ( 'output ledger produced an oversized log prefix' )
2026-07-21 18:19:07 +08:00
retained . push ( prefix )
2026-07-21 21:56:38 +08:00
retainedBytes += prefixBytes + separatorBytes
2026-07-21 18:19:07 +08:00
}
2026-07-21 21:56:38 +08:00
break
2026-07-21 04:34:14 +08:00
}
const availableMessageBytes = this . maxBytes - retainedBytes
2026-07-21 21:56:38 +08:00
const message = truncateJsonStringBytes ( fullMessage , availableMessageBytes )
2026-07-21 18:19:07 +08:00
return { logs : retained , error : { kind : 'output-limit' , message } }
2026-07-21 04:34:14 +08:00
}
}
2026-07-08 11:39:57 +08:00
2026-07-08 11:00:06 +08:00
/**
* The shipped {@link CodeRuntime} backend (`ctx.codeRuntime`). Registers as
* the `codeRuntime` service; every cap comes from validated config. See the
2026-08-09 15:34:32 +08:00
* module doc for the containment model and the Service Definition's class JSDoc for
2026-07-08 11:00:06 +08:00
* 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 ) ,
2026-07-21 04:34:14 +08:00
maxOutputBytes : z.number ( ) . default ( 67 _108_864 ) ,
2026-07-08 11:00:06 +08:00
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 ) } ` )
}
2026-07-21 04:34:14 +08:00
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 ) } ` )
}
2026-07-27 14:01:15 +08:00
// maxWallMs reaches setTimeout, which clamps any delay above
// MAX_TIMER_DELAY_MS to 1 ms; the positivity check above accepts such a
// value, so a 25-day ceiling would time the run out immediately.
if ( this . config . maxWallMs > MAX_TIMER_DELAY_MS ) {
throw new Error ( ` dsh-code-runtime-worker: config.maxWallMs must be at most ${ MAX_TIMER_DELAY_MS } (Node clamps a longer setTimeout delay to 1ms), got ${ String ( this . config . maxWallMs ) } ` )
}
2026-07-08 11:00:06 +08:00
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
2026-08-09 15:34:32 +08:00
* `result.error`; the method rejects only for Service Definition contract misuse (a disposed
2026-07-08 11:00:06 +08:00
* 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 ) {
2026-07-21 21:20:41 +08:00
return this . failureBeforeWorker ( { kind : 'abort' , message : String ( request . signal . reason ) } )
2026-07-08 11:00:06 +08:00
}
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.
2026-07-21 21:20:41 +08:00
return this . failureBeforeWorker ( { kind : 'exception' , message : messageOf ( error ) } )
2026-07-08 11:00:06 +08:00
}
return await this . execute ( request , code , bindings )
}
2026-07-21 21:20:41 +08:00
/** Apply the outer-output ledger to failures that occur before a worker owns one. */
private failureBeforeWorker ( error : CodeRunFailure ) : CodeRunResult {
return new OutputLedger ( this . config . maxOutputBytes ) . failure ( [ ] , error )
}
2026-08-09 15:34:32 +08:00
/** Reject malformed binding globals or typed-error declarations as Service Definition contract misuse. */
2026-07-23 01:32:17 +08:00
private validateBindings ( request : CodeRunRequest ) : Map < string , CodeBindingNamespace > {
const bindings = new Map < string , CodeBindingNamespace > ( )
2026-07-08 11:00:06 +08:00
for ( const namespace of request . bindings ) {
2026-07-31 18:30:32 +08:00
if ( ! IDENTIFIER . test ( namespace . global ) || PORTABLE_RESERVED_WORDS . has ( namespace . global ) ) {
2026-07-08 11:00:06 +08:00
throw new Error ( ` dsh-code-runtime-worker: binding global ${ JSON . stringify ( namespace . global ) } is not a usable identifier ` )
}
2026-07-31 18:10:33 +08:00
// RESERVED_BINDING_GLOBALS is the seam's shared backend-owned set:
2026-08-01 15:53:48 +08:00
// `console` is THIS backend's log-capture slot; the dunder entries exist
// for the Python side — its seeded/wrapped slots plus the `__debug__`
// compile-time constant — refused here too so the namespace list stays
// portable across backends. The seam declaration is the single home for
// why each entry is reserved.
2026-07-31 18:30:32 +08:00
if ( RESERVED_BINDING_GLOBALS . has ( namespace . global ) ) {
throw new Error ( ` dsh-code-runtime-worker: reserved binding global ${ JSON . stringify ( namespace . global ) } ` )
}
if ( bindings . has ( namespace . global ) ) {
2026-07-08 11:00:06 +08:00
throw new Error ( ` dsh-code-runtime-worker: duplicate binding global ${ JSON . stringify ( namespace . global ) } ` )
}
2026-07-23 01:32:17 +08:00
bindings . set ( namespace . global , namespace )
}
const errorClassNames = new Set < string > ( )
for ( const namespace of request . bindings ) {
const descriptor = namespace . errorClass
if ( ! descriptor ) continue
2026-07-31 18:30:32 +08:00
if ( ! IDENTIFIER . test ( descriptor . name ) || PORTABLE_RESERVED_WORDS . has ( descriptor . name ) ) {
2026-07-23 01:32:17 +08:00
throw new Error ( ` dsh-code-runtime-worker: binding error class ${ JSON . stringify ( descriptor . name ) } is not a usable identifier ` )
}
2026-07-31 18:30:32 +08:00
if ( RESERVED_BINDING_GLOBALS . has ( descriptor . name ) ) {
throw new Error ( ` dsh-code-runtime-worker: reserved binding global ${ JSON . stringify ( descriptor . name ) } ` )
}
if ( bindings . has ( descriptor . name ) || errorClassNames . has ( descriptor . name ) ) {
2026-07-23 01:32:17 +08:00
throw new Error ( ` dsh-code-runtime-worker: duplicate injected global ${ JSON . stringify ( descriptor . name ) } ` )
}
2026-07-31 18:10:33 +08:00
const member = descriptor . memberNameProperty
2026-07-31 18:30:32 +08:00
if ( member . length === 0 || RESERVED_ERROR_MEMBERS . has ( member ) || DUNDER_MEMBER . test ( member ) ) {
2026-07-23 01:32:17 +08:00
throw new Error ( ` dsh-code-runtime-worker: binding error member property ${ JSON . stringify ( descriptor . memberNameProperty ) } is not usable ` )
}
errorClassNames . add ( descriptor . name )
2026-07-08 11:00:06 +08:00
}
return bindings
}
/** Spawn the worker for one validated, type-stripped run and drive it to settlement. */
private execute (
request : CodeRunRequest ,
code : string ,
2026-07-23 01:32:17 +08:00
bindings : Map < string , CodeBindingNamespace > ,
2026-07-08 11:00:06 +08:00
) : Promise < CodeRunResult > {
const bootData : WorkerBootData = {
code ,
2026-07-23 01:32:17 +08:00
namespaces : [ . . . bindings ] . map ( ( [ global , namespace ] ) = > ( {
global ,
names : Object.keys ( namespace . functions ) ,
. . . namespace . errorClass ? { errorClass : namespace.errorClass } : { } ,
} ) ) ,
2026-07-21 04:34:14 +08:00
maxOutputBytes : this.config.maxOutputBytes ,
2026-07-08 11:00:06 +08:00
}
2026-07-13 20:55:47 +08:00
const worker = new Worker ( WORKER_PATH , {
2026-07-08 11:00:06 +08:00
workerData : bootData ,
// Model code gets NO ambient environment — stronger than the scrubbed
// env the defensive-patterns rule requires for spawned commands.
env : { } ,
2026-07-12 03:36:43 +08:00
// 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.
2026-07-08 11:00:06 +08:00
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 > ( )
2026-07-14 03:07:41 +08:00
const logs : string [ ] = [ ]
const strayLogs : string [ ] = [ ]
2026-07-21 04:34:14 +08:00
const output = new OutputLedger ( this . config . maxOutputBytes )
2026-07-22 01:27:03 +08:00
let terminalOverride : CodeRunResult | undefined
2026-07-08 11:39:57 +08:00
2026-07-22 01:27:03 +08:00
// Pipe and message-port delivery are independent. Continue bounded pipe
// capture after a terminal message while worker termination drains bytes
// that were already queued; `finish` materializes the result only after
// termination completes.
2026-07-14 03:07:41 +08:00
const captureStray = ( chunk : Buffer ) : void = > {
2026-07-22 01:42:41 +08:00
/* v8 ignore next -- a second post-overflow chunk races immediate worker termination; the first overflow path is covered. */
2026-07-22 01:27:03 +08:00
if ( terminalOverride !== undefined ) return
2026-07-21 18:19:07 +08:00
const text = chunk . toString ( 'utf8' )
2026-07-22 01:27:03 +08:00
if ( ! output . admit ( text , strayLogs ) ) {
const limited = output . limit ( [ . . . logs , . . . strayLogs , text ] )
terminalOverride = limited
2026-07-22 01:42:41 +08:00
finish ( limited )
2026-07-22 01:27:03 +08:00
}
2026-07-08 11:00:06 +08:00
}
2026-07-14 03:07:41 +08:00
worker . stdout . on ( 'data' , captureStray )
worker . stderr . on ( 'data' , captureStray )
2026-07-08 11:00:06 +08:00
2026-07-13 23:27:00 +08:00
// Exactly one outcome wins. Every path cleans up, terminates, and awaits the worker;
// logs captured before timeout, abort, or failure remain in the result.
2026-07-08 11:00:06 +08:00
let finishResolve ! : ( ) = > void
const finished = new Promise < void > ( ( done ) = > { finishResolve = done } )
2026-07-22 01:42:41 +08:00
const finish = ( finalize : CodeRunResult | ( ( ) = > CodeRunResult ) ) : void = > {
2026-07-08 11:00:06 +08:00
if ( settled ) return
settled = true
clearInterval ( eluTimer )
clearTimeout ( wallTimer )
request . signal ? . removeEventListener ( 'abort' , onAbort )
this . live . delete ( live )
2026-07-22 01:27:03 +08:00
// Let the poll phase deliver pipe bytes already queued independently
// of the terminal port message before termination closes the streams.
void new Promise < void > ( ( resume ) = > { setImmediate ( resume ) } ) . then ( async ( ) = > {
const stdoutDrained = waitForPipeDrain ( worker . stdout )
const stderrDrained = waitForPipeDrain ( worker . stderr )
await Promise . all ( [ worker . terminate ( ) , stdoutDrained , stderrDrained ] )
2026-07-22 01:42:41 +08:00
const result = terminalOverride ? ? ( typeof finalize === 'function' ? finalize ( ) : finalize )
2026-07-08 11:00:06 +08:00
finishResolve ( )
2026-07-21 04:34:14 +08:00
resolve ( result )
2026-07-08 11:00:06 +08:00
} )
}
const onDone = ( message : WorkerToHost ) : void = > {
if ( message . type !== 'done' ) return
2026-07-21 04:34:14 +08:00
if ( message . error ) {
2026-07-22 01:27:03 +08:00
const error = message . error
finish ( ( ) = > output . failure ( [ . . . logs , . . . strayLogs ] , error ) )
2026-07-21 04:34:14 +08:00
return
}
if ( message . value === undefined ) {
2026-07-22 01:27:03 +08:00
finish ( ( ) = > output . success ( [ . . . logs , . . . strayLogs ] ) )
2026-07-21 04:34:14 +08:00
return
}
2026-07-22 19:09:27 +08:00
const value = decodeWorkerJson ( message . value )
2026-07-22 01:27:03 +08:00
if ( value === undefined ) {
finish ( ( ) = > output . failure ( [ . . . logs , . . . strayLogs ] , { kind : 'invalid-output' , message : 'program completion must be lossless JSON' } ) )
} else {
finish ( ( ) = > output . success ( [ . . . logs , . . . strayLogs ] , value ) )
}
2026-07-08 11:00:06 +08:00
}
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
2026-07-21 04:34:14 +08:00
// Canonical resolutions were snapshotted as lossless JSON before
// this point, so this payload is structured-cloneable by contract.
worker . postMessage ( payload )
2026-07-08 11:00:06 +08:00
}
2026-07-23 01:32:17 +08:00
const record = bindings . get ( message . global ) ? . functions
2026-07-08 11:00:06 +08:00
// 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
}
2026-07-22 19:09:27 +08:00
const args = decodeWorkerJson ( message . args )
2026-07-21 21:20:41 +08:00
if ( args === undefined ) {
reply ( { type : 'reply' , id : message.id , ok : false , message : 'binding arguments must be lossless JSON' } )
return
}
2026-07-08 11:00:06 +08:00
void ( async ( ) = > {
try {
2026-07-21 21:20:41 +08:00
const resolved = await fn ( args )
2026-07-21 04:34:14 +08:00
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 {
2026-07-22 19:09:27 +08:00
reply ( { type : 'reply' , id : message.id , ok : true , value : encodeWorkerJson ( value ) } )
2026-07-21 04:34:14 +08:00
}
2026-07-08 11:00:06 +08:00
} catch ( error : unknown ) {
reply ( { type : 'reply' , id : message.id , ok : false , message : messageOf ( error ) } )
}
} ) ( )
}
2026-07-08 11:39:57 +08:00
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
2026-07-21 04:34:14 +08:00
if ( message . type === 'log' && ! settled && ! output . admit ( message . text , logs ) ) {
2026-07-22 01:27:03 +08:00
const limited = output . limit ( [ . . . logs , . . . strayLogs , message . text ] )
2026-07-22 01:42:41 +08:00
finish ( limited )
2026-07-21 04:34:14 +08:00
return
}
if ( message . type === 'output-limit' && ! settled ) {
2026-07-22 01:27:03 +08:00
const limited = output . limit ( [ . . . logs , . . . strayLogs ] )
2026-07-22 01:42:41 +08:00
finish ( limited )
2026-07-21 04:34:14 +08:00
return
}
2026-07-08 11:00:06 +08:00
onCall ( message )
onDone ( message )
} )
worker . on ( 'error' , ( error : Error ) = > {
2026-07-22 01:27:03 +08:00
finish ( ( ) = > output . failure ( [ . . . logs , . . . strayLogs ] , { kind : 'worker-exit' , message : ` worker error: ${ error . message } ` } ) )
2026-07-08 11:00:06 +08:00
} )
worker . on ( 'exit' , ( exitCode : number ) = > {
2026-07-22 01:27:03 +08:00
finish ( ( ) = > output . failure ( [ . . . logs , . . . strayLogs ] , { kind : 'worker-exit' , message : ` worker exited with code ${ exitCode } before completing ` } ) )
2026-07-08 11:00:06 +08:00
} )
// 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 ) {
2026-07-22 01:27:03 +08:00
finish ( ( ) = > output . failure ( [ . . . logs , . . . strayLogs ] , { kind : 'timeout' , message : ` compute budget exhausted ( ${ this . config . computeMs } ms busy) ` } ) )
2026-07-08 11:00:06 +08:00
}
} , ELU_POLL_INTERVAL_MS )
const wallTimer = setTimeout ( ( ) = > {
2026-07-22 01:27:03 +08:00
finish ( ( ) = > output . failure ( [ . . . logs , . . . strayLogs ] , { kind : 'timeout' , message : ` wall-clock ceiling reached ( ${ this . config . maxWallMs } ms) ` } ) )
2026-07-08 11:00:06 +08:00
} , this . config . maxWallMs )
const onAbort = ( ) : void = > {
2026-07-22 01:27:03 +08:00
finish ( ( ) = > output . failure ( [ . . . logs , . . . strayLogs ] , { kind : 'abort' , message : String ( request . signal ? . reason ) } ) )
2026-07-08 11:00:06 +08:00
}
request . signal ? . addEventListener ( 'abort' , onAbort , { once : true } )
const live : LiveRun = {
worker ,
finished ,
2026-07-22 01:27:03 +08:00
settle : ( failure : CodeRunFailure ) = > { finish ( ( ) = > output . failure ( [ . . . logs , . . . strayLogs ] , failure ) ) } ,
2026-07-08 11:00:06 +08:00
}
this . live . add ( live )
} )
}
}
export default WorkerCodeRuntime