diff --git a/packages/subprocess/subprocess-local/src/windows-job.ts b/packages/subprocess/subprocess-local/src/windows-job.ts index e680e96a5b..6031c91748 100644 --- a/packages/subprocess/subprocess-local/src/windows-job.ts +++ b/packages/subprocess/subprocess-local/src/windows-job.ts @@ -18,22 +18,21 @@ import { createWindowsStdioBridge } from './windows-stdio.ts' function observeCollectedStream( mode: SubprocessSpawnSpec['stdio']['stdout'], - stream: Readable | null | undefined, + stream: Readable | null, ): Promise { - if (mode === 'pipe' || mode === 'inherit' || stream === null || stream === undefined - || stream.readableEnded || stream.destroyed) { - return Promise.resolve() - } + if (mode === 'pipe' || mode === 'inherit') return Promise.resolve() + // The bridge creates collect streams synchronously before the runner starts. + const collected = stream as Readable return new Promise((resolve) => { const settle = (): void => { - stream.off('end', settle) - stream.off('close', settle) - stream.off('error', settle) + collected.off('end', settle) + collected.off('close', settle) + collected.off('error', settle) resolve() } - stream.once('end', settle) - stream.once('close', settle) - stream.once('error', settle) + collected.once('end', settle) + collected.once('close', settle) + collected.once('error', settle) }) } @@ -73,7 +72,7 @@ class WindowsJobOwner implements BoundProcessOwner { this.observation = new Promise((resolve, reject) => { runner.once('close', (exitCode, signal) => { this.runnerClosed = true - if (exitCode === 0 && signal === null) { + if (this.startupFailureReported || this.runner.pid === undefined || (exitCode === 0 && signal === null)) { this.stopped = true resolve() return @@ -92,7 +91,7 @@ class WindowsJobOwner implements BoundProcessOwner { } signal(_signal: NodeJS.Signals): void { - if (this.stopped || this.runnerClosed || this.startupFailureReported) return + if (this.stopped || this.runnerClosed || this.startupFailureReported || this.runner.pid === undefined) return try { if (this.runner.connected) { this.runner.send({ type: 'terminate' }, (error) => { diff --git a/packages/subprocess/subprocess-local/tests/windows-job.spec.ts b/packages/subprocess/subprocess-local/tests/windows-job.spec.ts index b8d24b8876..8d9005da18 100644 --- a/packages/subprocess/subprocess-local/tests/windows-job.spec.ts +++ b/packages/subprocess/subprocess-local/tests/windows-job.spec.ts @@ -1,11 +1,13 @@ import { spawn, spawnSync } from 'node:child_process' import type { ChildProcess } from 'node:child_process' import { EventEmitter } from 'node:events' +import { PassThrough } from 'node:stream' import { fileURLToPath } from 'node:url' import { describe, expect, it, vi } from 'vitest' import type { SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess' import { appendRunnerEvent } from '../src/runner-protocol.ts' import { launchWindowsJob, probeWindowsJob } from '../src/windows-job.ts' +import type { WindowsStdioBridge } from '../src/windows-stdio.ts' const fixture = fileURLToPath(new URL('fixtures/fake-job-runner.ts', import.meta.url)) const invocation = [process.execPath, '--import', 'tsx/esm', fixture] @@ -110,6 +112,20 @@ describe('Windows Job runner adapter', () => { await expect(launch.owner.waitForExit()).resolves.toBe(true) }) + it('treats a wrapper that never started as an empty managed range', async () => { + const child = new EventEmitter() as ChildProcess + const kill = vi.fn(() => true) + Object.assign(child, { pid: undefined, connected: false, kill }) + const run = vi.fn(() => child) as unknown as typeof spawn + const launch = launchWindowsJob(spec(['fake-target']), { spawn: run, runnerInvocation: ['missing-runner'] }) + + await expect(launch.direct).rejects.toThrow('runner failed to start') + launch.owner.signal('SIGTERM') + expect(kill).not.toHaveBeenCalled() + child.emit('close', -2, null) + await expect(launch.owner.waitForExit()).resolves.toBe(true) + }) + it('falls back to killing the runner when IPC delivery is unavailable or fails', async () => { for (const mode of ['callback-error', 'disconnected', 'throw'] as const) { const child = new EventEmitter() as ChildProcess @@ -212,4 +228,78 @@ describe('Windows Job runner adapter', () => { vi.resetModules() } }) + + it('cleans synchronous setup failures and waits for collected streams', async () => { + const bridgeFailure = new Error('bridge failed') + const spawnFailure = new Error('spawn threw') + const bridges: Array; closeInput: ReturnType }> = [] + let collectedStreams: { stdout: PassThrough; stderr: PassThrough } | undefined + vi.resetModules() + vi.doMock('../src/windows-stdio.ts', async importOriginal => ({ + ...await importOriginal(), + createWindowsStdioBridge: vi.fn((request: SubprocessSpawnSpec): WindowsStdioBridge => { + if (request.argv[0] === 'bridge-failure') throw bridgeFailure + const stdout = typeof request.stdio.stdout === 'object' ? new PassThrough() : null + const stderr = typeof request.stdio.stderr === 'object' ? new PassThrough() : null + if (stdout !== null && stderr !== null) collectedStreams = { stdout, stderr } + const bridge = { + stdin: null, + stdout, + stderr, + runnerArgs: [], + runnerStdio: ['ignore', 'ignore', 'ignore', 'ipc'], + closeInput: vi.fn(), + dispose: vi.fn(() => { + stdout?.destroy() + stderr?.destroy() + }), + } satisfies WindowsStdioBridge + bridges.push(bridge) + return bridge + }), + })) + try { + const isolated = await import('../src/windows-job.ts') + expect(() => isolated.launchWindowsJob(spec(['bridge-failure']), { runnerInvocation: ['fake-runner'] })) + .toThrow(bridgeFailure) + + expect(() => isolated.launchWindowsJob(spec(['spawn-failure']), { + spawn: vi.fn(() => { throw spawnFailure }) as unknown as typeof spawn, + runnerInvocation: ['fake-runner'], + })).toThrow(spawnFailure) + expect(bridges.at(-1)?.dispose).toHaveBeenCalledOnce() + + const child = new EventEmitter() as ChildProcess + Object.assign(child, { pid: 432, connected: true, kill: vi.fn(), send: vi.fn() }) + const run = vi.fn((_command: string, args: readonly string[]) => { + const eventsPath = args[args.indexOf('--events') + 1] as string + appendRunnerEvent(eventsPath, { type: 'started', pid: 432 }) + appendRunnerEvent(eventsPath, { type: 'exit', exitCode: 0, signal: null }) + setImmediate(() => { child.emit('close', 0, null) }) + return child + }) as unknown as typeof spawn + const launch = isolated.launchWindowsJob({ + ...spec(['collect']), + stdio: { + stdin: 'ignore', + stdout: { maxBytes: 1024 }, + stderr: { maxBytes: 1024 }, + }, + }, { spawn: run, runnerInvocation: ['fake-runner'] }) + let streamsSettled = false + void launch.closed.then(() => { streamsSettled = true }) + await launch.direct + await new Promise(resolve => setImmediate(resolve)) + expect(streamsSettled).toBe(false) + collectedStreams?.stdout.emit('end') + await Promise.resolve() + expect(streamsSettled).toBe(false) + collectedStreams?.stderr.emit('error', new Error('stream closed')) + await expect(launch.closed).resolves.toBeUndefined() + expect(bridges.at(-1)?.closeInput).toHaveBeenCalledOnce() + } finally { + vi.doUnmock('../src/windows-stdio.ts') + vi.resetModules() + } + }) })