diff --git a/packages/ptc-runtime/ptc-runtime-node/tests/persistent-host.spec.ts b/packages/ptc-runtime/ptc-runtime-node/tests/persistent-host.spec.ts new file mode 100644 index 0000000000..b585e1b6a6 --- /dev/null +++ b/packages/ptc-runtime/ptc-runtime-node/tests/persistent-host.spec.ts @@ -0,0 +1,215 @@ +import { Duplex, PassThrough } from 'node:stream' +import { setImmediate } from 'node:timers/promises' +import { Context } from '@deepseek-ai/cordis' +import { describe, expect, it, onTestFinished, vi } from 'vitest' +import FileSystem from '@deepseek-ai/dsh-fs-local' +import Subprocess from '@deepseek-ai/dsh-subprocess-local' +import type { PtcRunRequest } from '@deepseek-ai/dsh-ptc-runtime' +import type { SubprocessHandle, SubprocessOutcome } from '@deepseek-ai/dsh-subprocess' +import PersistentRuntime, { type Config } from '../src/persistent.ts' +import { JsonChannel } from '../src/channel.ts' +import { encodePtcJsonWire } from '../src/json-wire.ts' + +const request: PtcRunRequest = { program: '1', bindings: [], stateOwner: {} } +const NO_INITIAL_FRAME = Symbol('no initial frame') +async function setup(config: Config = {}) { + const ctx = new Context() + onTestFinished(async () => { await ctx.fiber.dispose() }) + await ctx.plugin(FileSystem) + await ctx.plugin(Subprocess) + await ctx.plugin(PersistentRuntime, config) + const runtime = ctx.ptcRuntime + const control = new Duplex({ + read() {}, + write(chunk: Buffer, _encoding, callback) { + if (childControl.destroyed) { callback(new Error('peer closed')); return } + childControl.push(chunk) + callback() + }, + }) + const childControl = new Duplex({ + read() {}, + write(chunk: Buffer, _encoding, callback) { + if (control.destroyed) { callback(new Error('peer closed')); return } + control.push(chunk) + callback() + }, + }) + const stdout = new PassThrough() + const stderr = new PassThrough() + const direct = Promise.withResolvers() + 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, + stderr, + control, + collected: {}, + done: direct.promise, + terminate, + waitForExit, + } + const writes = new Set>() + let receive: (message: unknown) => void = () => {} + const peer = new JsonChannel(childControl, 1024 * 1024, (message) => { + messages.push(message) + receive(message) + }, () => {}) + const emit = (message: unknown): void => { + const write = peer.send(message).catch(() => {}).finally(() => { writes.delete(write) }) + writes.add(write) + } + const resolveExecutable = vi.spyOn(ctx.subprocess, 'resolveExecutable').mockResolvedValue(process.execPath) + const spawn = vi.spyOn(ctx.subprocess, 'spawn').mockReturnValue(handle) + onTestFinished(async () => { + peer.close() + control.destroy() + stdout.destroy() + stderr.destroy() + direct.resolve({ exitCode: 0, signal: null }) + await Promise.all(writes) + }) + const start = (input: PtcRunRequest = request, first: unknown = { type: 'ready' }) => { + const result = runtime.run(runtime.resolve(input)) + if (first !== NO_INITIAL_FRAME) queueMicrotask(() => { emit(first) }) + return result + } + const onBoot = (callback: () => void): void => { + receive = (message) => { + if (typeof message === 'object' && message !== null && 'type' in message && message.type === 'run') callback() + } + } + return { + ctx, runtime, handle, terminate, waitForExit, direct, stdout, stderr, control, peer, messages, + spawn, resolveExecutable, emit, start, onBoot, + receive: (callback: typeof receive) => { receive = callback }, + } +} + +function frame(payload: unknown, id = 1) { return { id, payload } } +function call(id = 1, args: unknown = encodePtcJsonWire(null)) { return frame({ type: 'call', id, global: 'tools', name: 'test', args }) } +const bound: PtcRunRequest = { ...request, bindings: [{ global: 'tools', functions: { test: async () => 42 } }] } + +describe('persistent PTC host protocol', () => { + it.each([ + null, + { id: 99, payload: {} }, + frame({ type: 'other' }), + frame({ type: 'log', text: 1 }), + frame({ type: 'done', error: { kind: 'other', message: '' } }), + frame({ type: 'done', error: { kind: 'exception', message: 1 } }), + call(2), + call(1, 'not a wire value'), + ])('rejects invalid process traffic %j', async (raw) => { + const h = await setup() + h.onBoot(() => { h.emit(raw) }) + expect((await h.start(bound)).error?.kind).toBe('protocol') + expect(h.terminate).toHaveBeenCalledTimes(1) + expect(h.waitForExit).toHaveBeenCalledTimes(1) + }) + + it('requires the readiness handshake and all requested pipes', async () => { + const h = await setup() + expect((await h.start(request, { type: 'other' })).error?.kind).toBe('protocol') + const missing = await setup() + missing.spawn.mockReturnValue({ ...missing.handle, control: undefined }) + expect((await missing.start(request, NO_INITIAL_FRAME)).error?.message).toContain('did not supply') + }) + + it.each(['exception', 'invalid-output', 'output-limit'])('accepts a terminal %s failure', async (kind) => { + const h = await setup() + h.onBoot(() => { h.emit(frame({ type: 'done', error: { kind, message: 'example' } })) }) + expect((await h.start()).error?.kind).toBe(kind) + }) + + it('rejects lossy terminal values and bounds complete results', async () => { + const h = await setup() + h.onBoot(() => { h.emit(frame({ type: 'done', value: 'bad' })) }) + expect((await h.start()).error?.kind).toBe('invalid-output') + const limited = await setup({ maxOutputBytes: 30 }) + limited.onBoot(() => { limited.emit(frame({ type: 'log', text: 'x'.repeat(100) })) }) + expect((await limited.start()).error?.kind).toBe('output-limit') + }) + + it('returns current binding successes, exceptions and unavailable bindings', async () => { + for (const functions of [{ test: async () => 42 }, { test: async () => { throw new Error('denied') } }, {}]) { + const h = await setup() + h.receive((raw) => { + const message = raw as Record + if (message.type === 'run') h.emit(call()) + else if (message.type === 'reply') h.emit(frame({ type: 'done', value: encodePtcJsonWire(message.ok === true) })) + }) + const result = await h.start({ ...request, bindings: [{ global: 'tools', functions }] }) + expect(result.error).toBeUndefined() + const reply = h.messages.find(raw => (raw as { type: string }).type === 'reply') as { ok: boolean; message?: string } + expect(result.value).toBe(reply.ok) + if (!reply.ok) expect(reply.message).toMatch(/denied|unavailable/) + } + }) + + it('bounds outstanding binding calls and drops replies after a run ends', async () => { + const h = await setup({ maxPendingCalls: 1 }) + const pending = Promise.withResolvers() + h.onBoot(() => { h.emit(call()); h.emit(call(2)) }) + try { + expect((await h.start({ ...request, bindings: [{ global: 'tools', functions: { test: () => pending.promise } }] })).error?.kind).toBe('protocol') + } finally { pending.resolve(null) } + await setImmediate() + expect(h.messages.some(raw => (raw as { type: string }).type === 'reply')).toBe(false) + }) + + it('cleans up after launch failure and cancellation during executable resolution', async () => { + const h = await setup() + h.resolveExecutable.mockRejectedValue(new Error('missing executable')) + expect((await h.start(request, NO_INITIAL_FRAME)).error?.message).toContain('missing executable') + const waiting = await setup() + const ready = Promise.withResolvers() + waiting.resolveExecutable.mockImplementation(() => ready.promise) + const controller = new AbortController() + const result = waiting.start({ ...request, signal: controller.signal }, NO_INITIAL_FRAME) + controller.abort() + ready.resolve(process.execPath) + expect((await result).error?.kind).toBe('abort') + expect(waiting.spawn).not.toHaveBeenCalled() + }) + + it('reports raw pipe output, stream errors and process failures without retaining state', async () => { + for (const action of ['raw', 'stream', 'exit', 'rejection'] as const) { + const h = await setup() + h.onBoot(() => { + if (action === 'raw') h.stdout.write('unattributed') + else if (action === 'stream') h.stderr.emit('error', new Error('pipe failed')) + else if (action === 'exit') h.direct.resolve({ exitCode: 7, signal: null }) + else h.direct.reject(new Error('process failed')) + }) + const result = await h.start() + expect(result.error?.kind).toBe(action === 'raw' ? 'protocol' : 'worker-exit') + expect(result.logs).toEqual([]) + } + }) + + it('reports failed managed cleanup and waits for both exit and pipe drain', async () => { + const h = await setup() + h.waitForExit.mockRejectedValue(new Error('cleanup failed')) + h.onBoot(() => { h.emit(frame({ type: 'output-limit' })) }) + expect((await h.start()).error?.message).toContain('cleanup failed') + }) + + it('rejects invalid resolved inputs and deployment bounds', async () => { + const h = await setup() + await expect(h.runtime.run({ ...request, cwd: '.', timeoutMs: null })).rejects.toThrow('resolved inputs') + await expect(h.runtime.run({ ...request, cwd: process.cwd(), timeoutMs: Infinity })).rejects.toThrow('resolved inputs') + for (const config of [ + { timeoutMs: 0 }, { maxMessageBytes: 0x1_0000_0000 }, { maxOutputBytes: 3 }, { graceMs: 2_147_483_648 }, + { nodeExecutable: '' }, { bootstrapPath: 'relative' }, + ]) { + await expect(setup(config)).rejects.toThrow('persistent PTC') + } + }) +}) diff --git a/packages/ptc-runtime/ptc-runtime-node/tests/persistent-process.spec.ts b/packages/ptc-runtime/ptc-runtime-node/tests/persistent-process.spec.ts index a6a2cfb147..208e2254a6 100644 --- a/packages/ptc-runtime/ptc-runtime-node/tests/persistent-process.spec.ts +++ b/packages/ptc-runtime/ptc-runtime-node/tests/persistent-process.spec.ts @@ -57,7 +57,11 @@ it('keeps an environment while capturing each evaluation and binding replies sep }) it('contains directory failures and output overflow in the active evaluation', async () => { - const badDirectory = await setup(4096, () => { throw new Error('directory missing') }) + const badDirectory = await setup(4096, () => { + const error = new Error('directory missing') + error.stack = 'Error: directory missing' + throw error + }) expect(await badDirectory.run('1')).toMatchObject({ error: { kind: 'exception', message: expect.stringContaining('directory missing') } }) const h = await setup() await h.run('console.log("x".repeat(100))', [], 20)