index.ts 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561
  1. /**
  2. * Worker-thread code runtime: a fresh worker runs each host-type-stripped TypeScript program
  3. * and bridges bindings over its message port. This is containment, not a security boundary:
  4. * model code has bash-equivalent trust despite an empty environment, a heap cap, measured
  5. * event-loop busy-time and wall-time budgets, and termination that also stops synchronous loops.
  6. * @module @deepseek-ai/dsh-code-runtime-node
  7. */
  8. import { Worker } from 'node:worker_threads'
  9. import { stripTypeScriptTypes } from 'node:module'
  10. import type { Readable } from 'node:stream'
  11. import { fileURLToPath } from 'node:url'
  12. import { Context } from '@deepseek-ai/cordis'
  13. import z from '@deepseek-ai/schemastery'
  14. import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
  15. import { CodeRuntime, DUNDER_MEMBER, PORTABLE_RESERVED_WORDS, RESERVED_BINDING_GLOBALS, RESERVED_ERROR_MEMBERS } from '@deepseek-ai/dsh-code-runtime'
  16. import type { CodeBindingNamespace, CodeJsonValue, CodeRunFailure, CodeRunRequest, CodeRunResult } from '@deepseek-ai/dsh-code-runtime'
  17. import { snapshotJsonValue } from '@deepseek-ai/dsh-util-values'
  18. import type { ReplyMessage, WorkerBootData, WorkerToHost } from './protocol.ts'
  19. import { jsonStringBytesUpTo, jsonValueBytesUpTo, truncateJsonStringBytes } from './output-json.ts'
  20. import { decodeWorkerJson, encodeWorkerJson } from './worker-json.ts'
  21. import type { WorkerJsonWire } from './worker-json.ts'
  22. /** Plugin config: every execution cap, changeable from `cordis.yml` (no hardcoded tunables). */
  23. export interface Config {
  24. /**
  25. * Busy-time budget in milliseconds: the run fails with kind `'timeout'`
  26. * once the worker's MEASURED event-loop active time
  27. * (`worker.performance.eventLoopUtilization()`) exceeds this. Metering
  28. * measured busy time — not wall time, not host-side pending-call
  29. * bookkeeping — is what makes the budget both fair (a program awaiting a
  30. * slow tool accrues nothing) and ungameable (a hot loop accrues whether
  31. * or not a decoy dispatch is in flight).
  32. */
  33. computeMs?: number
  34. /**
  35. * Wall-clock ceiling in milliseconds; never pauses for anything. The
  36. * backstop for what busy-time cannot see (a program awaiting a promise
  37. * nobody will resolve). At most `2_147_483_647` (Node's maximum
  38. * `setTimeout` delay, about 24.9 days): a longer value is rejected at load
  39. * because `setTimeout` would clamp it to 1 ms.
  40. */
  41. maxWallMs?: number
  42. /**
  43. * Hard cap for serialized log-array, completion-value, and failure-message payloads;
  44. * fixed result-envelope syntax is excluded.
  45. */
  46. maxOutputBytes?: number
  47. /** The worker's max old-generation heap in MiB (`resourceLimits`); overflow kills the worker, surfacing as kind `'worker-exit'`. */
  48. maxOldGenerationSizeMb?: number
  49. }
  50. /** {@link Config} after schemastery fills the defaults (every field present). */
  51. type ResolvedConfig = Required<Config>
  52. /**
  53. * How often the host samples the worker's event-loop utilization for the
  54. * `computeMs` budget. An internal cadence, not config: the only effect of
  55. * the interval is budget-expiry granularity (a run can overshoot by up to
  56. * one interval), and nothing a deployment could tune here improves that
  57. * without burning host CPU.
  58. */
  59. const ELU_POLL_INTERVAL_MS = 25
  60. /** Smallest cap that can represent the counted payloads: an empty logs array plus an empty JSON failure message. */
  61. const MIN_OUTPUT_BYTES = 4
  62. /**
  63. * The seam's language-portable identifier subset (see
  64. * `CodeBindingNamespace.global`): no `$`, which is JS-only spelling — the same
  65. * namespace list must be usable against every backend regardless of language.
  66. */
  67. const IDENTIFIER = /^[A-Za-z_][A-Za-z0-9_]*$/
  68. /**
  69. * The shell a program is wrapped in for the type-strip, matching the
  70. * grammatical context it will execute in (an async function body, where
  71. * top-level `return` and `await` are legal — a bare module parse would
  72. * reject the `return`). Strip mode is position-preserving (removed syntax
  73. * becomes whitespace, nothing shifts), so the wrapper survives the strip
  74. * byte-identical and the body slices back out with the model's own
  75. * line/column positions intact.
  76. */
  77. const STRIP_WRAP = { prefix: 'async function __dsh_program__() {\n', suffix: '\n}' } as const
  78. /** One in-flight run's host-side state, tracked for disposal. */
  79. interface LiveRun {
  80. worker: Worker
  81. settle(failure: CodeRunFailure): void
  82. finished: Promise<void>
  83. }
  84. /**
  85. * The worker entry path. Source runs unbuilt (`src/worker.ts`, loadable
  86. * directly on this repo's Node range via native type stripping — the file
  87. * is erasable-only with type-only relative imports); the built package
  88. * ships it as a sibling CommonJS bundle (`lib/worker.cjs`, its own tsdown
  89. * entry) because pkg's VFS Worker hook compiles string-path entries as
  90. * CommonJS.
  91. * The URL *pathname*'s extension says which world this module is in —
  92. * pathname, because dev-time module runners (vitest) may suffix
  93. * `import.meta.url` with a query string; relative resolution drops it. Worker
  94. * receives a filesystem string so pkg's VFS Worker hook can resolve it.
  95. */
  96. /* v8 ignore next -- the './worker.cjs' arm is the built-lib world, unreachable unbuilt by construction; the built-lib e2e pins it. */
  97. const WORKER_PATH = fileURLToPath(new URL(new URL(import.meta.url).pathname.endsWith('.ts') ? './worker.ts' : './worker.cjs', import.meta.url))
  98. /** Render an unknown thrown value as a message, `Error` or not. */
  99. function messageOf(error: unknown): string {
  100. return error instanceof Error ? error.message : String(error)
  101. }
  102. /** Resolve after a worker pipe emits all queued data, or closes/errors during termination. */
  103. function waitForPipeDrain(stream: Readable): Promise<void> {
  104. if (stream.readableEnded || stream.destroyed) return Promise.resolve()
  105. return new Promise((resolve) => {
  106. const done = (): void => {
  107. stream.off('end', done)
  108. stream.off('close', done)
  109. stream.off('error', done)
  110. resolve()
  111. }
  112. stream.once('end', done)
  113. stream.once('close', done)
  114. stream.once('error', done)
  115. // Close the event-registration race if termination finished between the
  116. // initial state check and the listeners above.
  117. /* v8 ignore next -- this race cannot be scheduled deterministically between the adjacent state check and listener registration. */
  118. if (stream.readableEnded || stream.destroyed) done()
  119. })
  120. }
  121. /**
  122. * Runtime shape gate for inbound port traffic. The peer runs MODEL CODE and
  123. * can post anything — `null`, primitives, objects with poisoned fields — so
  124. * the compile-time `WorkerToHost` type means nothing here: everything is
  125. * re-validated and REBUILT field by field (a forged extra field never rides
  126. * along; a non-number call id can never be echoed into a reply). Junk returns
  127. * `undefined` and is dropped — a throw in the host's `message` listener would
  128. * crash the host process.
  129. */
  130. function parseWorkerMessage(raw: unknown): WorkerToHost | undefined {
  131. if (typeof raw !== 'object' || raw === null) return undefined
  132. const m = raw as Record<string, unknown>
  133. switch (m.type) {
  134. case 'call': {
  135. if (typeof m.id !== 'number' || typeof m.global !== 'string' || typeof m.name !== 'string') return undefined
  136. return { type: 'call', id: m.id, global: m.global, name: m.name, args: m.args as WorkerJsonWire }
  137. }
  138. case 'log': {
  139. if (typeof m.text !== 'string') return undefined
  140. return { type: 'log', text: m.text }
  141. }
  142. case 'output-limit': return { type: 'output-limit' }
  143. case 'done': {
  144. if (m.error === undefined) return { type: 'done', ...m.value !== undefined ? { value: m.value as WorkerJsonWire } : {} }
  145. const error = m.error
  146. if (typeof error !== 'object' || error === null) return undefined
  147. const { kind, message } = error as Record<string, unknown>
  148. if ((kind !== 'exception' && kind !== 'invalid-output' && kind !== 'output-limit') || typeof message !== 'string') return undefined
  149. return { type: 'done', error: { kind, message } }
  150. }
  151. default: return undefined
  152. }
  153. }
  154. /** One run's combined outer-output ledger; binding values never enter it. */
  155. class OutputLedger {
  156. private bytes = 2 // JSON serialization of the empty logs array: []
  157. private entries = 0
  158. constructor(private readonly maxBytes: number) {}
  159. /** Admit one exact log entry, or report that the hard cap was crossed. */
  160. admit(text: string, sink: string[]): boolean {
  161. const separatorBytes = this.entries > 0 ? 1 : 0
  162. const stringBytes = jsonStringBytesUpTo(text, this.maxBytes - this.bytes - separatorBytes)
  163. if (stringBytes === undefined) return false
  164. this.bytes += stringBytes + separatorBytes
  165. this.entries += 1
  166. sink.push(text)
  167. return true
  168. }
  169. /** Finalize a successful absent-or-JSON completion against the combined cap. */
  170. success(logs: string[], value?: CodeJsonValue): CodeRunResult {
  171. if (value !== undefined && jsonValueBytesUpTo(value, this.maxBytes - this.bytes) === undefined) return this.limit(logs)
  172. return { logs, ...value !== undefined ? { value } : {} }
  173. }
  174. /** Finalize a failure diagnostic, with output-limit taking precedence when combined bytes exceed the cap. */
  175. failure(logs: string[], error: CodeRunFailure): CodeRunResult {
  176. if (jsonStringBytesUpTo(error.message, this.maxBytes - this.bytes) === undefined) return this.limit(logs)
  177. return { logs, error }
  178. }
  179. /** Build the explicit output-limit failure while retaining a fitting prefix of the final log. */
  180. limit(logs: string[]): CodeRunResult {
  181. const fullMessage = `outer output exceeded ${this.maxBytes} bytes`
  182. // The fixed diagnostic is ASCII, so every character is one byte plus the quotes.
  183. const messageBytes = fullMessage.length + 2
  184. const retained: string[] = []
  185. let retainedBytes = 2
  186. const logBudget = this.maxBytes - messageBytes
  187. for (const text of logs) {
  188. const separatorBytes = retained.length > 0 ? 1 : 0
  189. const availableBytes = logBudget - retainedBytes - separatorBytes
  190. const stringBytes = jsonStringBytesUpTo(text, availableBytes)
  191. if (stringBytes !== undefined) {
  192. retained.push(text)
  193. retainedBytes += stringBytes + separatorBytes
  194. continue
  195. }
  196. const prefix = truncateJsonStringBytes(text, availableBytes)
  197. if (prefix.length > 0) {
  198. const prefixBytes = jsonStringBytesUpTo(prefix, availableBytes)
  199. /* v8 ignore next -- truncateJsonStringBytes guarantees its returned prefix fits the same budget. */
  200. if (prefixBytes === undefined) throw new Error('output ledger produced an oversized log prefix')
  201. retained.push(prefix)
  202. retainedBytes += prefixBytes + separatorBytes
  203. }
  204. break
  205. }
  206. const availableMessageBytes = this.maxBytes - retainedBytes
  207. const message = truncateJsonStringBytes(fullMessage, availableMessageBytes)
  208. return { logs: retained, error: { kind: 'output-limit', message } }
  209. }
  210. }
  211. /**
  212. * The shipped {@link CodeRuntime} backend (`ctx.codeRuntime`). Registers as
  213. * the `codeRuntime` service; every cap comes from validated config. See the
  214. * module doc for the containment model and the Service Definition's class JSDoc for
  215. * the contract this implements (error-as-field, hostile-peer port,
  216. * no cross-run state, dispose to quiescence).
  217. */
  218. export class NodeCodeRuntime extends CodeRuntime {
  219. static Config: z<Config> = z.object({
  220. computeMs: z.number().default(60_000),
  221. maxWallMs: z.number().default(600_000),
  222. maxOutputBytes: z.number().default(67_108_864),
  223. maxOldGenerationSizeMb: z.number().default(512),
  224. })
  225. readonly language = 'typescript'
  226. readonly isolation = 'worker-thread'
  227. private readonly config: ResolvedConfig
  228. private readonly live = new Set<LiveRun>()
  229. private disposed = false
  230. constructor(ctx: Context, config: Config) {
  231. super(ctx)
  232. // Schemastery filled the defaults; the cast records that. Positivity is a
  233. // semantic check the schema's plain number type does not carry.
  234. this.config = config as ResolvedConfig
  235. for (const [key, value] of Object.entries(this.config)) {
  236. if (!(Number.isFinite(value) && value > 0)) throw new Error(`dsh-code-runtime-node: config.${key} must be a positive number, got ${String(value)}`)
  237. }
  238. if (!Number.isSafeInteger(this.config.maxOutputBytes) || this.config.maxOutputBytes < MIN_OUTPUT_BYTES) {
  239. throw new Error(`dsh-code-runtime-node: config.maxOutputBytes must be a safe integer of at least ${MIN_OUTPUT_BYTES}, got ${String(this.config.maxOutputBytes)}`)
  240. }
  241. // maxWallMs reaches setTimeout, which clamps any delay above
  242. // MAX_TIMER_DELAY_MS to 1 ms; the positivity check above accepts such a
  243. // value, so a 25-day ceiling would time the run out immediately.
  244. if (this.config.maxWallMs > MAX_TIMER_DELAY_MS) {
  245. throw new Error(`dsh-code-runtime-node: config.maxWallMs must be at most ${MAX_TIMER_DELAY_MS} (Node clamps a longer setTimeout delay to 1ms), got ${String(this.config.maxWallMs)}`)
  246. }
  247. ctx.effect(() => () => this.teardown(), 'worker code-runtime teardown')
  248. }
  249. /**
  250. * Dispose to quiescence: mark the service unusable, fail every in-flight
  251. * run as aborted, and AWAIT each worker's exit so no worker outlives the
  252. * fiber.
  253. */
  254. private async teardown(): Promise<void> {
  255. this.disposed = true
  256. const runs = [...this.live]
  257. for (const run of runs) run.settle({ kind: 'abort', message: 'runtime disposed' })
  258. await Promise.all(runs.map(run => run.finished))
  259. }
  260. /**
  261. * Execute one program in a fresh worker. Program outcomes — including a
  262. * type-strip syntax error, which never spawns a worker — resolve with
  263. * `result.error`; the method rejects only for Service Definition contract misuse (a disposed
  264. * runtime, an invalid binding namespace).
  265. * @param request - the program, its bindings, and the abort signal.
  266. * @returns the run's outcome per the seam contract.
  267. */
  268. async run(request: CodeRunRequest): Promise<CodeRunResult> {
  269. if (this.disposed) throw new Error('dsh-code-runtime-node: run() after disposal')
  270. const bindings = this.validateBindings(request)
  271. if (request.signal?.aborted) {
  272. return this.failureBeforeWorker({ kind: 'abort', message: String(request.signal.reason) })
  273. }
  274. let code: string
  275. try {
  276. const stripped = stripTypeScriptTypes(STRIP_WRAP.prefix + request.program + STRIP_WRAP.suffix)
  277. code = stripped.slice(STRIP_WRAP.prefix.length, stripped.length - STRIP_WRAP.suffix.length)
  278. } catch (error: unknown) {
  279. // A program that does not survive the type-strip (syntax error,
  280. // non-erasable syntax like `enum`) is a program failure, reported the
  281. // same way a thrown exception would be — and no worker ever spawns.
  282. return this.failureBeforeWorker({ kind: 'exception', message: messageOf(error) })
  283. }
  284. return await this.execute(request, code, bindings)
  285. }
  286. /** Apply the outer-output ledger to failures that occur before a worker owns one. */
  287. private failureBeforeWorker(error: CodeRunFailure): CodeRunResult {
  288. return new OutputLedger(this.config.maxOutputBytes).failure([], error)
  289. }
  290. /** Reject malformed binding globals or typed-error declarations as Service Definition contract misuse. */
  291. private validateBindings(request: CodeRunRequest): Map<string, CodeBindingNamespace> {
  292. const bindings = new Map<string, CodeBindingNamespace>()
  293. for (const namespace of request.bindings) {
  294. if (!IDENTIFIER.test(namespace.global) || PORTABLE_RESERVED_WORDS.has(namespace.global)) {
  295. throw new Error(`dsh-code-runtime-node: binding global ${JSON.stringify(namespace.global)} is not a usable identifier`)
  296. }
  297. // RESERVED_BINDING_GLOBALS is the seam's shared backend-owned set:
  298. // `console` is THIS backend's log-capture slot; the dunder entries exist
  299. // for the Python side — its seeded/wrapped slots plus the `__debug__`
  300. // compile-time constant — refused here too so the namespace list stays
  301. // portable across backends. The seam declaration is the single home for
  302. // why each entry is reserved.
  303. if (RESERVED_BINDING_GLOBALS.has(namespace.global)) {
  304. throw new Error(`dsh-code-runtime-node: reserved binding global ${JSON.stringify(namespace.global)}`)
  305. }
  306. if (bindings.has(namespace.global)) {
  307. throw new Error(`dsh-code-runtime-node: duplicate binding global ${JSON.stringify(namespace.global)}`)
  308. }
  309. bindings.set(namespace.global, namespace)
  310. }
  311. const errorClassNames = new Set<string>()
  312. for (const namespace of request.bindings) {
  313. const descriptor = namespace.errorClass
  314. if (!descriptor) continue
  315. if (!IDENTIFIER.test(descriptor.name) || PORTABLE_RESERVED_WORDS.has(descriptor.name)) {
  316. throw new Error(`dsh-code-runtime-node: binding error class ${JSON.stringify(descriptor.name)} is not a usable identifier`)
  317. }
  318. if (RESERVED_BINDING_GLOBALS.has(descriptor.name)) {
  319. throw new Error(`dsh-code-runtime-node: reserved binding global ${JSON.stringify(descriptor.name)}`)
  320. }
  321. if (bindings.has(descriptor.name) || errorClassNames.has(descriptor.name)) {
  322. throw new Error(`dsh-code-runtime-node: duplicate injected global ${JSON.stringify(descriptor.name)}`)
  323. }
  324. const member = descriptor.memberNameProperty
  325. if (member.length === 0 || RESERVED_ERROR_MEMBERS.has(member) || DUNDER_MEMBER.test(member)) {
  326. throw new Error(`dsh-code-runtime-node: binding error member property ${JSON.stringify(descriptor.memberNameProperty)} is not usable`)
  327. }
  328. errorClassNames.add(descriptor.name)
  329. }
  330. return bindings
  331. }
  332. /** Spawn the worker for one validated, type-stripped run and drive it to settlement. */
  333. private execute(
  334. request: CodeRunRequest,
  335. code: string,
  336. bindings: Map<string, CodeBindingNamespace>,
  337. ): Promise<CodeRunResult> {
  338. const bootData: WorkerBootData = {
  339. code,
  340. namespaces: [...bindings].map(([global, namespace]) => ({
  341. global,
  342. names: Object.keys(namespace.functions),
  343. ...namespace.errorClass ? { errorClass: namespace.errorClass } : {},
  344. })),
  345. maxOutputBytes: this.config.maxOutputBytes,
  346. }
  347. const worker = new Worker(WORKER_PATH, {
  348. workerData: bootData,
  349. // Model code gets NO ambient environment — stronger than the scrubbed
  350. // env the defensive-patterns rule requires for spawned commands.
  351. env: {},
  352. // Hermetic flags too: without this the worker inherits the host process's execArgv (a
  353. // test runner's or tsx's loader hooks), which a bare isolate with an empty environment
  354. // cannot satisfy.
  355. execArgv: [],
  356. resourceLimits: { maxOldGenerationSizeMb: this.config.maxOldGenerationSizeMb },
  357. // Backstop capture: the bootstrap patches JS-level writes into its own
  358. // ordered buffer, so these pipes normally stay silent; anything that
  359. // still arrives (native-level writes) is appended after the done logs.
  360. stdout: true,
  361. stderr: true,
  362. })
  363. return new Promise<CodeRunResult>((resolve) => {
  364. let settled = false
  365. const answered = new Set<number>()
  366. const logs: string[] = []
  367. const strayLogs: string[] = []
  368. const output = new OutputLedger(this.config.maxOutputBytes)
  369. let terminalOverride: CodeRunResult | undefined
  370. // Pipe and message-port delivery are independent. Continue bounded pipe
  371. // capture after a terminal message while worker termination drains bytes
  372. // that were already queued; `finish` materializes the result only after
  373. // termination completes.
  374. const captureStray = (chunk: Buffer): void => {
  375. /* v8 ignore next -- a second post-overflow chunk races immediate worker termination; the first overflow path is covered. */
  376. if (terminalOverride !== undefined) return
  377. const text = chunk.toString('utf8')
  378. if (!output.admit(text, strayLogs)) {
  379. const limited = output.limit([...logs, ...strayLogs, text])
  380. terminalOverride = limited
  381. finish(limited)
  382. }
  383. }
  384. worker.stdout.on('data', captureStray)
  385. worker.stderr.on('data', captureStray)
  386. // Exactly one outcome wins. Every path cleans up, terminates, and awaits the worker;
  387. // logs captured before timeout, abort, or failure remain in the result.
  388. let finishResolve!: () => void
  389. const finished = new Promise<void>((done) => { finishResolve = done })
  390. const finish = (finalize: CodeRunResult | (() => CodeRunResult)): void => {
  391. if (settled) return
  392. settled = true
  393. clearInterval(eluTimer)
  394. clearTimeout(wallTimer)
  395. request.signal?.removeEventListener('abort', onAbort)
  396. this.live.delete(live)
  397. // Let the poll phase deliver pipe bytes already queued independently
  398. // of the terminal port message before termination closes the streams.
  399. void new Promise<void>((resume) => { setImmediate(resume) }).then(async () => {
  400. const stdoutDrained = waitForPipeDrain(worker.stdout)
  401. const stderrDrained = waitForPipeDrain(worker.stderr)
  402. await Promise.all([worker.terminate(), stdoutDrained, stderrDrained])
  403. const result = terminalOverride ?? (typeof finalize === 'function' ? finalize() : finalize)
  404. finishResolve()
  405. resolve(result)
  406. })
  407. }
  408. const onDone = (message: WorkerToHost): void => {
  409. if (message.type !== 'done') return
  410. if (message.error) {
  411. const error = message.error
  412. finish(() => output.failure([...logs, ...strayLogs], error))
  413. return
  414. }
  415. if (message.value === undefined) {
  416. finish(() => output.success([...logs, ...strayLogs]))
  417. return
  418. }
  419. const value = decodeWorkerJson(message.value)
  420. if (value === undefined) {
  421. finish(() => output.failure([...logs, ...strayLogs], { kind: 'invalid-output', message: 'program completion must be lossless JSON' }))
  422. } else {
  423. finish(() => output.success([...logs, ...strayLogs], value))
  424. }
  425. }
  426. const onCall = (message: WorkerToHost): void => {
  427. if (message.type !== 'call' || settled) return
  428. // Hostile-peer rules: a duplicate id is ignored, an unknown name is
  429. // answered with a failure, and a binding throw/reject becomes the
  430. // program-side rejection — contained here, never a host crash.
  431. if (answered.has(message.id)) return
  432. answered.add(message.id)
  433. const reply = (payload: ReplyMessage): void => {
  434. if (settled) return
  435. // Canonical resolutions were snapshotted as lossless JSON before
  436. // this point, so this payload is structured-cloneable by contract.
  437. worker.postMessage(payload)
  438. }
  439. const record = bindings.get(message.global)?.functions
  440. // Own-property lookup only: a forged name like 'constructor' or
  441. // 'hasOwnProperty' must not walk the record's prototype chain and
  442. // reach a callable the consumer never declared.
  443. const fn = record && Object.hasOwn(record, message.name) ? record[message.name] : undefined
  444. if (typeof fn !== 'function') {
  445. reply({ type: 'reply', id: message.id, ok: false, message: `unknown binding ${JSON.stringify(`${message.global}.${message.name}`)}` })
  446. return
  447. }
  448. const args = decodeWorkerJson(message.args)
  449. if (args === undefined) {
  450. reply({ type: 'reply', id: message.id, ok: false, message: 'binding arguments must be lossless JSON' })
  451. return
  452. }
  453. void (async () => {
  454. try {
  455. const resolved = await fn(args)
  456. let value: CodeJsonValue | undefined
  457. try {
  458. value = snapshotJsonValue(resolved)
  459. } catch {
  460. value = undefined
  461. }
  462. if (value === undefined) {
  463. reply({ type: 'reply', id: message.id, ok: false, message: 'binding resolution must be lossless JSON' })
  464. } else {
  465. reply({ type: 'reply', id: message.id, ok: true, value: encodeWorkerJson(value) })
  466. }
  467. } catch (error: unknown) {
  468. reply({ type: 'reply', id: message.id, ok: false, message: messageOf(error) })
  469. }
  470. })()
  471. }
  472. worker.on('message', (raw: unknown) => {
  473. // Parse before touching: the peer can post ANY shape, and a throw in
  474. // this listener would crash the host process. Junk drops silently.
  475. const message = parseWorkerMessage(raw)
  476. if (!message) return
  477. if (message.type === 'log' && !settled && !output.admit(message.text, logs)) {
  478. const limited = output.limit([...logs, ...strayLogs, message.text])
  479. finish(limited)
  480. return
  481. }
  482. if (message.type === 'output-limit' && !settled) {
  483. const limited = output.limit([...logs, ...strayLogs])
  484. finish(limited)
  485. return
  486. }
  487. onCall(message)
  488. onDone(message)
  489. })
  490. worker.on('error', (error: Error) => {
  491. finish(() => output.failure([...logs, ...strayLogs], { kind: 'worker-exit', message: `worker error: ${error.message}` }))
  492. })
  493. worker.on('exit', (exitCode: number) => {
  494. finish(() => output.failure([...logs, ...strayLogs], { kind: 'worker-exit', message: `worker exited with code ${exitCode} before completing` }))
  495. })
  496. // The compute budget reads the worker's own measured busy time, so a
  497. // hot loop expires it no matter what dispatches are in flight, while a
  498. // program idling on a slow binding accrues nothing.
  499. const eluTimer = setInterval(() => {
  500. const elu = worker.performance.eventLoopUtilization()
  501. if (elu.active > this.config.computeMs) {
  502. finish(() => output.failure([...logs, ...strayLogs], { kind: 'timeout', message: `compute budget exhausted (${this.config.computeMs}ms busy)` }))
  503. }
  504. }, ELU_POLL_INTERVAL_MS)
  505. const wallTimer = setTimeout(() => {
  506. finish(() => output.failure([...logs, ...strayLogs], { kind: 'timeout', message: `wall-clock ceiling reached (${this.config.maxWallMs}ms)` }))
  507. }, this.config.maxWallMs)
  508. const onAbort = (): void => {
  509. finish(() => output.failure([...logs, ...strayLogs], { kind: 'abort', message: String(request.signal?.reason) }))
  510. }
  511. request.signal?.addEventListener('abort', onAbort, { once: true })
  512. const live: LiveRun = {
  513. worker,
  514. finished,
  515. settle: (failure: CodeRunFailure) => { finish(() => output.failure([...logs, ...strayLogs], failure)) },
  516. }
  517. this.live.add(live)
  518. })
  519. }
  520. }
  521. export default NodeCodeRuntime