feat(subprocess): carry a duplex control pipe through managed launches

packages/fs/tool-fs-search/tests/tools.spec.ts

@@ -97,6 +97,7 @@ class FakeReader implements SubprocessOutputReader {
  * abort→terminate escalation.
  */
 class FakeHandle implements SubprocessHandle {
+  readonly control = undefined
   readonly stdin = undefined
   readonly stdout = undefined
   readonly stderr = undefined

packages/sandbox/sandbox-windows-acl/package.json

@@ -33,7 +33,8 @@
   ],
   "license": "MIT",
   "peerDependencies": {
-    "@deepseek-ai/cordis": "workspace:^"
+    "@deepseek-ai/cordis": "workspace:^",
+    "@deepseek-ai/dsh-subprocess": "workspace:^"
   },
   "dependencies": {
     "@deepseek-ai/dsh-win32-process": "workspace:^",
@@ -42,6 +43,7 @@
   "devDependencies": {
     "@deepseek-ai/dsh-pwsh-local": "workspace:^",
     "@deepseek-ai/dsh-sandbox-local": "workspace:^",
-    "@deepseek-ai/cordis": "workspace:^"
+    "@deepseek-ai/cordis": "workspace:^",
+    "@deepseek-ai/dsh-subprocess": "workspace:^"
   }
 }

packages/sandbox/sandbox-windows-acl/src/index.ts

@@ -112,6 +112,8 @@ export interface AclSandboxSpawnOptions {
    * child dies with the caller; stdout/stderr in the result are empty.
    */
   stdio?: 'pipe' | 'inherit'
+  /** Control pipe forwarded to the same payload descriptor in inherited-stdio mode. */
+  controlFileDescriptor?: 7
 }
 
 /** A settled confined child: captured stdio and the exit code. */
@@ -349,11 +351,17 @@ export class AclSandbox {
     const api = this.api
     const token = this.token
     if (api === undefined || token === undefined) throw new Error('AclSandbox is not initialized: call init() first')
+    if (options.controlFileDescriptor !== undefined && options.stdio !== 'inherit') {
+      throw new Error('control pipe requires inherited stdio')
+    }
     const args = options.args ?? []
     const cwd = options.cwd ?? process.cwd()
 
     if (options.stdio === 'inherit') {
-      const native = spawnSandboxedInherited(api, token, { command: options.command, args, cwd })
+      const native = spawnSandboxedInherited(api, token, {
+        command: options.command, args, cwd,
+        ...options.controlFileDescriptor === undefined ? {} : { controlFileDescriptor: options.controlFileDescriptor },
+      })
       let exitCodePromise: Promise<number> | undefined
       return {
         pid: native.pid,

packages/sandbox/sandbox-windows-acl/src/runner.ts

@@ -44,7 +44,8 @@
  * @module @deepseek-ai/dsh-sandbox-windows-acl/runner
  */
 
-import { existsSync, mkdtempSync, rmSync, statSync } from 'node:fs'
+import { SUBPROCESS_CONTROL_ENV, SUBPROCESS_CONTROL_FD } from '@deepseek-ai/dsh-subprocess/control'
+import { closeSync, existsSync, mkdtempSync, rmSync, statSync } from 'node:fs'
 import { join } from 'node:path'
 
 import { win32 } from './ffi.ts'
@@ -182,7 +183,9 @@ async function main(): Promise<number> {
       command: parsed.command,
       args: parsed.args,
       stdio: 'inherit',
+      ...process.env[SUBPROCESS_CONTROL_ENV] === 'pipe' ? { controlFileDescriptor: SUBPROCESS_CONTROL_FD } : {},
     })
+    if (process.env[SUBPROCESS_CONTROL_ENV] === 'pipe') closeSync(SUBPROCESS_CONTROL_FD)
     const result = await child.wait()
     return result.exitCode
   } finally {

packages/sandbox/sandbox-windows-acl/src/spawn.ts

@@ -44,7 +44,7 @@ export function spawnSandboxed(
 export function spawnSandboxedInherited(
   api: Win32Bindings,
   token: NativePtr,
-  options: { command: string; args: readonly string[]; cwd: string },
+  options: { command: string; args: readonly string[]; cwd: string; controlFileDescriptor?: 7 },
 ): SpawnedInherited {
   return spawnInheritedJobProcess(api, { ...options, token })
 }

packages/sandbox/sandbox-windows-acl/tsconfig.json

@@ -4,7 +4,9 @@
     "rootDir": "src",
     "outDir": "lib/types"
   },
-  "include": ["src"],
+  "include": [
+    "src"
+  ],
   "references": [
     {
       "path": "../../../vendor/cosmokit"
@@ -17,6 +19,9 @@
     },
     {
       "path": "../../subprocess/win32-process"
+    },
+    {
+      "path": "../../subprocess/subprocess"
     }
   ]
 }

packages/shell/bash-local/tests/executor.spec.ts

@@ -307,6 +307,7 @@ describe('LocalBashExecutor.start (background process handles)', () => {
       }),
     }
     vi.spyOn(ctx.subprocess, 'spawn').mockReturnValue({
+      control: undefined,
       stdin: undefined,
       stdout: undefined,
       stderr: undefined,
@@ -336,6 +337,7 @@ describe('LocalBashExecutor.start (background process handles)', () => {
       value: () => { throw new Error('provider formatting must not escape') },
     })
     vi.spyOn(ctx.subprocess, 'spawn').mockReturnValue({
+      control: undefined,
       stdin: undefined,
       stdout: undefined,
       stderr: undefined,

packages/shell/bash-sandbox/tests/sandbox.spec.ts

@@ -575,6 +575,7 @@ describe('background sandbox facts', () => {
       readFrom: () => ({ text: '', nextOffset: 0, lossy: false }),
     }
     vi.spyOn(ctx.subprocess, 'spawn').mockReturnValue({
+      control: undefined,
       stdin: undefined,
       stdout: undefined,
       stderr: undefined,

packages/shell/pwsh-local/tests/executor.spec.ts

@@ -206,6 +206,7 @@ describe('spawn construction (pure, every platform)', () => {
     override spawn(spec: SubprocessSpawnSpec): SubprocessHandle {
       this.specs.push(spec)
       return {
+        control: undefined,
         stdin: undefined,
         stdout: undefined,
         stderr: undefined,

packages/subagent/subagent-acp/tests/subagent-acp.spec.ts

@@ -88,6 +88,7 @@ async function waitForFile(file: string, timeoutMs: number): Promise<void> {
 
 function rejectFinalExitWait(child: SubprocessHandle, message: string): SubprocessHandle {
   return {
+    control: child.control,
     stdin: child.stdin,
     stdout: child.stdout,
     stderr: child.stderr,
@@ -111,6 +112,7 @@ function rejectFinalExitWaitAfterExit(child: SubprocessHandle, message: string):
 
 function tapBoundedExitWait(child: SubprocessHandle, onWait: () => void): SubprocessHandle {
   return {
+    control: child.control,
     stdin: child.stdin,
     stdout: child.stdout,
     stderr: child.stderr,
@@ -132,6 +134,7 @@ function replaceProtocolStreams(
   if (child.stdin === undefined) throw new Error('expected piped child stdin')
   stdin.pipe(child.stdin)
   return {
+    control: child.control,
     stdin,
     stdout,
     stderr: child.stderr,
@@ -169,6 +172,7 @@ function closeProtocolOnPrompt(child: SubprocessHandle, onClose: () => void = ()
 
 function replaceProcessOutcome(child: SubprocessHandle, outcome: SubprocessOutcome): SubprocessHandle {
   return {
+    control: child.control,
     stdin: child.stdin,
     stdout: child.stdout,
     stderr: child.stderr,
@@ -298,6 +302,7 @@ describe('disposeAcpChild (the backend-owned teardown ladder over seam verbs)',
     const stdin = new PassThrough()
     const calls: string[] = []
     const child: SubprocessHandle = {
+      control: undefined,
       stdin,
       stdout: undefined,
       stderr: undefined,
@@ -360,6 +365,7 @@ describe('disposeAcpChild (the backend-owned teardown ladder over seam verbs)',
       .mockResolvedValueOnce(true)
     const terminate = vi.fn()
     const child: SubprocessHandle = {
+      control: undefined,
       stdin: new PassThrough(),
       stdout: undefined,
       stderr: undefined,
@@ -381,6 +387,7 @@ describe('disposeAcpChild (the backend-owned teardown ladder over seam verbs)',
       .mockRejectedValueOnce(initialFailure)
       .mockRejectedValueOnce(finalFailure)
     const child: SubprocessHandle = {
+      control: undefined,
       stdin: new PassThrough(),
       stdout: undefined,
       stderr: undefined,
@@ -407,6 +414,7 @@ describe('disposeAcpChild (the backend-owned teardown ladder over seam verbs)',
     const stdin = new PassThrough()
     const stdout = new PassThrough()
     const child: SubprocessHandle = {
+      control: undefined,
       stdin,
       stdout,
       stderr: undefined,
@@ -826,6 +834,7 @@ describe('dsh-subagent-acp', () => {
       disposeEofGraceMs: 50,
       disposeGraceMs: 50,
       spawn: () => ({
+        control: undefined,
         stdin,
         stdout,
         stderr: undefined,
@@ -1311,6 +1320,7 @@ describe('dsh-subagent-acp', () => {
         const child = spawnSubprocess(spec)
         realChild = child
         return closeProtocolOnPrompt({
+          control: child.control,
           stdin: child.stdin,
           stdout: child.stdout,
           stderr: child.stderr,

packages/subagent/subagent-claude-code/tests/subagent-claude-code.spec.ts

@@ -167,6 +167,7 @@ function fakeChild(options: FakeChildOptions = {}): FakeChild {
     })
   })
   const handle: SubprocessHandle = {
+    control: undefined,
     stdin,
     stdout,
     stderr: undefined,

packages/subagent/subagent-codex/tests/subagent-codex.spec.ts

@@ -210,6 +210,7 @@ function fakeChild(options: FakeChildOptions = {}): FakeChild {
     })
   })
   const handle: SubprocessHandle = {
+    control: undefined,
     stdin: toChild,
     stdout: fromChild,
     stderr,

packages/subprocess/subprocess-local/src/control-spawn.ts

@@ -0,0 +1,35 @@
+/** Parent-side setup for one explicitly requested inherited control pipe. */
+
+import { SUBPROCESS_CONTROL_ENV, SUBPROCESS_CONTROL_FD } from '@deepseek-ai/dsh-subprocess/control'
+import type { Duplex, Readable, Writable } from 'node:stream'
+
+/**
+ * Read the optional extra pipe from Node's stdio tuple.
+ * @param child - child whose requested extra pipe was allocated by Node.
+ * @param control - requested transport, or undefined when absent.
+ * @returns the parent duplex endpoint, absent when not requested or native startup failed.
+ */
+export function controlPipe(
+  child: { readonly stdio: ReadonlyArray<Readable | Writable | null | undefined> },
+  control?: 'pipe',
+): Duplex | undefined {
+  // Node's type declaration names only the first five descriptor slots.
+  const streams: ReadonlyArray<Readable | Writable | null | undefined> = child.stdio
+  return control === 'pipe' ? streams[SUBPROCESS_CONTROL_FD] as Duplex | undefined : undefined
+}
+
+/**
+ * Stamp the private marker on a fresh child environment after rejecting a caller override.
+ * @param env - newly materialized child environment, owned by the caller.
+ * @param control - requested control transport, or undefined when absent.
+ * @returns the same environment with the provider-owned launch marker when requested.
+ */
+export function controlEnvironment<T extends NodeJS.ProcessEnv>(env: T, control?: 'pipe'): T {
+  for (const [key, value] of Object.entries(env)) {
+    if (key.toUpperCase() === SUBPROCESS_CONTROL_ENV && value !== undefined) {
+      throw new Error(`${SUBPROCESS_CONTROL_ENV} is reserved for subprocess control-channel setup`)
+    }
+  }
+  if (control === 'pipe') Object.assign(env, { [SUBPROCESS_CONTROL_ENV]: 'pipe' })
+  return env
+}

packages/subprocess/subprocess-local/src/index.ts

@@ -104,7 +104,7 @@ export class LocalSubprocessRuntime extends SubprocessRuntime {
       pending.push(Promise.all([
         handle.done.catch(() => {}),
         handle.waitForExit(),
-      ]).then(() => { this.live.delete(handle) }))
+      ]).then(() => { handle.control?.destroy(); this.live.delete(handle) }))
     }
     for (const terminal of this.terminals) {
       pending.push(terminal.terminate().then(() => { this.terminals.delete(terminal) }))

packages/subprocess/subprocess-local/src/linux-execve.ts

@@ -2,12 +2,14 @@
 
 import { getSystemErrorMessage, getSystemErrorName } from 'node:util'
 import koffi from 'koffi'
+import { SUBPROCESS_CONTROL_FD } from '@deepseek-ai/dsh-subprocess/control'
 
 /** Replace the current process image while preserving the supplied argv and environment. */
 export type LinuxExecve = (
   file: string,
   argv: string[],
   env: Record<string, string>,
+  control?: 'pipe',
 ) => never
 
 type NativeExecve = (
@@ -51,8 +53,11 @@ export function loadLinuxExecve(): LinuxExecve {
   const nativeFcntl = libc.func(
     'int fcntl(int fd, int cmd, int arg)',
   ) as NativeFcntl
-  cachedExecve = (file, argv, env) => {
-    for (const fd of STANDARD_FILE_DESCRIPTORS) {
+  cachedExecve = (file, argv, env, control) => {
+    const descriptors = control === 'pipe'
+      ? [...STANDARD_FILE_DESCRIPTORS, SUBPROCESS_CONTROL_FD]
+      : STANDARD_FILE_DESCRIPTORS
+    for (const fd of descriptors) {
       const flags = nativeFcntl(fd, F_GETFD, 0)
       if (flags === -1) throw systemError(koffi.errno(), 'fcntl')
       if ((flags & FD_CLOEXEC) === 0) continue

packages/subprocess/subprocess-local/src/linux-scope.ts

@@ -1,5 +1,6 @@
 /** Linux user-systemd scope launch and managed-range ownership. */
 
+import { controlPipe } from './control-spawn.ts'
 import { execFile, spawn, spawnSync } from 'node:child_process'
 import { randomBytes } from 'node:crypto'
 import { existsSync } from 'node:fs'
@@ -512,7 +513,10 @@ export function launchLinuxScope(
   internals: LinuxScopeInternals = {},
 ): ManagedProcessLaunch {
   const invocation = internals.runnerInvocation ?? spawnRunnerInvocation()
-  const files = createLinuxLaunchFiles({ cwd: spec.cwd, env: targetEnv })
+  const files = createLinuxLaunchFiles({
+    cwd: spec.cwd, env: targetEnv,
+    ...spec.stdio.control === undefined ? {} : { control: spec.stdio.control },
+  })
   const startup = new LinuxScopeStartup(files, 'subprocess')
   const unitBase = unitStem('dsh-subprocess')
   let child: ReturnType<typeof spawn>
@@ -547,6 +551,7 @@ export function launchLinuxScope(
     stdin: child.stdin,
     stdout: child.stdout,
     stderr: child.stderr,
+    control: controlPipe(child, spec.stdio.control),
     direct: directOutcome(child, startup),
     owner,
   }

packages/subprocess/subprocess-local/src/managed-owner.ts

@@ -1,6 +1,6 @@
 /** Minimal managed-range ownership bound to one ordinary subprocess handle. */
 
-import type { Readable, Writable } from 'node:stream'
+import type { Duplex, Readable, Writable } from 'node:stream'
 import type { SubprocessOutcome } from '@deepseek-ai/dsh-subprocess'
 
 /** Platform owner used by termination and whole-range settlement. */
@@ -20,6 +20,7 @@ export interface ManagedProcessLaunch {
   stdin: Writable | null
   stdout: Readable | null
   stderr: Readable | null
+  control?: Duplex | undefined
   direct: Promise<SubprocessOutcome>
   owner: BoundProcessOwner
 }

packages/subprocess/subprocess-local/src/runner-launch.ts

@@ -7,6 +7,8 @@ import { inspect } from 'node:util'
 import { fileURLToPath } from 'node:url'
 import type { SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
 import { childEnv } from './spawn.ts'
+import { controlEnvironment } from './control-spawn.ts'
+import { SUBPROCESS_CONTROL_FD } from '@deepseek-ai/dsh-subprocess/control'
 
 /** The one private environment variable consumed before target state is restored. */
 export const SUBPROCESS_RUNNER_ENV = 'DSH_SUBPROCESS_RUNNER' as const
@@ -124,8 +126,14 @@ export function runnerStdio(
     spec.stdio.stdout === 'inherit' ? 'inherit' : 'pipe',
     spec.stdio.stderr === 'inherit' ? 'inherit' : 'pipe',
   ]
-  if (!ipc) return targetStdio
-  return [
+  if (!ipc) {
+    if (spec.stdio.control === 'pipe') {
+      while (targetStdio.length < SUBPROCESS_CONTROL_FD) targetStdio.push('ignore')
+      targetStdio.push('pipe')
+    }
+    return targetStdio
+  }
+  const runner: StdioOptions = [
     'ignore',
     'ignore',
     'ignore',
@@ -134,6 +142,8 @@ export function runnerStdio(
     spec.stdio.stdout === 'inherit' ? 1 : 'pipe',
     spec.stdio.stderr === 'inherit' ? 2 : 'pipe',
   ]
+  if (spec.stdio.control === 'pipe') runner.push('pipe')
+  return runner
 }
 
 function windowsEnvironmentValue(
@@ -288,7 +298,7 @@ function validateNoNullByte(property: string, value: string, argument = false):
  * @returns complete target environment after Node-equivalent validation.
  */
 export function targetEnvironment(
-  spec: Pick<SubprocessSpawnSpec, 'argv' | 'cwd' | 'env'>,
+  spec: Pick<SubprocessSpawnSpec, 'argv' | 'cwd' | 'env'> & { stdio?: SubprocessSpawnSpec['stdio'] },
 ): Record<string, string> {
   spec.argv.forEach((value, index) => {
     validateNoNullByte(index === 0 ? 'file' : `args[${String(index - 1)}]`, value, true)
@@ -301,5 +311,5 @@ export function targetEnvironment(
     validateNoNullByte(`options.env['${key}']`, key)
     validateNoNullByte(`options.env['${key}']`, value)
   }
-  return env
+  return controlEnvironment(env, spec.stdio?.control)
 }

packages/subprocess/subprocess-local/src/runner-protocol.ts

@@ -17,6 +17,7 @@ import { basename, dirname, isAbsolute, join } from 'node:path'
 export interface LinuxLaunchRequest {
   cwd: string
   env: Record<string, string>
+  control?: 'pipe'
 }
 
 /** Bounded Node-shaped error fields allowed across a private runner boundary. */
@@ -37,6 +38,7 @@ export interface WindowsStartRequest {
   type: 'start'
   cwd: string
   env: Record<string, string>
+  control?: 'pipe'
 }
 
 /** The only parent-to-runner control message on Windows. */
@@ -137,11 +139,12 @@ export function consumeLinuxLaunchRequest(requestPath: string): LinuxLaunchReque
   const text = readFileSync(requestPath, 'utf8')
   unlinkSync(requestPath)
   const value: unknown = JSON.parse(text)
-  if (!isRecord(value) || !hasExactKeys(value, ['cwd', 'env'])
-    || typeof value.cwd !== 'string' || !isStringRecord(value.env)) {
+  if (!isRecord(value) || !hasExactKeys(value, ['cwd', 'env'], ['control'])
+    || typeof value.cwd !== 'string' || !isStringRecord(value.env)
+    || (value.control !== undefined && value.control !== 'pipe')) {
     throw new Error('subprocess runner received an invalid Linux launch request')
   }
-  return { cwd: value.cwd, env: value.env }
+  return { cwd: value.cwd, env: value.env, ...value.control === 'pipe' ? { control: 'pipe' as const } : {} }
 }
 
 /**
@@ -171,11 +174,12 @@ export function readLinuxStartupError(path: string): LinuxStartupError | undefin
  * @returns validated target start request.
  */
 export function parseWindowsStartRequest(value: unknown): WindowsStartRequest {
-  if (!isRecord(value) || !hasExactKeys(value, ['type', 'cwd', 'env'])
-    || value.type !== 'start' || typeof value.cwd !== 'string' || !isStringRecord(value.env)) {
+  if (!isRecord(value) || !hasExactKeys(value, ['type', 'cwd', 'env'], ['control'])
+    || value.type !== 'start' || typeof value.cwd !== 'string' || !isStringRecord(value.env)
+    || (value.control !== undefined && value.control !== 'pipe')) {
     throw new Error('subprocess runner received an invalid Windows start request')
   }
-  return { type: 'start', cwd: value.cwd, env: value.env }
+  return { type: 'start', cwd: value.cwd, env: value.env, ...value.control === 'pipe' ? { control: 'pipe' as const } : {} }
 }
 
 /**

packages/subprocess/subprocess-local/src/spawn-runner.ts

@@ -1,5 +1,6 @@
 /** One-shot Linux exec bootstrap and Windows Job-owning subprocess runner. */
 
+import { SUBPROCESS_CONTROL_FD } from '@deepseek-ai/dsh-subprocess/control'
 import { closeSync } from 'node:fs'
 import {
   closeHandleChecked,
@@ -42,7 +43,7 @@ type RunnerHost = Pick<NodeJS.Process, 'env' | 'exitCode' | 'connected' | 'cwd'
 
 /** Injectable operations used by the protocol-owner tests. */
 export interface SpawnRunnerInternals {
-  execve(file: string, argv: string[], env: Record<string, string>): never
+  execve(file: string, argv: string[], env: Record<string, string>, control?: 'pipe'): never
   loadWin32ProcessBindings(): CurrentTokenProcessBindings
   spawnCurrentTokenJobProcess: typeof spawnCurrentTokenJobProcess
   closeFileDescriptor(fileDescriptor: number): void
@@ -55,7 +56,7 @@ export interface SpawnRunnerInternals {
 
 const defaultInternals: SpawnRunnerInternals = {
   /* v8 ignore next -- source/built/packaged subprocess smoke executes this only in a replaceable child process. */
-  execve: (file, argv, env) => loadLinuxExecve()(file, argv, env),
+  execve: (file, argv, env, control) => loadLinuxExecve()(file, argv, env, control),
   loadWin32ProcessBindings,
   spawnCurrentTokenJobProcess,
   closeFileDescriptor: closeSync,
@@ -134,22 +135,25 @@ function execLinuxFile(
   argv: string[],
   env: Record<string, string>,
   internals: SpawnRunnerInternals,
+  control?: 'pipe',
 ): never {
   try {
-    return internals.execve(file, argv, env)
+    return control === undefined ? internals.execve(file, argv, env) : internals.execve(file, argv, env, control)
   } catch (error) {
     if ((error as NodeJS.ErrnoException).code !== 'ENOEXEC') throw error
-    return internals.execve('/bin/sh', ['/bin/sh', file, ...argv.slice(1)], env)
+    return control === undefined
+      ? internals.execve('/bin/sh', ['/bin/sh', file, ...argv.slice(1)], env)
+      : internals.execve('/bin/sh', ['/bin/sh', file, ...argv.slice(1)], env, control)
   }
 }
 
 function execLinuxTarget(
-  request: { cwd: string; env: Record<string, string> },
+  request: { cwd: string; env: Record<string, string>; control?: 'pipe' },
   argv: string[],
   internals: SpawnRunnerInternals,
 ): never {
   const program = argv[0] as string
-  if (program.includes('/')) return execLinuxFile(program, argv, request.env, internals)
+  if (program.includes('/')) return execLinuxFile(program, argv, request.env, internals, request.control)
   const path = request.env.PATH ?? '/usr/bin:/bin'
   let permissionFailure: Error | undefined
   for (const directory of path.split(':')) {
@@ -158,7 +162,7 @@ function execLinuxTarget(
       : `${request.cwd}${request.cwd.endsWith('/') ? '' : '/'}${directory}`
     const candidate = `${root}${root.endsWith('/') ? '' : '/'}${program}`
     try {
-      return execLinuxFile(candidate, argv, request.env, internals)
+      return execLinuxFile(candidate, argv, request.env, internals, request.control)
     } catch (error) {
       const code = (error as NodeJS.ErrnoException).code
       if (code === 'EACCES') {
@@ -309,11 +313,11 @@ class WindowsJobRunner {
         args,
         cwd: request.cwd,
         env: request.env,
-        stdio: { stdin: 4, stdout: 5, stderr: 6 },
+        stdio: { stdin: 4, stdout: 5, stderr: 6, ...request.control === 'pipe' ? { control: SUBPROCESS_CONTROL_FD } : {} },
       })
       this.processHandle = spawned.process
       this.jobHandle = spawned.job
-      for (const fileDescriptor of [4, 5, 6]) {
+      for (const fileDescriptor of request.control === 'pipe' ? [4, 5, 6, SUBPROCESS_CONTROL_FD] : [4, 5, 6]) {
         this.internals.closeFileDescriptor(fileDescriptor)
       }
       this.pollTimer = setInterval(() => { this.poll() }, 10)

packages/subprocess/subprocess-local/src/spawn.ts

@@ -27,6 +27,8 @@ import type {
 } from '@deepseek-ai/dsh-subprocess'
 import type { BoundProcessOwner, ManagedProcessLaunch } from './managed-owner.ts'
 import { waitWithAbort } from './managed-owner.ts'
+import { controlEnvironment, controlPipe } from './control-spawn.ts'
+import { SUBPROCESS_CONTROL_FD } from '@deepseek-ai/dsh-subprocess/control'
 import { linuxProcessGroupHasLiveMembers } from './process-inspector.ts'
 
 type SpawnProcess = (
@@ -625,6 +627,7 @@ export function bindManagedProcess(
     stdin: stdinMode === 'pipe' ? stdin ?? undefined : undefined,
     stdout: outMode === 'pipe' ? stdout ?? undefined : undefined,
     stderr: errMode === 'pipe' ? stderr ?? undefined : undefined,
+    control: launch.control,
     /* v8 ignore stop */
     collected: {
       ...stdoutCollector !== undefined ? { stdout: stdoutCollector } : {},
@@ -647,14 +650,19 @@ export function spawnSubprocess(spec: SubprocessSpawnSpec, internals: SpawnInter
   const binding = prepareManagedProcessBinding(internals)
   const platform = internals.platform ?? process.platform
   const [program, ...args] = spec.argv
+  const stdio: import('node:child_process').StdioOptions = [
+    spec.stdio.stdin === 'ignore' ? 'ignore' : 'pipe',
+    spec.stdio.stdout === 'inherit' ? 'inherit' : 'pipe',
+    spec.stdio.stderr === 'inherit' ? 'inherit' : 'pipe',
+  ]
+  if (spec.stdio.control === 'pipe') {
+    while (stdio.length < SUBPROCESS_CONTROL_FD) stdio.push('ignore')
+    stdio.push('pipe')
+  }
   const child = (internals.spawn ?? spawn)(program as string, args, {
     cwd: spec.cwd,
-    env: childEnv(spec.env),
-    stdio: [
-      spec.stdio.stdin === 'ignore' ? 'ignore' : 'pipe',
-      spec.stdio.stdout === 'inherit' ? 'inherit' : 'pipe',
-      spec.stdio.stderr === 'inherit' ? 'inherit' : 'pipe',
-    ],
+    env: controlEnvironment(childEnv(spec.env), spec.stdio.control),
+    stdio,
     detached: platform !== 'win32',
     windowsHide: platform === 'win32',
   })
@@ -672,6 +680,7 @@ export function spawnSubprocess(spec: SubprocessSpawnSpec, internals: SpawnInter
     stdin: child.stdin,
     stdout: child.stdout,
     stderr: child.stderr,
+    control: controlPipe(child, spec.stdio.control),
     direct,
     owner,
   }, binding)

packages/subprocess/subprocess-local/src/windows-job.ts

@@ -1,5 +1,6 @@
 /** Windows parent-side launch and ownership for the private Job runner. */
 
+import { controlPipe } from './control-spawn.ts'
 import { spawn } from 'node:child_process'
 import { closeSync, openSync } from 'node:fs'
 import { devNull } from 'node:os'
@@ -186,7 +187,7 @@ export function launchWindowsJob(
     runnerSpawned = true
     try {
       if (child.send === undefined) throw new Error('subprocess-local: Windows runner has no IPC channel')
-      child.send({ type: 'start', cwd: spec.cwd, env: targetEnv }, (error) => {
+      child.send({ type: 'start', cwd: spec.cwd, env: targetEnv, ...spec.stdio.control === undefined ? {} : { control: spec.stdio.control } }, (error) => {
         if (error === null) return
         failInfrastructure(error)
         owner.terminateForHostExit()
@@ -226,6 +227,7 @@ export function launchWindowsJob(
     stdin: spec.stdio.stdin === 'ignore' ? null : targetStdin,
     stdout: child.stdio[5] as Readable | null,
     stderr: child.stdio[6] as Readable | null,
+    control: controlPipe(child, spec.stdio.control),
     direct: direct.promise,
     owner,
   }

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

@@ -0,0 +1,65 @@
+import { mkdtemp, rm } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import { fileURLToPath } from 'node:url'
+import { afterEach, describe, expect, it } from 'vitest'
+import { Context } from '@deepseek-ai/cordis'
+import type { SubprocessHandle } from '@deepseek-ai/dsh-subprocess'
+import { SUBPROCESS_CONTROL_ENV } from '@deepseek-ai/dsh-subprocess/control'
+import { LocalSubprocessRuntime } from '../src/index.ts'
+
+const fixture = fileURLToPath(new URL('./fixtures/control-child.ts', import.meta.url))
+const helper = fileURLToPath(new URL('../../subprocess/src/control.ts', import.meta.url))
+let ctx: Context | undefined
+let root: string | undefined
+let handle: SubprocessHandle | undefined
+
+afterEach(async () => {
+  handle?.control?.destroy()
+  await ctx?.fiber.dispose()
+  if (root !== undefined) await rm(root, { recursive: true, force: true })
+  ctx = undefined
+  root = undefined
+  handle = undefined
+})
+
+describe('managed subprocess control pipe', () => {
+  it('returns exact binary control bytes independently of stdout and stderr', async () => {
+    root = await mkdtemp(join(tmpdir(), 'dsh-control-'))
+    ctx = new Context()
+    await ctx.plugin(LocalSubprocessRuntime)
+    const input = Buffer.alloc(256 * 1024)
+    for (let index = 0; index < input.length; index++) input[index] = index % 256
+    handle = ctx.subprocess.spawn({
+      argv: [process.execPath, fixture, helper, String(input.length)],
+      cwd: root,
+      stdio: { stdin: 'ignore', stdout: { maxBytes: 1024 }, stderr: { maxBytes: 1024 }, control: 'pipe' },
+      graceMs: 1000,
+    })
+    const channel = handle.control
+    if (channel === undefined) throw new Error('requested control pipe is absent')
+    const received = (async () => {
+      const chunks: Buffer[] = []
+      for await (const chunk of channel) chunks.push(Buffer.from(chunk))
+      return Buffer.concat(chunks)
+    })()
+    channel.write(input)
+    expect(await received).toEqual(input)
+    expect(await handle.done).toEqual({ exitCode: 0, signal: null })
+    expect(handle.collected.stdout?.readFrom(0).text).toBe('ordinary stdout\n')
+    expect(handle.collected.stderr?.readFrom(0).text).toBe('ordinary stderr\n')
+    expect(await handle.waitForExit()).toBe(true)
+  })
+
+  it('rejects a caller-authored control marker before starting a child', async () => {
+    ctx = new Context()
+    await ctx.plugin(LocalSubprocessRuntime)
+    expect(() => ctx?.subprocess.spawn({
+      argv: [process.execPath, '-e', 'throw new Error("must not execute")'],
+      cwd: process.cwd(),
+      env: { [SUBPROCESS_CONTROL_ENV]: 'pipe' },
+      stdio: { stdin: 'ignore', stdout: 'pipe', stderr: 'pipe' },
+      graceMs: 1000,
+    })).toThrow('reserved')
+  })
+})

packages/subprocess/subprocess-local/tests/fixtures/control-child.ts

@@ -0,0 +1,21 @@
+/** A plain Node child that echoes binary control bytes independently of stdio. */
+
+import { pathToFileURL } from 'node:url'
+import type { Duplex } from 'node:stream'
+
+const helper = await import(pathToFileURL(process.argv[2] as string).href) as { openInheritedControlChannel(): Duplex }
+const channel = helper.openInheritedControlChannel()
+const size = Number(process.argv[3])
+process.stdout.write('ordinary stdout\n')
+process.stderr.write('ordinary stderr\n')
+const chunks: Buffer[] = []
+let received = 0
+channel.on('error', (error) => { throw error })
+channel.on('data', (chunk: Buffer) => {
+  chunks.push(chunk)
+  received += chunk.length
+  if (received === size) {
+    channel.removeAllListeners('data')
+    channel.end(Buffer.concat(chunks), () => { channel.destroy() })
+  }
+})

packages/subprocess/subprocess/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@deepseek-ai/dsh-subprocess",
-  "description": "Subprocess seam (ctx.subprocess) for the DeepSeek Harness — managed process groups, bounded spill-backed output, and escalated kills behind one abstract service",
+  "description": "Subprocess seam (ctx.subprocess) for the DeepSeek Harness \u2014 managed process groups, bounded spill-backed output, and escalated kills behind one abstract service",
   "version": "0.1.5-rc.2",
   "publishConfig": {
     "access": "public"
@@ -19,10 +19,15 @@
       "default": "./lib/index.js"
     },
     "./src/*": "./src/*",
-    "./package.json": "./package.json"
+    "./package.json": "./package.json",
+    "./control": {
+      "types": "./lib/types/control.d.ts",
+      "default": "./lib/control.js"
+    }
   },
   "files": [
     "lib/index.js",
+    "lib/control.js",
     "lib/types/**/*.d.ts"
   ],
   "license": "MIT",

packages/subprocess/subprocess/src/control.ts

@@ -0,0 +1,24 @@
+/** Inherited byte-channel protocol shared by subprocess launchers and Node children. */
+
+import { Socket } from 'node:net'
+import type { Duplex } from 'node:stream'
+
+/** Child descriptor reserved for the optional subprocess control channel. */
+export const SUBPROCESS_CONTROL_FD = 7
+
+/** Private launch marker consumed before a Node child executes application code. */
+export const SUBPROCESS_CONTROL_ENV = 'DSH_SUBPROCESS_CONTROL' as const
+
+/**
+ * Consume the launch marker and open the inherited control pipe at fd 7.
+ * The returned stream owns the descriptor. Call once before executing untrusted code;
+ * messages remain untrusted even though the endpoint was inherited.
+ * @returns a connected byte-mode duplex stream owned by the caller.
+ * @throws when the marker is missing/invalid or the inherited descriptor cannot be opened.
+ */
+export function openInheritedControlChannel(): Duplex {
+  const marker = process.env[SUBPROCESS_CONTROL_ENV]
+  Reflect.deleteProperty(process.env, SUBPROCESS_CONTROL_ENV)
+  if (marker !== 'pipe') throw new Error('subprocess control channel was not inherited')
+  return new Socket({ fd: SUBPROCESS_CONTROL_FD, readable: true, writable: true, allowHalfOpen: true })
+}

packages/subprocess/subprocess/src/types.ts

@@ -7,7 +7,7 @@
  * @module dsh-subprocess/types
  */
 
-import type { Readable, Writable } from 'node:stream'
+import type { Duplex, Readable, Writable } from 'node:stream'
 
 /** Namespace prefix reserved for DeepSeek Harness-managed child environment facts. */
 export const DSH_ENV_PREFIX = 'DSH_' as const
@@ -64,6 +64,8 @@ export interface SubprocessStdio {
   stdin: SubprocessStdinMode
   stdout: SubprocessOutputMode
   stderr: SubprocessOutputMode
+  /** Request a separate byte-mode duplex channel; omission creates none. */
+  control?: 'pipe'
 }
 
 /**
@@ -171,6 +173,8 @@ export interface SubprocessHandle {
   readonly stdout: Readable | undefined
   /** The child's raw stderr, present iff spawned with `stderr: 'pipe'`. */
   readonly stderr: Readable | undefined
+  /** Separate caller-owned byte channel, present exactly when requested in stdio. */
+  readonly control: Duplex | undefined
   /** Offset-based readers for collect-mode streams (also readable after exit). */
   readonly collected: SubprocessCollectedOutputs
   /** Resolves with spawned-command exit facts; rejects for spawn or provider failures. */

packages/subprocess/subprocess/tests/service.spec.ts

@@ -26,6 +26,7 @@ class StubSubprocessRuntime extends SubprocessRuntime {
       ? { stdout: { readFrom: () => read } }
       : {}
     return {
+      control: undefined,
       stdin: undefined,
       stdout: undefined,
       stderr: undefined,

packages/subprocess/subprocess/tsdown.config.ts

@@ -0,0 +1,15 @@
+import { defineConfig } from 'tsdown'
+
+export default defineConfig({
+  entry: {
+    index: 'lib/types/index.js',
+    control: 'lib/types/control.js',
+  },
+  outDir: 'lib',
+  format: ['esm'],
+  platform: 'node',
+  target: 'es2024',
+  fixedExtension: false,
+  dts: false,
+  clean: false,
+})

packages/subprocess/win32-process/src/control-stdio.ts

@@ -0,0 +1,47 @@
+/** Windows CRT startup descriptors for a Node payload with one inherited control pipe. */
+
+import type { NativePtr, Win32ProcessBindings } from './ffi.ts'
+
+/** Native stdio handles plus the single additional descriptor selected by the launcher. */
+export interface InheritedControlStdio {
+  stdin: NativePtr
+  stdout: NativePtr
+  stderr: NativePtr
+  control: { fileDescriptor: 7; handle: NativePtr }
+}
+
+const HANDLE_BYTES = 8
+const INVALID_HANDLE = 0xffff_ffff_ffff_ffffn
+const FOPEN = 0x01
+const FPIPE = 0x08
+const FDEV = 0x40
+const FILE_TYPE_CHAR = 2
+const FILE_TYPE_PIPE = 3
+
+/**
+ * Encode the CRT's descriptor table before the child runtime allocates descriptors.
+ * Supported Windows targets use 64-bit handles. Empty slots remain closed, and the
+ * backing Buffer must remain alive until CreateProcess returns.
+ * @param api - native file-type inspection for inherited handles.
+ * @param stdio - standard handles and the provider-owned fd-7 control pipe.
+ * @returns descriptor count, flag bytes, and handle values for STARTUPINFO's reserved CRT fields.
+ */
+export function inheritedControlStdio(api: Pick<Win32ProcessBindings, 'getFileType'>, stdio: InheritedControlStdio): Buffer {
+  const count = stdio.control.fileDescriptor + 1
+  const handleOffset = 4 + count
+  const bytes = Buffer.alloc(handleOffset + count * HANDLE_BYTES)
+  bytes.writeUInt32LE(count, 0)
+  for (let index = 0; index < count; index++) {
+    bytes.writeBigUInt64LE(INVALID_HANDLE, handleOffset + index * HANDLE_BYTES)
+  }
+  const entries = [[0, stdio.stdin], [1, stdio.stdout], [2, stdio.stderr], [stdio.control.fileDescriptor, stdio.control.handle]] as const
+  for (const [fd, handle] of entries) {
+    const kind = api.getFileType(handle)
+    if (fd === stdio.control.fileDescriptor && kind !== FILE_TYPE_PIPE) {
+      throw new Error('subprocess control descriptor is not a Windows pipe')
+    }
+    bytes[4 + fd] = FOPEN | (kind === FILE_TYPE_PIPE ? FPIPE : kind === FILE_TYPE_CHAR ? FDEV : 0)
+    bytes.writeBigUInt64LE(handle, handleOffset + fd * HANDLE_BYTES)
+  }
+  return bytes
+}

packages/subprocess/win32-process/src/ffi.ts

@@ -43,6 +43,8 @@ export interface StartupInfoInput {
   hStdInput: NativePtr
   hStdOutput: NativePtr
   hStdError: NativePtr
+  cbReserved2?: number
+  lpReserved2?: NativePtr
 }
 
 /** Decoded PROCESS_INFORMATION result. */
@@ -57,6 +59,8 @@ export interface ProcessInfoOutput {
 export interface Win32ProcessBindings {
   closeHandle(handle: NativePtr): number
   getLastError(): number
+  getFileType(handle: NativePtr): number
+  uvGetOsfhandle(fileDescriptor: number): NativePtr | null
   formatMessageW(
     flags: number,
     source: null,
@@ -258,6 +262,7 @@ function bindings(): CurrentTokenProcessBindings {
   cached = {
     closeHandle: bind(kernel32, 'CloseHandle', 'int', [PVOID]),
     getLastError: bind(kernel32, 'GetLastError', 'uint32', []),
+    getFileType: bind(kernel32, 'GetFileType', 'uint32', [PVOID]),
     formatMessageW: bind(kernel32, 'FormatMessageW', 'uint32', [
       'uint32', PVOID, 'uint32', 'uint32', PVOID, 'uint32', PVOID,
     ]),

packages/subprocess/win32-process/src/process.ts

@@ -2,6 +2,7 @@
 
 import koffi from 'koffi'
 import * as abi from './abi.ts'
+import { inheritedControlStdio } from './control-stdio.ts'
 import {
   allocProcessInfo,
   allocPtrSlot,
@@ -92,12 +93,16 @@ export interface CurrentTokenStdioFileDescriptors {
   stdin: number
   stdout: number
   stderr: number
+  /** Optional carrier and target descriptor for the inherited control pipe. */
+  control?: 7
 }
 
 /** Restricted-token process creation inputs owned by the Windows ACL sandbox. */
 export interface RestrictedProcessSpawnOptions extends ProcessSpawnOptions {
   /** Restricted primary token supplied by sandbox policy. */
   token: NativePtr
+  /** Optional control pipe inherited at the same descriptor in the payload. */
+  controlFileDescriptor?: 7
 }
 
 /** Piped child resources whose process and read handles remain caller-owned. */
@@ -355,13 +360,14 @@ interface ProcessStandardHandles {
   stdin: NativePtr
   stdout: NativePtr
   stderr: NativePtr
+  control?: { fileDescriptor: 7; handle: NativePtr }
 }
 
 // Koffi exposes PVOID as an unsigned 64-bit bigint on supported Windows hosts.
 const UV_INVALID_OS_FILE_HANDLE = 0xffff_ffff_ffff_ffffn
 const UV_INVALID_FILE_DESCRIPTOR = 0xffff_ffff_ffff_fffen
 
-function inheritedStandardHandles(api: Win32ProcessBindings): ProcessStandardHandles {
+function inheritedStandardHandles(api: Win32ProcessBindings, controlFileDescriptor?: 7): ProcessStandardHandles {
   const get = (selector: number, label: string): NativePtr => {
     const handle = api.getStdHandle(selector)
     if (!isNullPtr(handle)) return handle
@@ -371,28 +377,35 @@ function inheritedStandardHandles(api: Win32ProcessBindings): ProcessStandardHan
     stdin: get(abi.STD_INPUT_HANDLE, 'stdin'),
     stdout: get(abi.STD_OUTPUT_HANDLE, 'stdout'),
     stderr: get(abi.STD_ERROR_HANDLE, 'stderr'),
+    ...controlFileDescriptor === undefined ? {} : {
+      control: { fileDescriptor: controlFileDescriptor, handle: descriptorHandle(api, controlFileDescriptor, 'control') },
+    },
   }
 }
 
+function descriptorHandle(api: Win32ProcessBindings, fileDescriptor: number, label: string): NativePtr {
+  const handle = api.uvGetOsfhandle(fileDescriptor)
+  if (
+    isNullPtr(handle)
+      || handle === UV_INVALID_OS_FILE_HANDLE
+      || handle === UV_INVALID_FILE_DESCRIPTOR
+  ) {
+    throw new Error(`uv_get_osfhandle returned an invalid handle for target ${label} fd ${String(fileDescriptor)}`)
+  }
+  return handle
+}
+
 function targetCarrierHandles(
   api: CurrentTokenProcessBindings,
   descriptors: CurrentTokenStdioFileDescriptors,
 ): ProcessStandardHandles {
-  const get = (fileDescriptor: number, label: string): NativePtr => {
-    const handle = api.uvGetOsfhandle(fileDescriptor)
-    if (
-      isNullPtr(handle)
-      || handle === UV_INVALID_OS_FILE_HANDLE
-      || handle === UV_INVALID_FILE_DESCRIPTOR
-    ) {
-      throw new Error(`uv_get_osfhandle returned an invalid handle for target ${label} fd ${String(fileDescriptor)}`)
-    }
-    return handle
-  }
   return {
-    stdin: get(descriptors.stdin, 'stdin'),
-    stdout: get(descriptors.stdout, 'stdout'),
-    stderr: get(descriptors.stderr, 'stderr'),
+    stdin: descriptorHandle(api, descriptors.stdin, 'stdin'),
+    stdout: descriptorHandle(api, descriptors.stdout, 'stdout'),
+    stderr: descriptorHandle(api, descriptors.stderr, 'stderr'),
+    ...descriptors.control === undefined ? {} : {
+      control: { fileDescriptor: descriptors.control, handle: descriptorHandle(api, descriptors.control, 'control') },
+    },
   }
 }
 
@@ -408,20 +421,30 @@ function spawnJobProcess(
   const enabled: NativePtr[] = []
   let startupInfo: NativePtr | undefined
   let processInfo: NativePtr | undefined
+  let controlDescriptorBlock: { pointer: NativePtr; length: number } | undefined
   let created = 0
   let createFailureCode = 0
   try {
     const stdio = resolveStdio()
-    for (const [handle, label] of [
+    const inherited: Array<readonly [NativePtr, string]> = [
       [stdio.stdin, 'stdin'],
       [stdio.stdout, 'stdout'],
       [stdio.stderr, 'stderr'],
-    ] as const) {
+    ]
+    if (stdio.control !== undefined) inherited.push([stdio.control.handle, 'control'])
+    for (const [handle, label] of inherited) {
       if (api.setHandleInformation(handle, abi.HANDLE_FLAG_INHERIT, abi.HANDLE_FLAG_INHERIT) === 0) {
         throwLastError(api, 'SetHandleInformation', `${label} (enable inherit)`)
       }
       enabled.push(handle)
     }
+    const controlBytes = stdio.control === undefined
+      ? undefined
+      : inheritedControlStdio(api, { ...stdio, control: stdio.control })
+    if (controlBytes !== undefined) {
+      controlDescriptorBlock = { pointer: koffi.alloc('uint8', controlBytes.length) as NativePtr, length: controlBytes.length }
+      koffi.encode(controlDescriptorBlock.pointer, 'uint8', controlBytes, controlBytes.length)
+    }
     startupInfo = allocStartupInfo()
     encodeStartupInfo(startupInfo, {
       cb: abi.STARTUPINFOW_SIZE,
@@ -429,6 +452,10 @@ function spawnJobProcess(
       hStdInput: stdio.stdin,
       hStdOutput: stdio.stdout,
       hStdError: stdio.stderr,
+      ...controlDescriptorBlock === undefined ? {} : {
+        cbReserved2: controlDescriptorBlock.length,
+        lpReserved2: controlDescriptorBlock.pointer,
+      },
     })
     processInfo = allocProcessInfo()
     created = create(startupInfo, processInfo)
@@ -439,6 +466,7 @@ function spawnJobProcess(
     throw error
   } finally {
     freeNative(startupInfo)
+    freeNative(controlDescriptorBlock?.pointer)
     for (const handle of enabled) {
       // The runner spawns nothing else; cleanup failure must not mask the child.
       api.setHandleInformation(handle, abi.HANDLE_FLAG_INHERIT, 0)
@@ -501,7 +529,7 @@ export function spawnInheritedJobProcess(
   options: RestrictedProcessSpawnOptions,
 ): SpawnedJobProcess {
   const commandLine = buildCommandLine(options.command, options.args)
-  return spawnJobProcess(api, options, () => inheritedStandardHandles(api), 'CreateProcessAsUserW', (startupInfo, processInfo) =>
+  return spawnJobProcess(api, options, () => inheritedStandardHandles(api, options.controlFileDescriptor), 'CreateProcessAsUserW', (startupInfo, processInfo) =>
     createRestrictedProcess(
       api,
       options,

packages/subprocess/win32-process/tests/ordinary-process.spec.ts

@@ -75,6 +75,41 @@ function api(overrides: Partial<CurrentTokenProcessBindings> = {}): CurrentToken
 }
 
 describe('ordinary Job process operations', () => {
+  it('supplies fd 7 in the child CRT startup table and releases its temporary inheritance', () => {
+    let descriptorBytes: Buffer | undefined
+    const flags = vi.fn(() => 1)
+    const bindings = api({
+      getFileType: vi.fn(() => 3),
+      setHandleInformation: flags,
+      createProcessW: vi.fn((_app, _line, _pa, _ta, _inherit, _flags, _env, _cwd, startupPointer, processInfo) => {
+        const startup = koffi.decode(startupPointer, STARTUPINFOW) as { cbReserved2: number; lpReserved2: NativePtr }
+        descriptorBytes = Buffer.from(koffi.decode(startup.lpReserved2, 'uint8', startup.cbReserved2) as number[])
+        koffi.encode(processInfo, PROCESS_INFORMATION, { hProcess: 60n, hThread: 61n, dwProcessId: 1234, dwThreadId: 5678 })
+        return 1
+      }),
+    })
+    expect(spawnCurrentTokenJobProcess(bindings, options({
+      stdio: { stdin: 4, stdout: 5, stderr: 6, control: 7 },
+    }))).toEqual({ pid: 1234, process: 60n, job: 50n })
+    const bytes = descriptorBytes as Buffer
+    expect(bytes.readUInt32LE(0)).toBe(8)
+    expect([...bytes.subarray(4, 12)]).toEqual([9, 9, 9, 0, 0, 0, 0, 9])
+    expect(bytes.readBigUInt64LE(12 + 7 * 8)).toBe(107n)
+    expect(bytes.readBigUInt64LE(12 + 3 * 8)).toBe(0xffff_ffff_ffff_ffffn)
+    expect(flags).toHaveBeenCalledWith(107n, 1, 1)
+    expect(flags).toHaveBeenCalledWith(107n, 1, 0)
+  })
+
+  it('refuses a control carrier that is not a pipe before creating the target', () => {
+    const bindings = api({ getFileType: vi.fn(() => 1) })
+    expect(() => spawnCurrentTokenJobProcess(bindings, options({
+      stdio: { stdin: 4, stdout: 5, stderr: 6, control: 7 },
+    }))).toThrow('not a Windows pipe')
+    expect(bindings.createProcessW).not.toHaveBeenCalled()
+    expect(bindings.closeHandle).toHaveBeenCalledWith(50n)
+    expect(bindings.setHandleInformation).toHaveBeenCalledWith(107n, 1, 0)
+  })
+
   it('creates suspended, assigns the Job, and resumes before returning', () => {
     const events: string[] = []
     const createProcessW = vi.fn((

pnpm-lock.yaml

@@ -7597,6 +7597,9 @@ importers:
       '@deepseek-ai/dsh-sandbox-local':
         specifier: workspace:^
         version: link:../sandbox-local
+      '@deepseek-ai/dsh-subprocess':
+        specifier: workspace:^
+        version: link:../../subprocess/subprocess
 
   packages/schedule/schedule:
     dependencies:

tsconfig.base.json

@@ -401,6 +401,7 @@
       "@deepseek-ai/dsh-subagent-in-process-driver": ["./packages/subagent/subagent-in-process-driver/src"],
       "@deepseek-ai/dsh-subagent-spawn-in-process": ["./packages/subagent/subagent-spawn-in-process/src"],
       "@deepseek-ai/dsh-subprocess": ["./packages/subprocess/subprocess/src"],
+      "@deepseek-ai/dsh-subprocess/control": ["./packages/subprocess/subprocess/src/control.ts"],
       "@deepseek-ai/dsh-subprocess-local": ["./packages/subprocess/subprocess-local/src"],
       "@deepseek-ai/dsh-system-prompt": ["./packages/core/system-prompt/src"],
       "@deepseek-ai/dsh-system-prompt/invariant": ["./packages/core/system-prompt/src/invariant.ts"],