test(subprocess): cover control disposal and runner forwarding

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

@@ -2,6 +2,7 @@ import { mkdtemp, rm } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { fileURLToPath } from 'node:url'
+import { once } from 'node:events'
 import { afterEach, describe, expect, it } from 'vitest'
 import { Context } from '@deepseek-ai/cordis'
 import type { SubprocessHandle } from '@deepseek-ai/dsh-subprocess'
@@ -24,6 +25,24 @@ afterEach(async () => {
 })
 
 describe('managed subprocess control pipe', () => {
+  it('closes the caller endpoint when service disposal terminates an active program', async () => {
+    ctx = new Context()
+    await ctx.plugin(LocalSubprocessRuntime)
+    handle = ctx.subprocess.spawn({
+      argv: [process.execPath, '--input-type=module', '-e',
+        'import { Socket } from "node:net"; const c = new Socket({fd:7,readable:true,writable:true}); c.write("ready"); setInterval(()=>{},60000)'],
+      cwd: process.cwd(),
+      stdio: { stdin: 'ignore', stdout: { maxBytes: 32 }, stderr: { maxBytes: 32 }, control: 'pipe' },
+      graceMs: 1000,
+    })
+    const channel = handle.control
+    if (channel === undefined) throw new Error('requested control pipe is absent')
+    await once(channel, 'data')
+    await ctx.fiber.dispose()
+    expect(channel.destroyed).toBe(true)
+    expect(await handle.waitForExit()).toBe(true)
+  })
+
   it('leaves the channel absent on an ordinary spawn', async () => {
     ctx = new Context()
     await ctx.plugin(LocalSubprocessRuntime)

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

@@ -6,11 +6,11 @@ afterEach(() => {
 })
 
 describe.skipIf(process.platform !== 'linux')('Linux libc execve binding', () => {
-  it('preserves inherited stdio, null-terminates argv and envp, and reports execve errno', async () => {
+  it.each([undefined, 'pipe'] as const)('preserves inherited stdio with control %s and reports execve errno', async (control) => {
     const nativeExecve = vi.fn(() => -1)
     const nativeFcntl = vi.fn((fd: number, command: number) => {
       if (command === 2) return 0
-      return [1, 0, 5][fd]
+      return fd === 7 ? 1 : [1, 0, 5][fd]
     })
     const func = vi.fn((declaration: string) => declaration.includes('execve')
       ? nativeExecve
@@ -30,7 +30,7 @@ describe.skipIf(process.platform !== 'linux')('Linux libc execve binding', () =>
 
     let failure: unknown
     try {
-      execve('/missing/tool', ['tool', 'literal arg'], { A: '1', EMPTY: '' })
+      execve('/missing/tool', ['tool', 'literal arg'], { A: '1', EMPTY: '' }, control)
     } catch (error) {
       failure = error
     }
@@ -40,6 +40,7 @@ describe.skipIf(process.platform !== 'linux')('Linux libc execve binding', () =>
       [1, 1, 0],
       [2, 1, 0],
       [2, 2, 4],
+      ...control === 'pipe' ? [[7, 1, 0], [7, 2, 0]] : [],
     ])
     expect(nativeExecve).toHaveBeenCalledExactlyOnceWith(
       '/missing/tool',

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

@@ -126,6 +126,17 @@ async function runWindows(
 }
 
 describe('closed runner protocol', () => {
+  it('preserves the explicit control request and rejects other transport values', () => {
+    const files = track(createLinuxLaunchFiles({ cwd: '/target', env: {}, control: 'pipe' }))
+    expect(consumeLinuxLaunchRequest(files.requestPath)).toEqual({ cwd: '/target', env: {}, control: 'pipe' })
+    expect(parseWindowsStartRequest({ type: 'start', cwd: 'C:\\target', env: {}, control: 'pipe' }))
+      .toEqual({ type: 'start', cwd: 'C:\\target', env: {}, control: 'pipe' })
+    const invalid = track(createLinuxLaunchFiles({ cwd: '/target', env: {} }))
+    writeFileSync(invalid.requestPath, JSON.stringify({ cwd: '/target', env: {}, control: 'ipc' }))
+    expect(() => consumeLinuxLaunchRequest(invalid.requestPath)).toThrow('invalid Linux launch request')
+    expect(() => parseWindowsStartRequest({ type: 'start', cwd: 'C:\\target', env: {}, control: 'ipc' })).toThrow()
+  })
+
   it('creates, consumes, reports through, and cleans one private Linux exchange', () => {
     const files = track(createLinuxLaunchFiles({ cwd: '/target', env: { A: '1' } }))
     if (process.platform !== 'win32') {
@@ -714,7 +725,7 @@ describe('Windows Job runner protocol owner', () => {
     expect(host.env).toEqual({ SAFE: 'bootstrap' })
   })
 
-  it('closes every target carrier before the first Windows poll', async () => {
+  it.each([undefined, 'pipe'] as const)('closes every target carrier with control %s before the first Windows poll', async (control) => {
     const events: string[] = []
     const interval = vi.spyOn(globalThis, 'setInterval').mockImplementation((callback: () => void) => {
       events.push('interval')
@@ -730,8 +741,8 @@ describe('Windows Job runner protocol owner', () => {
           return 0
         }),
       })
-      await runWindows(host, native)
-      expect(events).toEqual(['close:4', 'close:5', 'close:6', 'interval', 'poll'])
+      await runWindows(host, native, { type: 'start', cwd: 'C:\\target', env: {}, ...control === undefined ? {} : { control } })
+      expect(events).toEqual(['close:4', 'close:5', 'close:6', ...control === 'pipe' ? ['close:7'] : [], 'interval', 'poll'])
       expect(host.sent).toEqual([{ type: 'target-exit', exitCode: 0 }])
       expect(host.exitCode).toBe(0)
     } finally {