fix(subprocess): settle pre-start Windows runner failures

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<void> {
-  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) => {

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<WindowsStdioBridge & { dispose: ReturnType<typeof vi.fn>; closeInput: ReturnType<typeof vi.fn> }> = []
+    let collectedStreams: { stdout: PassThrough; stderr: PassThrough } | undefined
+    vi.resetModules()
+    vi.doMock('../src/windows-stdio.ts', async importOriginal => ({
+      ...await importOriginal<typeof import('../src/windows-stdio.ts')>(),
+      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()
+    }
+  })
 })