fix(subprocess): clean managed processes on host exit

packages/subprocess/subprocess-local/tests/fixtures/managed-tree.ts

@@ -0,0 +1,16 @@
+import { spawn } from 'node:child_process'
+import { writeFile } from 'node:fs/promises'
+
+const [statePath] = process.argv.slice(2)
+if (statePath === undefined) throw new Error('usage: managed-tree.ts <state-path>')
+
+process.on('SIGTERM', () => {})
+process.on('SIGHUP', () => {})
+const descendant = spawn(process.execPath, [
+  '-e',
+  'process.on("SIGTERM",()=>{});process.on("SIGHUP",()=>{});setInterval(()=>{},60_000)',
+], { stdio: 'ignore' })
+if (descendant.pid === undefined) throw new Error('managed descendant did not publish a pid')
+
+await writeFile(statePath, JSON.stringify({ root: process.pid, descendant: descendant.pid }))
+setInterval(() => {}, 60_000)

packages/subprocess/subprocess-local/tests/fixtures/process-exit-host.ts

@@ -0,0 +1,79 @@
+import { access, readFile, writeFile } from 'node:fs/promises'
+import { join } from 'node:path'
+import { fileURLToPath } from 'node:url'
+import { Context } from '@deepseek-ai/cordis'
+import LocalSubprocessService from '@deepseek-ai/dsh-subprocess-local'
+
+const [kind, trigger, root] = process.argv.slice(2)
+if ((kind !== 'ordinary' && kind !== 'terminal')
+  || (trigger !== 'direct' && trigger !== 'uncaught-exception'
+    && trigger !== 'unhandled-rejection' && trigger !== 'dispose')
+  || root === undefined) {
+  throw new Error('usage: process-exit-host.ts <ordinary|terminal> <direct|uncaught-exception|unhandled-rejection|dispose> <root>')
+}
+
+const treeState = join(root, 'tree.json')
+const ready = join(root, 'ready')
+const proceed = join(root, 'proceed')
+const managedTree = fileURLToPath(new URL('./managed-tree.ts', import.meta.url))
+
+async function waitForFile(path: string): Promise<void> {
+  for (;;) {
+    try {
+      await access(path)
+      return
+    } catch (_notReady) {
+      await new Promise(resolve => setTimeout(resolve, 10))
+    }
+  }
+}
+
+const listenersBefore = process.listenerCount('exit')
+const ctx = new Context()
+const fiber = await ctx.plugin(LocalSubprocessService)
+const listenersAfterLoad = process.listenerCount('exit')
+if (kind === 'ordinary') {
+  ctx.subprocess.spawn({
+    argv: [process.execPath, managedTree, treeState],
+    cwd: process.cwd(),
+    stdio: {
+      stdin: 'ignore',
+      stdout: { maxBytes: 1024 },
+      stderr: { maxBytes: 1024 },
+    },
+    graceMs: trigger === 'dispose' ? 100 : 30_000,
+  })
+} else {
+  await ctx.subprocess.spawnTerminal({
+    argv: [process.execPath, managedTree, treeState],
+    cwd: process.cwd(),
+    rows: 24,
+    cols: 80,
+    graceMs: 30_000,
+  })
+}
+
+await waitForFile(treeState)
+const published = JSON.parse(await readFile(treeState, 'utf8')) as { root?: unknown; descendant?: unknown }
+if (!Number.isSafeInteger(published.root) || !Number.isSafeInteger(published.descendant)) {
+  throw new Error('managed tree published invalid process ids')
+}
+await writeFile(ready, 'ready')
+await waitForFile(proceed)
+
+if (trigger === 'dispose') {
+  await fiber.dispose()
+  await writeFile(join(root, 'dispose.json'), JSON.stringify({
+    listenersBefore,
+    listenersAfterLoad,
+    listenersAfterDispose: process.listenerCount('exit'),
+  }))
+} else if (trigger === 'direct') {
+  process.exit(23)
+} else if (trigger === 'uncaught-exception') {
+  setImmediate(() => { throw new Error('host-exit-uncaught-exception') })
+  await new Promise(() => {})
+} else {
+  void Promise.reject(new Error('host-exit-unhandled-rejection'))
+  await new Promise(() => {})
+}

packages/subprocess/subprocess-local/tests/local.spec.ts

@@ -21,6 +21,77 @@ function spec(command: string, overrides: Partial<SubprocessSpawnSpec> = {}): Su
 }
 
 describe('LocalSubprocessService', () => {
+  it('keeps the host-exit finalizer active until normal disposal reaches quiescence', async () => {
+    const before = new Set(process.listeners('exit'))
+    const ctx = new Context()
+    const fiber = await ctx.plugin(LocalSubprocessService)
+    const listener = process.listeners('exit').find(candidate => !before.has(candidate))
+    expect(listener).toBeTypeOf('function')
+
+    let finishExit!: () => void
+    const exited = new Promise<void>((resolve) => { finishExit = resolve })
+    const terminate = vi.fn()
+    const terminateForHostExit = vi.fn()
+    const live = (ctx.subprocess as unknown as {
+      live: Set<{
+        done: Promise<{ exitCode: number; signal: null }>
+        terminate(): void
+        terminateForHostExit(): void
+        waitForExit(): Promise<boolean>
+      }>
+    }).live
+    live.add({
+      done: Promise.resolve({ exitCode: 0, signal: null }),
+      terminate,
+      terminateForHostExit,
+      waitForExit: async () => { await exited; return true },
+    })
+
+    let disposed = false
+    const disposing = fiber.dispose().then(() => { disposed = true })
+    await new Promise(resolve => setImmediate(resolve))
+    expect(disposed).toBe(false)
+    expect(live.size).toBe(1)
+    listener?.(0)
+    expect(terminate).toHaveBeenCalledOnce()
+    expect(terminateForHostExit).toHaveBeenCalledOnce()
+
+    finishExit()
+    await disposing
+    expect(live.size).toBe(0)
+    expect(process.listeners('exit')).not.toContain(listener)
+  })
+
+  it('contains each host-exit termination failure and continues with the other targets', async () => {
+    const before = new Set(process.listeners('exit'))
+    const ctx = new Context()
+    const fiber = await ctx.plugin(LocalSubprocessService)
+    const listener = process.listeners('exit').find(candidate => !before.has(candidate))
+    expect(listener).toBeTypeOf('function')
+    const ordinaryFailure = vi.fn(() => { throw new Error('ordinary failed') })
+    const ordinarySuccess = vi.fn()
+    const terminalFailure = vi.fn(() => { throw new Error('terminal failed') })
+    const terminalSuccess = vi.fn()
+    const service = ctx.subprocess as unknown as {
+      live: Set<{ terminateForHostExit(): void }>
+      terminals: Set<{ terminateForHostExit(): void }>
+    }
+    service.live.add({ terminateForHostExit: ordinaryFailure })
+    service.live.add({ terminateForHostExit: ordinarySuccess })
+    service.terminals.add({ terminateForHostExit: terminalFailure })
+    service.terminals.add({ terminateForHostExit: terminalSuccess })
+
+    expect(() => { listener?.(0) }).not.toThrow()
+    expect(ordinaryFailure).toHaveBeenCalledOnce()
+    expect(ordinarySuccess).toHaveBeenCalledOnce()
+    expect(terminalFailure).toHaveBeenCalledOnce()
+    expect(terminalSuccess).toHaveBeenCalledOnce()
+
+    service.live.clear()
+    service.terminals.clear()
+    await fiber.dispose()
+  })
+
   it('resolves absolute and PATH executables and honors lookup cancellation', async () => {
     const ctx = new Context()
     const fiber = await ctx.plugin(LocalSubprocessService)
@@ -177,6 +248,30 @@ describe('LocalSubprocessService', () => {
     expect(disposalErrors).toEqual([failure])
   })
 
+  it('force-terminates remaining targets before releasing a failed disposal', async () => {
+    const before = new Set(process.listeners('exit'))
+    const ctx = new Context()
+    const fiber = await ctx.plugin(LocalSubprocessService)
+    const listener = process.listeners('exit').find(candidate => !before.has(candidate))
+    expect(listener).toBeTypeOf('function')
+    const failure = new Error('cleanup failed')
+    const terminateForHostExit = vi.fn(() => {
+      expect(process.listeners('exit')).toContain(listener)
+    })
+    const terminal = {
+      terminate: vi.fn(async () => { throw failure }),
+      terminateForHostExit,
+    }
+    const terminals = (ctx.subprocess as unknown as { terminals: Set<typeof terminal> }).terminals
+    terminals.add(terminal)
+
+    await fiber.dispose()
+
+    expect(terminateForHostExit).toHaveBeenCalledOnce()
+    expect(terminals.size).toBe(0)
+    expect(process.listeners('exit')).not.toContain(listener)
+  })
+
   it('releases a terminal after top-level exit reaches quiescence', async () => {
     let exitListener: ((event: { exitCode: number; signal?: number }) => void) | undefined
     const inspector = {

packages/subprocess/subprocess-local/tests/process-exit.spec.ts

@@ -0,0 +1,169 @@
+import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { fileURLToPath } from 'node:url'
+import { execa } from 'execa'
+import { describe, expect, it, vi } from 'vitest'
+import { resolveExampleLaunch } from '@deepseek-ai/dsh-loader-smoke'
+import { createProcessInspector } from '../src/process-inspector.ts'
+import type { ProcessIdentity, ProcessInspector } from '../src/process-inspector.ts'
+import { taskkillProcessTree } from '../src/spawn.ts'
+
+type ExitTrigger = 'direct' | 'uncaught-exception' | 'unhandled-rejection' | 'dispose'
+type ManagedKind = 'ordinary' | 'terminal'
+interface TreeState { root: number; descendant: number }
+
+const repoRoot = fileURLToPath(new URL('../../../../', import.meta.url))
+const hostScript = fileURLToPath(new URL('./fixtures/process-exit-host.ts', import.meta.url))
+const scenarioTimeoutMs = 30_000
+
+function processExists(pid: number): boolean {
+  try {
+    process.kill(pid, 0)
+    return true
+  } catch (error: unknown) {
+    if ((error as NodeJS.ErrnoException).code === 'ESRCH') return false
+    throw error
+  }
+}
+
+async function readTree(path: string): Promise<TreeState> {
+  return vi.waitFor(async () => {
+    const text = await readFile(path, 'utf8')
+    const state = JSON.parse(text) as Partial<TreeState>
+    if (!Number.isSafeInteger(state.root) || !Number.isSafeInteger(state.descendant)
+      || (state.root ?? 0) <= 0 || (state.descendant ?? 0) <= 0 || state.root === state.descendant) {
+      throw new Error(`invalid managed-tree state: ${text}`)
+    }
+    return state as TreeState
+  }, { interval: 10, timeout: scenarioTimeoutMs })
+}
+
+async function captureIdentities(inspector: ProcessInspector, state: TreeState): Promise<ProcessIdentity[]> {
+  return vi.waitFor(() => {
+    const expected = new Set([state.root, state.descendant])
+    const identities = inspector.processTree(state.root).filter(identity => expected.has(identity.pid))
+    if (identities.length !== expected.size) throw new Error('managed tree is not fully observable yet')
+    return identities
+  }, { interval: 10, timeout: scenarioTimeoutMs })
+}
+
+async function waitForGone(state: TreeState): Promise<void> {
+  await Promise.all([state.root, state.descendant].map(pid => vi.waitFor(() => {
+    if (processExists(pid)) throw new Error(`managed pid ${pid} is still alive`)
+  }, { interval: 25, timeout: 10_000 })))
+}
+
+function cleanupTree(state: TreeState | undefined, identities: ProcessIdentity[]): void {
+  if (state === undefined) return
+  if (process.platform === 'win32') {
+    taskkillProcessTree(state.root)
+    for (const pid of [state.descendant, state.root]) {
+      try {
+        process.kill(pid, 'SIGKILL')
+      } catch (_alreadyGone) {
+        // The exact recorded process already exited.
+      }
+    }
+    return
+  }
+  const inspector = createProcessInspector()
+  for (const identity of identities) {
+    try {
+      inspector.signalProcess(identity, 'SIGKILL')
+    } catch (_alreadyGone) {
+      // Exact start identity prevents PID-reuse cleanup from reaching another process.
+    }
+  }
+  if (identities.length === 0) {
+    for (const pid of [state.descendant, state.root]) {
+      try {
+        process.kill(pid, 'SIGKILL')
+      } catch (_alreadyGone) {
+        // The scenario failed before process identities became observable.
+      }
+    }
+  }
+}
+
+async function runScenario(kind: ManagedKind, trigger: ExitTrigger) {
+  const root = await mkdtemp(join(tmpdir(), `dsh-subprocess-host-exit-${kind}-${trigger}-`))
+  const launch = resolveExampleLaunch({
+    srcBin: hostScript,
+    mode: 'src',
+    tsconfigPath: join(repoRoot, 'tsconfig.json'),
+    configArgs: [kind, trigger, root],
+  })
+  const child = execa(launch.command, launch.args, {
+    cwd: repoRoot,
+    env: launch.env,
+    stdin: 'ignore',
+    reject: false,
+    timeout: scenarioTimeoutMs,
+  })
+  let state: TreeState | undefined
+  let identities: ProcessIdentity[] = []
+  let settled = false
+  try {
+    state = await readTree(join(root, 'tree.json'))
+    await vi.waitFor(() => readFile(join(root, 'ready'), 'utf8'), {
+      interval: 10,
+      timeout: scenarioTimeoutMs,
+    })
+    if (process.platform !== 'win32') identities = await captureIdentities(createProcessInspector(), state)
+    await writeFile(join(root, 'proceed'), 'proceed')
+    const outcome = await child
+    settled = true
+    await waitForGone(state)
+    const disposeCounts = trigger === 'dispose'
+      ? JSON.parse(await readFile(join(root, 'dispose.json'), 'utf8')) as {
+        listenersBefore: number
+        listenersAfterLoad: number
+        listenersAfterDispose: number
+      }
+      : undefined
+    return { outcome, disposeCounts }
+  } finally {
+    if (!settled) {
+      child.kill('SIGKILL')
+      await child.catch(() => {})
+    }
+    cleanupTree(state, identities)
+    if (state !== undefined) await waitForGone(state).catch(() => {})
+    await rm(root, { recursive: true, force: true })
+  }
+}
+
+describe('synchronous cleanup on host exit', () => {
+  it.each([
+    { trigger: 'direct' as const, expectedCode: 23, diagnostic: undefined },
+    { trigger: 'uncaught-exception' as const, expectedCode: 1, diagnostic: 'host-exit-uncaught-exception' },
+    { trigger: 'unhandled-rejection' as const, expectedCode: 1, diagnostic: 'host-exit-unhandled-rejection' },
+  ])('removes an ordinary managed tree after $trigger', { timeout: 45_000 }, async ({
+    trigger,
+    expectedCode,
+    diagnostic,
+  }) => {
+    const { outcome } = await runScenario('ordinary', trigger)
+    expect(outcome.exitCode).toBe(expectedCode)
+    expect(outcome.signal).toBeUndefined()
+    if (diagnostic !== undefined) expect(outcome.stderr).toContain(diagnostic)
+  })
+
+  it.skipIf(process.platform === 'win32')(
+    'removes a terminal root and descendant after direct exit',
+    { timeout: 45_000 },
+    async () => {
+      const { outcome } = await runScenario('terminal', 'direct')
+      expect(outcome.exitCode).toBe(23)
+      expect(outcome.signal).toBeUndefined()
+    },
+  )
+
+  it('preserves normal terminate-and-join disposal and removes the exit listener', { timeout: 45_000 }, async () => {
+    const { outcome, disposeCounts } = await runScenario('ordinary', 'dispose')
+    expect(outcome.exitCode).toBe(0)
+    expect(disposeCounts?.listenersAfterLoad).toBe((disposeCounts?.listenersBefore ?? 0) + 1)
+    expect(disposeCounts?.listenersAfterDispose).toBe(disposeCounts?.listenersBefore)
+  })
+})

packages/subprocess/subprocess-local/tests/spawn.spec.ts

@@ -582,6 +582,25 @@ describe('stdio dispositions', () => {
 })
 
 describe('windows tree semantics (injected platform)', () => {
+  it('host-exit termination routes through taskkill immediately', async () => {
+    const killed: number[] = []
+    const running = spawnSubprocess(spec('sleep 60', { graceMs: 60_000 }), {
+      spillDir,
+      platform: 'win32',
+      taskkill: (pid) => {
+        killed.push(pid)
+        try {
+          process.kill(pid, 'SIGKILL')
+        } catch {
+          // Already gone — matches taskkill's tolerated not-found status.
+        }
+      },
+    })
+    running.terminateForHostExit()
+    await running.done
+    expect(killed).toEqual([running.pid])
+  })
+
   it('terminate routes through taskkill by root pid', async () => {
     const killed: number[] = []
     const running = spawnSubprocess(spec('sleep 60', { graceMs: 100 }), {
@@ -631,6 +650,23 @@ describe('waitForExit', () => {
   })
 })
 
+describe('synchronous host-exit termination', () => {
+  it('force-kills the current process tree without waiting for the normal grace', async () => {
+    const running = spawnSubprocess(spec('trap "" TERM; sleep 60', { graceMs: 60_000 }))
+    running.terminateForHostExit()
+    await expect(running.done).resolves.toMatchObject({ exitCode: null, signal: 'SIGKILL' })
+    await expect(running.waitForExit()).resolves.toBe(true)
+
+    const kill = vi.spyOn(process, 'kill')
+    try {
+      running.terminateForHostExit()
+      expect(kill).not.toHaveBeenCalled()
+    } finally {
+      kill.mockRestore()
+    }
+  })
+})
+
 describe('tree-survivor escalation (terminate and bounded waits reach helpers the leader left behind)', () => {
   it('terminate() SIGKILLs a TERM-trapping descendant after the direct child settles', async () => {
     // The leader spawns a TERM-trapping helper with all stdio detached from

packages/subprocess/subprocess-local/tests/terminal.spec.ts

@@ -74,6 +74,7 @@ class FakeInspector implements ProcessInspector {
   }
   signalProcess(identity: ProcessIdentity, signal: 'SIGTERM' | 'SIGKILL') {
     if (this.throwProcess) throw new Error('process raced')
+    if (!this.isAlive(identity)) return
     this.processes.push([identity.pid, signal])
     if (this.removeOnSignal) this.alive.delete(identity.pid)
   }
@@ -82,6 +83,85 @@ class FakeInspector implements ProcessInspector {
 afterEach(() => { vi.useRealTimers() })
 
 describe('LocalTerminalHandle', () => {
+  it('force-kills descendants around the shell during synchronous host exit', () => {
+    const pty = new FakePty()
+    const inspector = new FakeInspector()
+    const first = { pid: 124, started: 'first' }
+    const late = { pid: 125, started: 'late' }
+    inspector.members = [first]
+    inspector.alive.add(pty.pid)
+    inspector.alive.add(first.pid)
+    const signalProcess = inspector.signalProcess.bind(inspector)
+    inspector.signalProcess = (identity, signal) => {
+      signalProcess(identity, signal)
+      if (identity.pid === pty.pid) {
+        inspector.members = [first, late]
+        inspector.alive.add(late.pid)
+      }
+    }
+    const handle = new LocalTerminalHandle(pty.asPty(), inspector, 10)
+
+    handle.terminateForHostExit()
+    expect(inspector.processes).toEqual([
+      [first.pid, 'SIGKILL'],
+      [pty.pid, 'SIGKILL'],
+      [late.pid, 'SIGKILL'],
+    ])
+    expect(pty.kills).toEqual([])
+
+    pty.emitExit()
+    handle.terminateForHostExit()
+    expect(pty.kills).toEqual([])
+  })
+
+  it('uses captured identities and contains shell races when final inspection fails', async () => {
+    const pty = new FakePty()
+    const inspector = new FakeInspector()
+    const captured = { pid: 124, started: 'captured' }
+    inspector.members = [captured]
+    inspector.alive.add(pty.pid)
+    inspector.alive.add(captured.pid)
+    const handle = new LocalTerminalHandle(pty.asPty(), inspector, 10)
+    await handle.inspectForeground()
+    inspector.processTree = () => { throw new Error('process table unavailable') }
+    inspector.throwProcess = true
+
+    expect(() => { handle.terminateForHostExit() }).not.toThrow()
+    expect(inspector.processes).toEqual([])
+    expect(pty.kills).toEqual([])
+  })
+
+  it('uses node-pty only when the shell start identity was unavailable', () => {
+    const pty = new FakePty()
+    const inspector = new FakeInspector()
+    inspector.root = undefined
+    const handle = new LocalTerminalHandle(pty.asPty(), inspector, 10)
+
+    handle.terminateForHostExit()
+    expect(pty.kills).toEqual(['SIGKILL'])
+
+    const racingPty = new FakePty()
+    const racingInspector = new FakeInspector()
+    racingInspector.root = undefined
+    racingPty.throwKill = true
+    const racingHandle = new LocalTerminalHandle(racingPty.asPty(), racingInspector, 10)
+    expect(() => { racingHandle.terminateForHostExit() }).not.toThrow()
+  })
+
+  it('does not signal a recycled terminal root before its delayed exit callback', () => {
+    const pty = new FakePty()
+    const inspector = new FakeInspector()
+    inspector.alive.add(pty.pid)
+    const handle = new LocalTerminalHandle(pty.asPty(), inspector, 10)
+    inspector.root = { pid: pty.pid, started: 'recycled' }
+    inspector.isAlive = identity => identity.started === 'recycled'
+
+    handle.terminateForHostExit()
+
+    expect(inspector.processes).toEqual([])
+    expect(pty.kills).toEqual([])
+  })
+
   it('bridges terminal bytes, foreground control, and signalled exit facts', async () => {
     const pty = new FakePty()
     const inspector = new FakeInspector()