fix(code-runtime): flatten worker JSON transport
This commit is contained in:
@@ -3,6 +3,7 @@ import { EventEmitter } from 'node:events'
|
||||
import { LogBuffer, makeConsoleShim, makeNamespaces, captureStreamWrites, prepareCompletion, runWorkerMain, ToolCallError, truncateUtf8Bytes, wireReplies } from '../src/bootstrap.ts'
|
||||
import type { BootstrapPort, PatchableStream, PendingCall } from '../src/bootstrap.ts'
|
||||
import type { ReplyMessage, WorkerToHost } from '../src/protocol.ts'
|
||||
import { decodeWorkerJson, encodeWorkerJson } from '../src/worker-json.ts'
|
||||
|
||||
/**
|
||||
* An in-process stand-in for the worker's parentPort: the test plays the
|
||||
@@ -37,6 +38,11 @@ class FakePort implements BootstrapPort {
|
||||
done(): WorkerToHost | undefined {
|
||||
return this.sent.find(message => message.type === 'done')
|
||||
}
|
||||
|
||||
doneValue(): unknown {
|
||||
const done = this.done()
|
||||
return done?.type === 'done' && done.value !== undefined ? decodeWorkerJson(done.value) : undefined
|
||||
}
|
||||
}
|
||||
|
||||
function fakeStreams(): { stdout: PatchableStream; stderr: PatchableStream } {
|
||||
@@ -127,7 +133,7 @@ describe('captureStreamWrites', () => {
|
||||
describe('prepareCompletion', () => {
|
||||
it('omits undefined and passes lossless JSON values exactly', () => {
|
||||
expect(prepareCompletion(undefined, 100)).toEqual({})
|
||||
expect(prepareCompletion({ a: [1, 'two'] }, 100)).toEqual({ value: { a: [1, 'two'] } })
|
||||
expect(prepareCompletion({ a: [1, 'two'] }, 100)).toEqual({ value: encodeWorkerJson({ a: [1, 'two'] }) })
|
||||
})
|
||||
|
||||
it('turns every lossy completion shape into invalid-output', () => {
|
||||
@@ -149,7 +155,7 @@ describe('prepareCompletion', () => {
|
||||
})
|
||||
|
||||
it('measures the exact JSON serialization at and over the boundary', () => {
|
||||
expect(prepareCompletion('€', 5)).toEqual({ value: '€' })
|
||||
expect(prepareCompletion('€', 5)).toEqual({ value: encodeWorkerJson('€') })
|
||||
expect(prepareCompletion('€', 4)).toEqual({
|
||||
error: { kind: 'output-limit', message: 'outer output exceeded 4 bytes' },
|
||||
})
|
||||
@@ -178,9 +184,20 @@ describe('truncateUtf8Bytes', () => {
|
||||
})
|
||||
|
||||
describe('makeNamespaces', () => {
|
||||
it('rejects a malformed success reply instead of resolving a lossy binding value', async () => {
|
||||
const port = new FakePort()
|
||||
const pending = new Map<number, PendingCall>()
|
||||
wireReplies(port, pending)
|
||||
const result = new Promise<unknown>((resolve, reject) => { pending.set(1, { resolve, reject }) })
|
||||
port.deliver({ type: 'reply', id: 1, ok: true, value: [undefined] as never })
|
||||
await expect(result).rejects.toThrow('binding resolution must be lossless JSON')
|
||||
})
|
||||
|
||||
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
|
||||
port.respond = message => message.type === 'call'
|
||||
? { type: 'reply', id: message.id, ok: true, value: encodeWorkerJson(`${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>>]
|
||||
@@ -268,14 +285,18 @@ describe('makeNamespaces', () => {
|
||||
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
|
||||
port.respond = (message) => {
|
||||
if (message.type !== 'call') return undefined
|
||||
const args = decodeWorkerJson(message.args) as { n: number }
|
||||
return { type: 'reply', id: message.id, ok: true, value: encodeWorkerJson(args.n * 2) }
|
||||
}
|
||||
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(['got 42'])
|
||||
expect(port.done()).toEqual({ type: 'done', value: { doubled: 42 } })
|
||||
expect(port.doneValue()).toEqual({ doubled: 42 })
|
||||
})
|
||||
|
||||
it('reports worker-side log capture overflow before completing', async () => {
|
||||
@@ -287,7 +308,7 @@ describe('runWorkerMain', () => {
|
||||
}, fakeStreams())
|
||||
expect(port.sent).toContainEqual({ type: 'log', text: '1234' })
|
||||
expect(port.sent).toContainEqual({ type: 'output-limit' })
|
||||
expect(port.done()).toEqual({ type: 'done', value: null })
|
||||
expect(port.doneValue()).toBeNull()
|
||||
})
|
||||
|
||||
it('reports a thrown program error on the done message', async () => {
|
||||
@@ -318,10 +339,7 @@ describe('runWorkerMain', () => {
|
||||
code: 'try { await tools.x({}) } catch (error) { return { caught: error instanceof ToolCallError, name: error.name, toolName: error.toolName, message: error.message } }',
|
||||
namespaces: [{ global: 'tools', names: ['x'] }],
|
||||
}, fakeStreams())
|
||||
expect(port.done()).toEqual({
|
||||
type: 'done',
|
||||
value: { caught: true, name: 'ToolCallError', toolName: 'x', message: 'denied by host' },
|
||||
})
|
||||
expect(port.doneValue()).toEqual({ caught: true, name: 'ToolCallError', toolName: 'x', message: 'denied by host' })
|
||||
expect(new ToolCallError('x', 'nope')).toMatchObject({ name: 'ToolCallError', toolName: 'x', message: 'nope' })
|
||||
})
|
||||
|
||||
@@ -330,15 +348,15 @@ describe('runWorkerMain', () => {
|
||||
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' }
|
||||
port.deliver({ type: 'reply', id: 9_999, ok: true, value: encodeWorkerJson('stray') })
|
||||
return { type: 'reply', id: message.id, ok: true, value: encodeWorkerJson('real') }
|
||||
}
|
||||
await runWorkerMain(port, {
|
||||
...BOOT,
|
||||
code: 'return await tools.x({})',
|
||||
namespaces: [{ global: 'tools', names: ['x'] }],
|
||||
}, fakeStreams())
|
||||
expect(port.done()).toEqual({ type: 'done', value: 'real' })
|
||||
expect(port.doneValue()).toBe('real')
|
||||
})
|
||||
|
||||
it('captures raw stream writes through the patched process streams', async () => {
|
||||
|
||||
@@ -92,7 +92,7 @@ describe('WorkerCodeRuntime — programs and bindings (real workers)', () => {
|
||||
cursor = Array.isArray(cursor) ? cursor[0] : undefined
|
||||
}
|
||||
expect(cursor).toBe('leaf')
|
||||
}, 60_000)
|
||||
}, 15_000)
|
||||
|
||||
it('reports non-erasable syntax as an exception without spawning a worker', async () => {
|
||||
const { runtime } = await setup()
|
||||
@@ -371,7 +371,7 @@ describe('WorkerCodeRuntime — budgets and containment (real workers)', () => {
|
||||
const { parentPort } = await import('node:worker_threads');
|
||||
const write = (text) => Object.getPrototypeOf(process.stdout).write.call(process.stdout, text);
|
||||
write('late-pipe-' + 'x'.repeat(100_000));
|
||||
parentPort.postMessage({ type: 'done', value: 'done' });
|
||||
parentPort.postMessage({ type: 'done', value: ['done'] });
|
||||
for (;;) {}
|
||||
`,
|
||||
bindings: [],
|
||||
@@ -438,7 +438,7 @@ describe('WorkerCodeRuntime — hostile programs (real workers)', () => {
|
||||
program: `
|
||||
const { parentPort } = await import('node:worker_threads');
|
||||
for (let i = 0; i < 50; i++) parentPort.postMessage({ type: 'log', text: 'F'.repeat(100), forged: true });
|
||||
parentPort.postMessage({ type: 'done', value: 'V'.repeat(100000) });
|
||||
parentPort.postMessage({ type: 'done', value: ['V'.repeat(100000)] });
|
||||
for (;;) {}
|
||||
`,
|
||||
bindings: [],
|
||||
@@ -482,8 +482,9 @@ describe('WorkerCodeRuntime — hostile programs (real workers)', () => {
|
||||
const result = await runtime.run({
|
||||
program: `
|
||||
const { parentPort } = await import('node:worker_threads');
|
||||
let value = null;
|
||||
for (let depth = 0; depth < 3_000; depth++) value = [value];
|
||||
const value = [];
|
||||
for (let depth = 0; depth < 3_000; depth++) value.push({ kind: 'array', length: 1 });
|
||||
value.push(null);
|
||||
setTimeout(() => { parentPort.postMessage({ type: 'done', value }) }, 25);
|
||||
// Prevent bootstrap's normal undefined completion from racing the forged terminal.
|
||||
await new Promise(() => {});
|
||||
@@ -500,7 +501,7 @@ describe('WorkerCodeRuntime — hostile programs (real workers)', () => {
|
||||
}
|
||||
expect(depth).toBe(3_000)
|
||||
expect(value).toBeNull()
|
||||
}, 60_000)
|
||||
}, 15_000)
|
||||
|
||||
it('turns forged over-limit error text into output-limit at the host', async () => {
|
||||
const { runtime } = await setup({ maxOutputBytes: 64 })
|
||||
|
||||
@@ -3,6 +3,7 @@ import { tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { Worker } from 'node:worker_threads'
|
||||
import { expect, it } from 'vitest'
|
||||
import { decodeWorkerJson } from '../src/worker-json.ts'
|
||||
|
||||
/**
|
||||
* Prove the unbuilt worker is a self-contained source closure. Copying it out
|
||||
@@ -28,7 +29,9 @@ it('boots the source worker without workspace package outputs', async () => {
|
||||
worker?.once('error', reject)
|
||||
})
|
||||
|
||||
expect(message).toEqual({ type: 'done', value: { answer: 42 } })
|
||||
expect(message).toMatchObject({ type: 'done' })
|
||||
const value = typeof message === 'object' && message !== null ? (message as { value?: unknown }).value : undefined
|
||||
expect(decodeWorkerJson(value)).toEqual({ answer: 42 })
|
||||
} finally {
|
||||
if (worker) await worker.terminate()
|
||||
await rm(directory, { recursive: true, force: true })
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
import { runInNewContext } from 'node:vm'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import { snapshotJsonValue } from '@deepseek-ai/dsh-session'
|
||||
import { snapshotCodeJsonValue } from '../src/worker-json.ts'
|
||||
import { decodeWorkerJson, encodeWorkerJson, snapshotCodeJsonValue } from '../src/worker-json.ts'
|
||||
|
||||
describe('snapshotCodeJsonValue', () => {
|
||||
it('matches the canonical scalar boundary', () => {
|
||||
@@ -142,3 +142,90 @@ describe('snapshotCodeJsonValue', () => {
|
||||
expect(snapshotCodeJsonValue({ after: true })).toEqual({ after: true })
|
||||
})
|
||||
})
|
||||
|
||||
describe('flat worker JSON wire', () => {
|
||||
it('round-trips every JSON root while preserving object keys and container order', () => {
|
||||
const withPrototypeKey = Object.create(null) as Record<string, unknown>
|
||||
withPrototypeKey.__proto__ = { safe: true }
|
||||
const values = [null, false, true, 1.25, 'text', [], {}, [1, { nested: [2] }], withPrototypeKey]
|
||||
for (const value of values) {
|
||||
const snapshot = snapshotCodeJsonValue(value)
|
||||
expect(snapshot).not.toBeUndefined()
|
||||
expect(decodeWorkerJson(encodeWorkerJson(snapshot!))).toEqual(snapshot)
|
||||
}
|
||||
const decoded = decodeWorkerJson(encodeWorkerJson(snapshotCodeJsonValue(withPrototypeKey)!)) as Record<string, unknown>
|
||||
expect(Object.hasOwn(decoded, '__proto__')).toBe(true)
|
||||
expect(decoded.__proto__).toEqual({ safe: true })
|
||||
})
|
||||
|
||||
it('round-trips deep values through a bounded-depth token array', () => {
|
||||
let value: unknown = 'leaf'
|
||||
for (let depth = 0; depth < 5_000; depth++) value = [value]
|
||||
const snapshot = snapshotCodeJsonValue(value)!
|
||||
const wire = encodeWorkerJson(snapshot)
|
||||
expect(wire).toHaveLength(5_001)
|
||||
|
||||
let cursor = decodeWorkerJson(wire)
|
||||
for (let depth = 0; depth < 5_000; depth++) {
|
||||
expect(Array.isArray(cursor)).toBe(true)
|
||||
cursor = Array.isArray(cursor) ? cursor[0] : undefined
|
||||
}
|
||||
expect(cursor).toBe('leaf')
|
||||
})
|
||||
|
||||
it('rejects malformed, incomplete, lossy, sparse, decorated, and throwing wire values', () => {
|
||||
const sparse = new Array(1)
|
||||
const compensatedSparse = new Array(1)
|
||||
Object.defineProperty(compensatedSparse, 'extra', { value: true })
|
||||
const decorated: unknown[] = [null]
|
||||
Object.defineProperty(decorated, 'extra', { value: true })
|
||||
const throwing: unknown[] = []
|
||||
Object.defineProperty(throwing, 0, { enumerable: true, get: () => { throw new Error('wire getter') } })
|
||||
const decoratedKeys: unknown[] = ['x']
|
||||
Object.defineProperty(decoratedKeys, 'extra', { value: true })
|
||||
const foreignMarker: Record<string, unknown> = { kind: 'array', length: 0 }
|
||||
Object.setPrototypeOf(foreignMarker, {})
|
||||
const hiddenMarker = Object.defineProperty({ kind: 'array', length: 0 }, 'hidden', { value: true })
|
||||
|
||||
for (const value of [
|
||||
undefined,
|
||||
null,
|
||||
{},
|
||||
[],
|
||||
sparse,
|
||||
compensatedSparse,
|
||||
decorated,
|
||||
throwing,
|
||||
[undefined],
|
||||
[-0],
|
||||
[Number.NaN],
|
||||
[Number.POSITIVE_INFINITY],
|
||||
[1, 2],
|
||||
[[]],
|
||||
[foreignMarker],
|
||||
[hiddenMarker],
|
||||
[{ kind: 'unknown' }],
|
||||
[{ kind: 'array' }],
|
||||
[{ kind: 'array', length: '1' }],
|
||||
[{ kind: 'array', length: -1 }],
|
||||
[{ kind: 'array', length: Number.MAX_SAFE_INTEGER + 1 }],
|
||||
[{ kind: 'array', length: 1 }],
|
||||
[{ kind: 'array', length: 2 }, { kind: 'array', length: 1 }, null],
|
||||
[{ kind: 'array', length: 0, extra: true }],
|
||||
[{ kind: 'object' }],
|
||||
[{ kind: 'object', keys: 'x' }],
|
||||
[{ kind: 'object', keys: decoratedKeys }],
|
||||
[{ kind: 'object', keys: [1] }],
|
||||
[{ kind: 'object', keys: ['x', 'x'] }, 1, 2],
|
||||
[{ kind: 'object', keys: ['x'] }],
|
||||
[{ kind: 'object', keys: [], extra: true }],
|
||||
]) {
|
||||
expect(decodeWorkerJson(value)).toBeUndefined()
|
||||
}
|
||||
})
|
||||
|
||||
it('rejects invalid values passed through a forged static type', () => {
|
||||
expect(() => encodeWorkerJson([undefined] as never)).toThrow(/sparse JSON array/)
|
||||
expect(() => encodeWorkerJson({ value: undefined } as never)).toThrow(/undefined JSON object property/)
|
||||
})
|
||||
})
|
||||
Reference in New Issue
Block a user