test(code-runtime): complete sandboxed process coverage and catalogs

This commit is contained in:
Tianyi Cui committed 2026-09-12 15:03:01 +08:00
1 parent 8e19ec4962
commit fc5cb0f649
31 files changed
+361 -169

No files matched your search

@@ -20,7 +20,7 @@ function defineBindingErrorField(error: Error, key: string, value: string): void
capturedObjectDefineProperty(error, key, attributes)
}
/** The port API the bootstrap needs — satisfied by `parentPort` and by the tests' fake. */
/** Program messages and host replies transported by the private process channel. */
export interface BootstrapPort {
postMessage(message: ProgramToHost): void
on(event: 'message', listener: (message: ReplyMessage) => void): void
@@ -76,7 +76,7 @@ export class LogBuffer {
if (prefix.length > 0) {
const prefixBytes = jsonStringBytesUpTo(prefix, availableBytes)
/* v8 ignore next -- truncateJsonStringBytes guarantees the returned prefix fits. */
if (prefixBytes === undefined) throw new CapturedError('worker output ledger produced an oversized log prefix')
if (prefixBytes === undefined) throw new CapturedError('program output ledger produced an oversized log prefix')
this.bytes += prefixBytes + separatorBytes
this.entries += 1
this.sink(prefix)
@@ -127,7 +127,7 @@ export function makeConsoleShim(logs: LogBuffer): Record<(typeof CONSOLE_LEVELS)
* @param logs - the buffer captured writes are pushed into.
* @param stream - the stream whose `write` slot is patched.
* @returns the restore function (the in-process tests un-patch; the real
* worker never needs to).
* child never needs to).
*/
export function captureStreamWrites(logs: LogBuffer, stream: PatchableStream): () => void {
// The slot's VALUE is stored for restore and reassigned — never invoked
@@ -154,7 +154,7 @@ const INSPECT_OPTIONS = { depth: 4, maxArrayLength: 100, maxStringLength: 10_000
* Prepare the program's completion value for the done message. Only lossless
* JSON crosses, and a value that does not fit the remaining combined outer
* budget reports `output-limit`; the host revalidates hostile traffic and
* remains authoritative for native pipe writes the worker cannot observe.
* remains authoritative for native pipe writes the program shim cannot observe.
*
* @param value - the program's completion value.
* @param remainingOutputBytes - exact bytes left after captured logs.
@@ -205,7 +205,7 @@ function prepareFailure(
/**
* Prepare a thrown program value without sending an unbounded stack or
* string across the worker port.
* string across the process channel.
* @param error - the value thrown by the program.
* @param remainingOutputBytes - exact bytes left after captured logs.
* @param maxOutputBytes - the configured cap named in an overflow diagnostic.
@@ -306,7 +306,7 @@ export function wireReplies(port: BootstrapPort, pending: Map<number, PendingCal
* @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).
* @param nextId - the shared mutable id counter (program-issued correlation ids).
* @param errorClasses - per-namespace constructors shared with program globals.
* @returns one namespace object per declaration, in declaration order.
*/
@@ -42,6 +42,8 @@ export class JsonChannel {
if (this.closed) return
try {
let offset = 0
// receive can synchronously close the channel while dispatching a preceding frame.
// oxlint-disable-next-line typescript/no-unnecessary-condition
while (offset < chunk.length && !this.closed) {
if (this.payload === undefined) {
const bytes = Math.min(4 - this.headerBytes, chunk.length - offset)
@@ -192,6 +192,8 @@ export class NodeCodeRuntime extends CodeRuntime {
signal.addEventListener('abort', onAbort, { once: true })
if (signal.aborted) onAbort()
try {
// Abort callbacks can settle execution before or during an awaited operation.
// oxlint-disable-next-line typescript/no-unnecessary-condition
if (settled) return await result.promise
const stripped = stripTypeScriptTypes(STRIP_PREFIX + spec.program + STRIP_SUFFIX)
parsing = false
@@ -205,6 +207,8 @@ export class NodeCodeRuntime extends CodeRuntime {
maxOutputBytes: this.config.maxOutputBytes,
}
const executable = await this.ctx.subprocess.resolveExecutable(this.config.nodeExecutable, undefined, signal)
// Abort callbacks can settle execution before or during an awaited operation.
// oxlint-disable-next-line typescript/no-unnecessary-condition
if (settled) return await result.promise
const argv = [executable, `--max-old-space-size=${this.config.maxOldGenerationSizeMb}`, ...bootstrapArgs(this.ctx.fs, this.config, this.config.maxMessageBytes)]
confined = policy.mode === 'danger-full-access' ? undefined : this.ctx.sandbox.confine(argv, { ...policy, mode: policy.mode })
@@ -261,7 +265,7 @@ export class NodeCodeRuntime extends CodeRuntime {
if (!ready) {
if (raw.type !== 'ready') { protocolFailure('program frame arrived before bootstrap readiness'); return }
ready = true
void transport.send({ type: 'boot', data }).catch((error) => { protocolFailure(messageOf(error)) })
void transport.send({ type: 'boot', data }).catch((error: unknown) => { protocolFailure(messageOf(error)) })
return
}
switch (raw.type) {
@@ -296,7 +300,7 @@ export class NodeCodeRuntime extends CodeRuntime {
const args = decodeCodeJsonWire(raw.args)
if (args === undefined) { protocolFailure('binding arguments must be lossless JSON'); return }
if (++pending > this.config.maxPendingCalls || (pendingBytes += bytes) > this.config.maxMessageBytes) { protocolFailure('pending binding calls exceed configured limits'); return }
const id = raw.id as number
const id = raw.id
void (async () => {
let reply: unknown
try {
@@ -310,7 +314,7 @@ export class NodeCodeRuntime extends CodeRuntime {
pendingBytes -= bytes
}
if (!settled) await transport.send(reply)
})().catch((error) => { protocolFailure(messageOf(error)) })
})().catch((error: unknown) => { protocolFailure(messageOf(error)) })
return
}
default: protocolFailure('unknown control message')
@@ -318,10 +322,10 @@ export class NodeCodeRuntime extends CodeRuntime {
}, (error, kind) => {
if (kind === 'protocol') protocolFailure(messageOf(error))
else if (ready) finish({ kind: 'worker-exit', message: messageOf(error) })
else void launched.done.then(processFinished, (failure) => { finish({ kind: 'worker-exit', message: messageOf(failure) }) })
else void launched.done.then(processFinished, (failure: unknown) => { finish({ kind: 'worker-exit', message: messageOf(failure) }) })
})
channel = transport
void launched.done.then((outcome) => { setImmediate(() => { processFinished(outcome) }) }, (error) => { finish({ kind: confined !== undefined && isRunnerSpawnFailure(error, confined.argv[0], spec.cwd) ? 'sandbox-unavailable' : 'worker-exit', message: messageOf(error) }) })
void launched.done.then((outcome) => { setImmediate(() => { processFinished(outcome) }) }, (error: unknown) => { finish({ kind: confined !== undefined && isRunnerSpawnFailure(error, confined.argv[0], spec.cwd) ? 'sandbox-unavailable' : 'worker-exit', message: messageOf(error) }) })
} catch (error: unknown) {
finish({ kind: error instanceof SandboxUnavailableError ? 'sandbox-unavailable' : parsing ? 'exception' : 'worker-exit', message: messageOf(error) })
}
@@ -1,11 +1,11 @@
/**
* Lossless-JSON snapshots for the dependency-free source worker closure.
* Lossless-JSON snapshots for the dependency-free source bootstrap closure.
* @module @deepseek-ai/dsh-code-runtime-node/json-wire
*/
import type { CodeJsonValue } from '@deepseek-ai/dsh-code-runtime'
/* jscpd:ignore-start -- the source worker mirrors session JSON helpers without workspace runtime imports */
/* jscpd:ignore-start -- the source bootstrap mirrors session JSON helpers without workspace runtime imports */
type IntrinsicCallable = (this: unknown, ...args: unknown[]) => unknown
const intrinsicFunctionToString = Reflect.get(Function.prototype, 'toString') as IntrinsicCallable
@@ -139,9 +139,9 @@ type SnapshotTask =
| { kind: 'leave'; source: object }
/**
* Validate and detach one worker-boundary value without loading another
* Validate and detach one process-boundary value without loading another
* workspace package at runtime. This mirrors the session-owned canonical
* JSON boundary while remaining safe to import from the unbuilt worker.
* JSON boundary while remaining safe to import from the unbuilt bootstrap.
* Its iterative traversal adds no JavaScript call-stack depth limit.
*
* @param value - the candidate completion value.
@@ -242,17 +242,17 @@ interface ObjectWireToken {
keys: string[]
}
type WorkerJsonToken = null | boolean | number | string | ArrayWireToken | ObjectWireToken
type CodeJsonToken = null | boolean | number | string | ArrayWireToken | ObjectWireToken
/**
* A pre-order, bounded-depth transport for one lossless JSON value. Container
* markers and scalar leaves share one flat token array, so `worker_threads`
* never has to structured-clone the value's application nesting.
* markers and scalar leaves share one flat token array, so JSON serialization
* does not recurse through the value's application nesting.
*/
export type CodeJsonWire = WorkerJsonToken[]
export type CodeJsonWire = CodeJsonToken[]
/**
* Flatten one validated JSON value for the worker-thread message port.
* Flatten one validated JSON value for the process control channel.
* @param value - the lossless JSON value to transport.
* @returns a pre-order token stream whose own nesting is bounded.
*/
@@ -340,7 +340,7 @@ function containerToken(value: object): ArrayWireToken | ObjectWireToken | undef
}
/**
* Rebuild one lossless JSON value from the flat worker-thread wire format.
* Rebuild one lossless JSON value from the flat process wire format.
* Malformed or incomplete traffic returns `undefined`; traversal is iterative
* and therefore independent of the transported value's application depth.
* @param input - untrusted message-port payload.
@@ -26,6 +26,7 @@ export function bootstrapArgs(fs: FileSystem, config: LaunchConfig, maxMessageBy
if (result === undefined) throw new Error(`code runtime bootstrap is unavailable in the subprocess execution world: ${path}`)
return result
}
/* v8 ignore next 3 -- built-lib.e2e.ts executes the bundled provider and sibling process.js under plain Node. */
if (!new URL(import.meta.url).pathname.endsWith('.ts')) {
return [mapped(fileURLToPath(new URL('./process.js', import.meta.url))), String(maxMessageBytes)]
}
@@ -64,6 +64,8 @@ export async function runNodeMain(stream: Duplex, maxMessageBytes: number, proce
await channel.drain()
} finally {
channel.close()
// Transport callbacks can set failed while the awaited program executes.
// oxlint-disable-next-line typescript/no-unnecessary-condition
if (failed) processState.exitCode = 1
}
}
@@ -25,8 +25,8 @@ describe('bounded process frames', () => {
it('reassembles split binary headers and UTF-8 bodies', async () => {
const { a, b } = pair()
const received = Promise.withResolvers<unknown>()
const channel = new JsonChannel(a, 1000, value => received.resolve(value), error => received.reject(error))
onTestFinished(() => channel.close())
const channel = new JsonChannel(a, 1000, (value) => { received.resolve(value) }, (error) => { received.reject(error) })
onTestFinished(() => { channel.close() })
const bytes = frame({ text: '你好🙂' })
for (const byte of bytes) b.write(Buffer.from([byte]))
expect(await received.promise).toEqual({ text: '你好🙂' })
@@ -39,8 +39,8 @@ describe('bounded process frames', () => {
const channel = new JsonChannel(a, 1000, (value) => {
received.push(value)
if (received.length === 2) done.resolve(undefined)
}, error => done.reject(error))
const peer = new JsonChannel(b, 1000, () => {}, error => done.reject(error))
}, (error) => { done.reject(error) })
const peer = new JsonChannel(b, 1000, () => {}, (error) => { done.reject(error) })
onTestFinished(() => { channel.close(); peer.close() })
await Promise.all([peer.send({ n: 1 }), peer.send({ n: 2 })])
await done.promise
@@ -51,8 +51,8 @@ describe('bounded process frames', () => {
it.each([0, 65])('rejects a declared %i-byte frame before accepting its body', async (length) => {
const { a, b } = pair()
const failure = Promise.withResolvers<Error>()
const channel = new JsonChannel(a, 64, () => { throw new Error('must not dispatch') }, error => failure.resolve(error))
onTestFinished(() => channel.close())
const channel = new JsonChannel(a, 64, () => { throw new Error('must not dispatch') }, (error) => { failure.resolve(error) })
onTestFinished(() => { channel.close() })
const header = Buffer.alloc(4)
header.writeUInt32BE(length)
b.write(header)
@@ -63,7 +63,7 @@ describe('bounded process frames', () => {
for (const payload of [Buffer.from('{'), Buffer.from([0xff]), Buffer.from('{}')]) {
const { a, b } = pair()
const failure = Promise.withResolvers<Error>()
const channel = new JsonChannel(a, 64, () => { throw new Error('receiver failed') }, error => failure.resolve(error))
const channel = new JsonChannel(a, 64, () => { throw new Error('receiver failed') }, (error) => { failure.resolve(error) })
const header = Buffer.alloc(4)
header.writeUInt32BE(payload.length)
b.write(Buffer.concat([header, payload]))
@@ -84,8 +84,8 @@ describe('bounded process frames', () => {
it('reports peer EOF while a program is active', async () => {
const { a, b } = pair()
const failure = Promise.withResolvers<Error>()
const channel = new JsonChannel(a, 64, () => {}, error => failure.resolve(error))
onTestFinished(() => channel.close())
const channel = new JsonChannel(a, 64, () => {}, (error) => { failure.resolve(error) })
onTestFinished(() => { channel.close() })
b.end()
expect((await failure.promise).message).toContain('ended')
})
@@ -117,7 +117,7 @@ it('ignores callbacks already captured by an emission when an earlier listener c
const { a } = pair()
let failures = 0
const channel = new JsonChannel(a, 64, () => { throw new Error('closed channel dispatched') }, () => { failures += 1 })
a.prependOnceListener(event, () => channel.close())
a.prependOnceListener(event, () => { channel.close() })
a.emit(event, frame({}))
a.emit('error', new Error('late closed stream error'))
expect(failures).toBe(0)
@@ -38,6 +38,12 @@ async function setup(config: Config = {}, mode: 'read-only' | 'danger-full-acces
const stderr = new PassThrough()
const direct = Promise.withResolvers<SubprocessOutcome>()
const messages: unknown[] = []
const terminate = vi.fn(() => {
stdout.end()
stderr.end()
direct.resolve({ exitCode: 0, signal: null })
})
const waitForExit = vi.fn(async () => true)
const handle: SubprocessHandle = {
stdin: undefined,
stdout,
@@ -45,12 +51,8 @@ async function setup(config: Config = {}, mode: 'read-only' | 'danger-full-acces
control,
collected: {},
done: direct.promise,
terminate: vi.fn(() => {
stdout.end()
stderr.end()
direct.resolve({ exitCode: 0, signal: null })
}),
waitForExit: vi.fn(async () => true),
terminate,
waitForExit,
}
const writes = new Set<Promise<void>>()
let receive: (message: unknown) => void = () => {}
@@ -83,7 +85,8 @@ async function setup(config: Config = {}, mode: 'read-only' | 'danger-full-acces
}
}
return {
ctx, runtime, handle, direct, stdout, stderr, control, peer, messages, spawn, resolveExecutable, emit, start, onBoot,
ctx, runtime, handle, terminate, waitForExit, direct, stdout, stderr, control, peer, messages,
spawn, resolveExecutable, emit, start, onBoot,
receive: (callback: typeof receive) => { receive = callback },
}
}
@@ -179,8 +182,8 @@ describe('Node runtime host failures', () => {
const h = await setup()
h.spawn.mockReturnValue({ ...h.handle, control: undefined })
expect((await h.start()).error?.message).toContain('did not supply the requested control')
expect(h.handle.terminate).toHaveBeenCalledOnce()
expect(h.handle.waitForExit).toHaveBeenCalledOnce()
expect(h.terminate).toHaveBeenCalledOnce()
expect(h.waitForExit).toHaveBeenCalledOnce()
})
it('reports an executable lookup failure without allocating a process', async () => {
@@ -269,7 +272,7 @@ describe('Node runtime host failures', () => {
const binding = vi.fn(async () => null)
h.onBoot(() => { h.emit(frame) })
const result = await h.start(withBinding(binding))
expect(result.error).toEqual({ kind: 'protocol', message: expect.stringContaining(message) })
expect(result.error).toEqual({ kind: 'protocol', message: expect.stringContaining(message) as unknown })
expect(binding).not.toHaveBeenCalled()
})
@@ -307,7 +310,7 @@ describe('Node runtime host failures', () => {
if (message.type === 'reply') h.emit({ type: 'done' })
})
expect((await h.start(withBinding(async () => Number.NaN))).error).toBeUndefined()
expect(h.messages).toContainEqual({ type: 'reply', id: 1, ok: false, message: expect.stringContaining('lossless JSON') })
expect(h.messages).toContainEqual({ type: 'reply', id: 1, ok: false, message: expect.stringContaining('lossless JSON') as unknown })
})
it('does not publish a late binding reply after the program has settled', async () => {
@@ -324,7 +327,7 @@ describe('Node runtime host failures', () => {
it('reports failed managed cleanup even when the program returns successfully', async () => {
const h = await setup()
vi.mocked(h.handle.waitForExit).mockRejectedValue(new Error('range cannot be observed'))
vi.mocked(h.waitForExit).mockRejectedValue(new Error('range cannot be observed'))
h.onBoot(() => { h.emit({ type: 'done', value: encodeCodeJsonWire(42) }) })
expect((await h.start()).error).toEqual({ kind: 'worker-exit', message: 'managed process cleanup failed: range cannot be observed' })
})
@@ -351,4 +354,93 @@ describe('Node runtime host failures', () => {
expect(result.error?.kind).toBe('output-limit')
expect(result.logs).toEqual(['retained'])
})
it.each([undefined, { kind: 'exception', message: 'program failed' }] as const)('bounds incomplete raw output after completion with %j', async (failure) => {
const h = await setup({ graceMs: 5 })
const cleaning = Promise.withResolvers<undefined>()
vi.mocked(h.terminate).mockImplementation(() => { h.direct.resolve({ exitCode: 0, signal: null }) })
vi.mocked(h.waitForExit).mockImplementation(async () => { cleaning.resolve(undefined); return true })
h.onBoot(() => { h.emit(failure === undefined ? { type: 'done' } : { type: 'done', error: failure }) })
vi.useFakeTimers({ toFake: ['setTimeout', 'clearTimeout'] })
try {
const pending = h.start()
await cleaning.promise
await vi.advanceTimersByTimeAsync(5)
expect((await pending).error).toEqual(failure ?? { kind: 'worker-exit', message: 'Node process output did not close cleanly' })
expect(h.stdout.destroyed).toBe(true)
expect(h.stderr.destroyed).toBe(true)
} finally { vi.useRealTimers() }
})
it('decodes fragmented native output and preserves incomplete final UTF-8 bytes', async () => {
const h = await setup()
h.onBoot(() => {
for (const stream of [h.stdout, h.stderr]) {
stream.write(Buffer.from([0xe2]))
stream.write(Buffer.from([0x82, 0xac]))
stream.end(Buffer.from([0xe2]))
}
h.emit({ type: 'done' })
})
const result = await h.start()
expect(result.error).toBeUndefined()
expect(result.logs.filter(text => text === '€')).toHaveLength(2)
expect(result.logs.filter(text => text === '�')).toHaveLength(2)
})
it('keeps the fitting native-output prefix and ignores later overflow chunks', async () => {
const h = await setup({ maxOutputBytes: 256 })
vi.mocked(h.terminate).mockImplementation(() => {
h.stderr.end('later output')
h.stdout.end()
h.direct.resolve({ exitCode: 0, signal: null })
})
h.onBoot(() => { h.stdout.write('x'.repeat(1024)) })
const result = await h.start()
expect(result.error?.kind).toBe('output-limit')
expect(result.logs.join('')).toMatch(/^x+$/)
expect(Buffer.byteLength(JSON.stringify(result.logs))).toBeLessThanOrEqual(256)
})
it.each(['stdout', 'stderr'] as const)('reports a broken raw %s pipe', async (name) => {
const h = await setup()
h.onBoot(() => { h[name].emit('error', new Error(`${name} closed`)) })
expect((await h.start()).error).toEqual({ kind: 'worker-exit', message: `${name} closed` })
})
it('retains stderr when a ready process exits without a completion frame', async () => {
const h = await setup()
h.onBoot(() => {
h.stderr.write('native fatal detail')
h.direct.resolve({ exitCode: 9, signal: null })
})
expect((await h.start()).error).toEqual({ kind: 'worker-exit', message: 'Node process exited before completing (9): native fatal detail' })
})
it.each([true, false])('attributes a failed confined spawn only with runner evidence (%s)', async (runnerFailed) => {
const h = await setup({}, 'read-only')
const runner = '/sandbox-runner'
vi.spyOn(h.ctx.sandbox, 'confine').mockImplementation(argv => confinement([runner, ...argv]))
h.onBoot(() => {
h.direct.reject(Object.assign(new Error('spawn rejected'), runnerFailed ? { code: 'ENOENT', path: runner, syscall: `spawn ${runner}` } : {}))
})
expect((await h.start()).error).toEqual({ kind: runnerFailed ? 'sandbox-unavailable' : 'worker-exit', message: 'spawn rejected' })
})
it('selects the private packaged bootstrap without leaking ambient environment', async () => {
const h = await setup()
h.onBoot(() => { h.emit({ type: 'done' }) })
const prior = Object.getOwnPropertyDescriptor(process, 'pkg')
try {
Object.defineProperty(process, 'pkg', { configurable: true, value: {} })
expect((await h.start()).error).toBeUndefined()
const spec = h.spawn.mock.calls[0]?.[0]
expect(spec?.env?.DSH_CODE_RUNTIME_NODE).toBe('1')
expect(spec?.argv.at(-1)).toBe('134217728')
expect(Object.entries(spec?.env ?? {}).filter(([, value]) => value !== undefined)).toEqual([['DSH_CODE_RUNTIME_NODE', '1']])
} finally {
if (prior === undefined) Reflect.deleteProperty(process, 'pkg')
else Object.defineProperty(process, 'pkg', prior)
}
})
})
@@ -27,7 +27,7 @@ it('clears process environment, dispatches a binding reply and flushes the termi
if (message.type === 'ready') void peer.send({ type: 'boot', data: { code: 'console.log("ready"); return await tools.echo({});', namespaces: [{ global: 'tools', names: ['echo'] }], maxOutputBytes: 1024 } })
if (message.type === 'call') void peer.send({ type: 'reply', id: message.id, ok: true, value: encodeCodeJsonWire(42) })
}, () => {})
onTestFinished(() => peer.close())
onTestFinished(() => { peer.close() })
await runNodeMain(child, 4096, state)
expect(state.env).toEqual({})
expect(state.exitCode).toBeUndefined()
@@ -42,7 +42,7 @@ it.each([0, -1, 1.5, 4294967296])('rejects an invalid bootstrap frame limit %i',
it('rejects an unexpected first control frame', async () => {
const { child, host } = endpoints()
const peer = new JsonChannel(host, 4096, () => { void peer.send({ type: 'reply' }) }, () => {})
onTestFinished(() => peer.close())
onTestFinished(() => { peer.close() })
await expect(runNodeMain(child, 4096, processState())).rejects.toThrow('expected program boot')
})
@@ -60,7 +60,7 @@ it('contains program writes that exceed queued control output', async () => {
const peer = new JsonChannel(host, 1024, (raw) => {
if ((raw as { type: string }).type === 'ready') void peer.send({ type: 'boot', data: { code: 'for(let i=0;i<30;i++) console.log("x".repeat(100));', namespaces: [], maxOutputBytes: 8000 } })
}, () => {})
onTestFinished(() => peer.close())
onTestFinished(() => { peer.close() })
await runNodeMain(child, 1024, state)
expect(state.exitCode).toBe(1)
})
@@ -17,13 +17,13 @@ it('boots an unbuilt source closure outside the workspace and exchanges tool rep
const source = `import {Socket} from 'node:net';import {runNodeMain} from ${JSON.stringify(pathToFileURL(join(directory, 'process.ts')).href)};await runNodeMain(new Socket({fd:7,readable:true,writable:true}),100000,process);`
const child = spawn(process.execPath, ['--input-type=module', '--eval', source], {
env: { PLACEHOLDER_SECRET: 'fixture-only' },
stdio: ['ignore', 'pipe', 'pipe', 'ignore', 'ignore', 'ignore', 'ignore', 'pipe'],
stdio: ['ignore', 'pipe', 'pipe', 'ignore', 'ignore', 'ignore', 'ignore', 'overlapped'],
})
let stderr = ''
child.stderr?.on('data', (chunk: Buffer) => { stderr += chunk.toString('utf8') })
const finished = new Promise<void>((resolve) => { child.once('close', () => { resolve() }) })
const completed = Promise.withResolvers<unknown>()
child.once('error', error => completed.reject(error))
child.once('error', (error) => { completed.reject(error) })
child.once('exit', (code) => { if (code !== 0) completed.reject(new Error(`child exit ${code}: ${stderr}`)) })
const control = Array.from(child.stdio)[7]
if (!(control instanceof Duplex)) { child.kill(); await finished; throw new Error('missing child control channel') }
@@ -34,15 +34,15 @@ it('boots an unbuilt source closure outside the workspace and exchanges tool rep
code: 'const answer = await tools.echo({ n: 21 }); return { answer, env: { ...process.env } }',
namespaces: [{ global: 'tools', names: ['echo'] }],
maxOutputBytes: 10_000,
} }).catch(error => completed.reject(error))
} }).catch((error: unknown) => { completed.reject(error) })
} else if (message.type === 'call') {
expect(decodeCodeJsonWire(message.args)).toEqual({ n: 21 })
void channel.send({ type: 'reply', id: message.id, ok: true, value: encodeCodeJsonWire(42) }).catch(error => completed.reject(error))
void channel.send({ type: 'reply', id: message.id, ok: true, value: encodeCodeJsonWire(42) }).catch((error: unknown) => { completed.reject(error) })
} else if (message.type === 'done') {
if (message.error !== undefined) completed.reject(new Error(JSON.stringify(message.error)))
else completed.resolve(decodeCodeJsonWire(message.value))
}
}, error => completed.reject(error))
}, (error) => { completed.reject(error) })
onTestFinished(async () => { channel.close(); child.kill(); await finished })
expect(await completed.promise).toEqual({ answer: 42, env: {} })
await finished
@@ -1,5 +1,6 @@
import { mkdtemp, mkdir, readFile, rm, symlink } from 'node:fs/promises'
import { homedir } from 'node:os'
import { createServer, type Socket } from 'node:net'
import { join } from 'node:path'
import { Context } from '@deepseek-ai/cordis'
import { describe, expect, it, onTestFinished } from 'vitest'
@@ -98,6 +99,52 @@ describe('Node program process', () => {
expect((await active).error).toEqual({ kind: 'abort', message: 'stop' })
})
it('closes a still-running descendant before returning the program result', async () => {
const { run } = await setup()
const connected = Promise.withResolvers<undefined>()
const disconnected = Promise.withResolvers<undefined>()
let peer: Socket | undefined
const server = createServer((socket) => {
peer = socket
// A terminated peer can reset its connection instead of sending FIN.
socket.on('error', () => {})
socket.once('close', () => { disconnected.resolve(undefined) })
connected.resolve(undefined)
})
onTestFinished(async () => {
peer?.destroy()
if (server.listening) await new Promise<void>((resolve, reject) => {
server.close((error) => { if (error) reject(error); else resolve() })
})
})
await new Promise<void>((resolve, reject) => {
server.once('error', reject)
server.listen(0, '127.0.0.1', resolve)
})
const address = server.address()
if (address === null || typeof address === 'string') throw new Error('expected bound TCP listener')
const child = `require('node:net').connect(${address.port},'127.0.0.1')`
const result = await run({
program: `const {spawn}=await import("node:child_process"); spawn(process.execPath,["-e",${JSON.stringify(child)}],{stdio:"ignore"}); await tools.connected({}); return 42;`,
bindings: bindings({ connected: async () => { await connected.promise; return null } }),
})
expect(result.error).toBeUndefined()
expect(result.value).toBe(42)
await disconnected.promise
})
it('applies the configured V8 old-generation ceiling to each fresh Node process', async () => {
const limits: number[] = []
for (const maxOldGenerationSizeMb of [32, 64]) {
const { run } = await setup({ maxOldGenerationSizeMb })
const result = await run({ program: 'return (await import("node:v8")).getHeapStatistics().heap_size_limit;', bindings: [] })
expect(result.error).toBeUndefined()
if (typeof result.value !== 'number') throw new Error('expected V8 heap limit')
limits.push(result.value)
}
expect(Number(limits[1]) - Number(limits[0])).toBe(32 * 1024 * 1024)
})
it('disposes active programs and rejects later execution', async () => {
const { ctx, run, runtime } = await setup()
const spec = runtime.resolve({ program: '', bindings: [] })