feat(subprocess): reshape the seam Node-ward for multi-consumer use

Review direction (tianyicui, PR #660): make the interface closer to Node's
API so the other process-running places can adopt it. The spec gains
per-stream stdio dispositions — 'pipe' (raw Readable/Writable for protocol
streams), 'inherit' (diagnostics to the parent), and collect mode ({maxBytes,
spill?} — the old bounded tail-keep shape, now with spill optional for
diagnostic tails). SubprocessOutcome carries exit facts only; collected
output stays readable through handle.collected after settlement (spill fds
are sealed at the settle boundary). The handle grows Node-style kill(signal)
(single signal, tree-scoped, no-op after settlement), terminate() (the
SIGTERM→grace→SIGKILL escalation, also driven by the spec signal),
waitForExit() (tree liveness, not just the direct child), and dispose()
(the cooperative stdin-EOF→SIGTERM→SIGKILL ladder from subagent-subprocess,
graces caller-supplied). Tree semantics are platform-correct: POSIX detached
groups with direct-child fallback; Windows taskkill /T with an injectable
runner. scrubbedParentEnv/SENSITIVE_ENV_PATTERN move to the seam as the one
shared scrub definition.

bash-local maps its config onto collect modes and batch stdin and reads
results through the collected readers; its kill() maps to terminate() so
task_kill keeps escalation semantics.

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

@@ -11,8 +11,8 @@
 import { Context } from 'cordis'
 import z from 'schemastery'
 import { BashExecutor } from '@deepseek-ai/dsh-bash'
-import type { BashExecRequest, BashExecSpec, BashProcess, BashProcessRead, BashRunResult } from '@deepseek-ai/dsh-bash'
-import type { SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
+import type { BashExecRequest, BashExecSpec, BashProcess, BashProcessRead, BashRunResult, CollectedOutput } from '@deepseek-ai/dsh-bash'
+import type { SubprocessCollect, SubprocessHandle, SubprocessOutputReader, SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
 import { clampTimeout, deadline, timeoutOf } from '@deepseek-ai/dsh-timeout'
 
 /**
@@ -54,6 +54,16 @@ export interface Config {
 /** The shape after schemastery applied the defaults (cwd has none). */
 type ResolvedConfig = Required<Omit<Config, 'cwd'>> & Pick<Config, 'cwd'>
 
+/** Project a settled collect-mode reader into the final CollectedOutput shape. */
+function finalOutput(reader: SubprocessOutputReader): CollectedOutput {
+  const read = reader.readFrom(0)
+  return {
+    text: read.text,
+    truncated: read.lossy,
+    ...read.spillPath !== undefined ? { spillPath: read.spillPath } : {},
+  }
+}
+
 function assertPositiveFinite(name: string, value: number): void {
   if (!Number.isFinite(value) || value <= 0) {
     throw new Error(`bash-local: ${name} must be a positive finite number`)
@@ -127,36 +137,58 @@ export class LocalBashExecutor extends BashExecutor {
     }
   }
 
-  /** Map one resolved bash spec onto a fully-specified process spawn. */
+  /** Map one resolved bash spec onto a fully-specified subprocess spawn. */
   // XXX(stateful-shell): evaluate persistent cwd or PTY sessions when workflows require shell state.
   private spawnSpec(spec: BashExecSpec, stdoutMaxBytes: number, signal: AbortSignal | undefined): SubprocessSpawnSpec {
+    const collect = (maxBytes: number): SubprocessCollect =>
+      ({ maxBytes, spill: { maxBytes: this.config.maxSpillBytes } })
     return {
       argv: ['bash', '-c', spec.command],
       cwd: spec.workdir,
-      stdoutMaxBytes,
-      stderrMaxBytes: this.config.maxOutputBytes,
-      maxSpillBytes: this.config.maxSpillBytes,
+      stdio: {
+        stdin: spec.stdin !== undefined ? { data: spec.stdin } : 'ignore',
+        stdout: collect(stdoutMaxBytes),
+        stderr: collect(this.config.maxOutputBytes),
+      },
       graceMs: this.config.graceMs,
       signal,
-      stdin: spec.stdin,
       env: { ...ENV_OVERRIDES, ...spec.env },
       dshEnv: spec.dshEnv,
     }
   }
 
+  /** The collect-mode readers the executor itself requested (present by construction). */
+  private static collected(handle: SubprocessHandle): { stdout: SubprocessOutputReader; stderr: SubprocessOutputReader } {
+    const { stdout, stderr } = handle.collected
+    if (stdout === undefined || stderr === undefined) {
+      throw new Error('bash-local: subprocess implementation dropped a requested collect stream')
+    }
+    return { stdout, stderr }
+  }
+
   async run(spec: BashExecSpec): Promise<BashRunResult> {
     // One deadline combines timeout and upstream cancellation; disposal clears its timer.
     using d = deadline(spec.signal, spec.timeoutMs, 'BASH_TIMEOUT')
-    const outcome = await this.ctx.subprocess.spawn(this.spawnSpec(spec, spec.stdoutMaxBytes, d.signal)).done
+    const handle = this.ctx.subprocess.spawn(this.spawnSpec(spec, spec.stdoutMaxBytes, d.signal))
+    const outcome = await handle.done
+    const collected = LocalBashExecutor.collected(handle)
     // Only this executor's timeout reason counts as timedOut; outer deadlines count as aborts.
     const timedOut = timeoutOf(d.signal, 'BASH_TIMEOUT') !== undefined
     const aborted = d.signal.aborted && !timedOut
-    return { ...outcome, timedOut, aborted, timeoutMs: spec.timeoutMs }
+    return {
+      ...outcome,
+      timedOut,
+      aborted,
+      timeoutMs: spec.timeoutMs,
+      stdout: finalOutput(collected.stdout),
+      stderr: finalOutput(collected.stderr),
+    }
   }
 
   start(spec: BashExecSpec): BashProcess {
     // Background runs ignore timeoutMs; callers stop them through kill() or spec.signal.
     const running = this.ctx.subprocess.spawn(this.spawnSpec(spec, this.config.maxOutputBytes, spec.signal))
+    const collected = LocalBashExecutor.collected(running)
 
     // A spawn failure produces no process output, so the subprocess service has nothing
     // to buffer; the note is delivered exactly once through the read path.
@@ -180,7 +212,7 @@ export class LocalBashExecutor extends BashExecutor {
         }
         proc.exitCode = outcome.exitCode
         proc.signal = outcome.signal
-        this.onProcessDone(proc, running.stderr.readFrom(0).text)
+        this.onProcessDone(proc, collected.stderr.readFrom(0).text)
       }, (error: unknown) => {
         // Background spawn failures settle as killed and surface through the read path.
         proc.status = 'killed'
@@ -188,8 +220,8 @@ export class LocalBashExecutor extends BashExecutor {
         this.onProcessDone(proc, spawnFailureNote)
       }),
       readOutput: (): BashProcessRead => {
-        const out = running.stdout.readFrom(stdoutOffset)
-        const err = running.stderr.readFrom(stderrOffset)
+        const out = collected.stdout.readFrom(stdoutOffset)
+        const err = collected.stderr.readFrom(stderrOffset)
         stdoutOffset = out.nextOffset
         stderrOffset = err.nextOffset
 
@@ -211,7 +243,7 @@ export class LocalBashExecutor extends BashExecutor {
       kill: (): boolean => {
         if (proc.status !== 'running') return false
         proc.status = 'killed'
-        running.kill()
+        running.terminate()
         return true
       },
     }

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

@@ -1,36 +1,38 @@
 /**
- * Local-subprocess implementation of the subprocess seam. Each spawn is
- * a detached process group with bounded, spill-backed output; disposal kills
- * and joins live groups. It has no config: every limit arrives on the spec,
- * so the deployment-varying choices stay with the calling seam's config (the
- * bash executor's, today).
+ * Local implementation of the subprocess seam. Each spawn is a detached
+ * process tree with the spec's per-stream stdio dispositions; disposal
+ * terminates and joins live trees. It has no config: every disposition and
+ * limit arrives on the spec, so the deployment-varying choices stay with the
+ * calling seam's config (the bash executor's, the LSP host's, …).
  * @module @deepseek-ai/dsh-subprocess-local
  */
 
 import { Context } from 'cordis'
 import { SubprocessService } from '@deepseek-ai/dsh-subprocess'
 import type { SubprocessHandle, SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
-import { spawnProcess } from './spawn.ts'
+import { spawnSubprocess } from './spawn.ts'
 import type { SpawnInternals } from './spawn.ts'
 
 /**
- * Local subprocess service: detached process groups, tail-keep truncation with
- * bounded spill files, credential-scrubbed environment, and group
- * SIGTERM→grace→SIGKILL escalation.
+ * Local subprocess service: detached process trees, Node-shaped stdio
+ * dispositions (raw pipes, inherit, bounded tail-keep collection with spill
+ * files), credential-scrubbed environment, tree-scoped signalling with
+ * SIGTERM→grace→SIGKILL escalation, and the cooperative dispose ladder.
  */
 export class LocalSubprocessService extends SubprocessService {
-  /** Live handles retained only so disposal can kill and join them. */
+  /** Live handles retained only so disposal can terminate and join them. */
   private live = new Set<SubprocessHandle>()
-  /** Test seam: spill knobs forwarded to spawnProcess. */
+  /** Test seam: spill and platform knobs forwarded to spawnSubprocess. */
   internals: SpawnInternals = {}
 
   constructor(ctx: Context) {
     super(ctx)
     ctx.effect(() => async () => {
-      // Await closure so even a TERM-trapping child cannot outlive the fiber.
+      // Terminate (escalating), then await closure so even a TERM-trapping
+      // child cannot outlive the fiber.
       const pending: Promise<unknown>[] = []
       for (const handle of this.live) {
-        handle.kill()
+        handle.terminate()
         // Spawn-failure rejections already settled and left the live set.
         pending.push(handle.done.catch(() => {}))
       }
@@ -40,7 +42,7 @@ export class LocalSubprocessService extends SubprocessService {
   }
 
   spawn(spec: SubprocessSpawnSpec): SubprocessHandle {
-    const handle = spawnProcess(spec, this.internals)
+    const handle = spawnSubprocess(spec, this.internals)
     this.live.add(handle)
     handle.done.then(
       () => { this.live.delete(handle) },

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

@@ -1,33 +1,36 @@
 /**
- * Process plumbing for the local subprocess service: detached process-group
- * spawn, tail-keep output with spill files, and SIGTERM→SIGKILL escalation.
+ * Process plumbing for the local subprocess service: detached process-tree
+ * spawn with per-stream stdio dispositions, tail-keep collection with spill
+ * files, tree-scoped signalling (POSIX groups; Windows taskkill), the
+ * SIGTERM→SIGKILL escalation, and the cooperative EOF-first dispose ladder.
  * This layer reacts to an abort signal; callers own deadlines and classify
  * causes.
  * @module dsh-subprocess-local/spawn
  */
 
-import { type ChildProcessByStdio, spawn } from 'node:child_process'
-import type { Readable, Writable } from 'node:stream'
+import { type ChildProcess, spawn, spawnSync } from 'node:child_process'
+import type { Readable } from 'node:stream'
 import { randomBytes } from 'node:crypto'
 import { closeSync, mkdtempSync, openSync, unlinkSync, writeSync } from 'node:fs'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
-import { DSH_ENV_PREFIX } from '@deepseek-ai/dsh-subprocess'
-import type { CollectedOutput, DshEnvironment, SubprocessHandle, SubprocessOutcome, SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
+import { setImmediate as yieldToEventLoop } from 'node:timers/promises'
+import { DSH_ENV_PREFIX, scrubbedParentEnv } from '@deepseek-ai/dsh-subprocess'
+import type {
+  CollectedOutput,
+  DshEnvironment,
+  SubprocessCollect,
+  SubprocessDisposeGraces,
+  SubprocessHandle,
+  SubprocessOutcome,
+  SubprocessOutputMode,
+  SubprocessSpawnSpec,
+} from '@deepseek-ai/dsh-subprocess'
 
 /**
- * Credential-shaped env vars are NOT forwarded to children (the harness's
- * own DEEPSEEK_API_KEY must not leak into `env` output, tool results, or
- * spill files). Same default pattern as Codex's env policy; a future config
- * can whitelist specific vars when a workflow genuinely needs one.
- */
-export const SENSITIVE_ENV_PATTERN = /KEY|SECRET|TOKEN/i
-
-/**
- * Build a child environment from scrubbed ambient values, ordinary caller
- * entries, and a managed `DSH_*` snapshot. Ambient managed names are removed;
- * ordinary and managed entries reject the other channel's namespace before
- * `dshEnv` merges last.
+ * Build a child environment from the scrubbed parent base, ordinary caller
+ * entries, and a managed `DSH_*` snapshot. Ordinary and managed entries
+ * reject the other channel's namespace before `dshEnv` merges last.
  * @param extra - caller entries; `DSH_*` names are rejected.
  * @param dshEnv - managed entries; non-`DSH_*` names are rejected.
  * @returns the environment to hand to `spawn` for the child process.
@@ -36,10 +39,6 @@ export function childEnv(
   extra?: Readonly<Record<string, string>>,
   dshEnv?: DshEnvironment,
 ): NodeJS.ProcessEnv {
-  const env: NodeJS.ProcessEnv = {}
-  for (const [key, value] of Object.entries(process.env)) {
-    if (!SENSITIVE_ENV_PATTERN.test(key) && !key.startsWith(DSH_ENV_PREFIX)) env[key] = value
-  }
   for (const key of Object.keys(extra ?? {})) {
     if (key.startsWith(DSH_ENV_PREFIX)) {
       throw new Error(`ordinary child env cannot set reserved variable "${key}"; use dshEnv`)
@@ -50,13 +49,17 @@ export function childEnv(
       throw new Error(`managed child env cannot set ordinary variable "${key}"; use env`)
     }
   }
-  return { ...env, ...extra, ...dshEnv }
+  return { ...scrubbedParentEnv(), ...extra, ...dshEnv }
 }
 
-/** Injectable knobs so tests can exercise spill behavior without the OS tmpdir. */
+/** Injectable knobs so tests can exercise spill and platform behavior deterministically. */
 export interface SpawnInternals {
   /** Directory for spill files (defaults to the OS temp dir). */
   spillDir?: string
+  /** Windows tree-termination runner (defaults to `taskkill /PID <pid> /T /F`). */
+  taskkill?: (pid: number) => void
+  /** Host platform override for signalling decisions. */
+  platform?: NodeJS.Platform
 }
 
 let spillCounter = 0
@@ -73,9 +76,11 @@ function privateSpillDir(): string {
 }
 
 /**
- * Collects one stream with a bounded in-memory tail. On first overflow a
- * spill file is created and every chunk (including those already collected)
- * is appended there while the full stream remains within `maxSpillBytes`.
+ * Collects one stream with a bounded in-memory tail. With a spill cap, on
+ * first overflow a spill file is created and every chunk (including those
+ * already collected) is appended there while the full stream remains within
+ * the cap; without one, only the in-memory tail is ever retained (the
+ * diagnostic-tail shape — a language server's stderr).
  *
  * Tail-keep rationale (pi/OpenCode): errors and final results cluster at the
  * end of command output; the spill file covers the head.
@@ -86,23 +91,25 @@ export class OutputCollector {
   private dropped = false
   private spillFd: number | undefined
   private spillFile: string | undefined
-  private spillDisabled = false
+  private spillDisabled: boolean
   /** Total bytes ever pushed (not just retained). */
   private total = 0
 
   constructor(
     private readonly maxBytes: number,
-    private readonly maxSpillBytes: number,
+    private readonly maxSpillBytes: number | undefined,
     private readonly label: string,
     private readonly spillDir: string,
-  ) {}
+  ) {
+    this.spillDisabled = maxSpillBytes === undefined
+  }
 
   /**
    * Ingest one stream chunk, counting it toward the whole-stream total. On
-   * first overflow of the in-memory cap a spill file is opened and every chunk
-   * (already-collected ones included) is appended there from then on; the
-   * in-memory tail then drops whole chunks from its head (or the head of a
-   * single over-cap chunk) until it fits the cap again.
+   * first overflow of the in-memory cap a spill file is opened (when spilling
+   * is enabled) and every chunk (already-collected ones included) is appended
+   * there from then on; the in-memory tail then drops whole chunks from its
+   * head (or the head of a single over-cap chunk) until it fits the cap again.
    * @param chunk - the raw bytes from one stream 'data' event.
    */
   push(chunk: Buffer): void {
@@ -130,7 +137,7 @@ export class OutputCollector {
 
   /** Open the spill file lazily and append `chunk` (and any prior chunks once). */
   private spillAll(chunk: Buffer): void {
-    if (this.total > this.maxSpillBytes) {
+    if (this.maxSpillBytes !== undefined && this.total > this.maxSpillBytes) {
       this.discardSpill()
       return
     }
@@ -194,22 +201,30 @@ export class OutputCollector {
   }
 
   /**
-   * Close the spill file (if any) and return the final output. A failed close
-   * (delayed writeback fault) stops advertising the spill path — the file may
-   * be missing its tail — but still returns the in-memory result.
+   * Close the spill file once the stream has ended. A failed close (delayed
+   * writeback fault) stops advertising the spill path — the file may be
+   * missing its tail — while every in-memory read keeps working. Idempotent;
+   * the spawn path seals both collectors at settlement so reads after exit
+   * never point at a still-open file.
+   */
+  seal(): void {
+    if (this.spillFd === undefined) return
+    try {
+      closeSync(this.spillFd)
+    } catch {
+      // A delayed writeback failure makes the spill unreliable; keep the
+      // in-memory result but stop advertising that file.
+      this.spillFile = undefined
+    }
+    this.spillFd = undefined
+  }
+
+  /**
+   * Seal the spill file and return the final output.
    * @returns the final collected output: tail text, truncation flag, and the spill path when intact.
    */
   finalize(): CollectedOutput {
-    if (this.spillFd !== undefined) {
-      try {
-        closeSync(this.spillFd)
-      } catch {
-        // A delayed writeback failure makes the spill unreliable; keep finalize
-        // total but stop advertising that file.
-        this.spillFile = undefined
-      }
-      this.spillFd = undefined
-    }
+    this.seal()
     return {
       text: Buffer.concat(this.chunks).toString('utf8'),
       truncated: this.dropped,
@@ -219,9 +234,9 @@ export class OutputCollector {
 }
 
 /**
- * Send `sig` to a detached process group. Never throws: delivery races process
- * exit and may run in a timer callback, so failures are contained and a
- * non-positive pid is a no-op.
+ * Send `sig` to a detached POSIX process group. Never throws: delivery races
+ * process exit and may run in a timer callback, so failures are contained and
+ * a non-positive pid is a no-op.
  * @param pid - the group leader's pid; non-positive means the spawn failed and the call is a no-op.
  * @param sig - the signal to deliver to the whole group.
  */
@@ -235,14 +250,60 @@ export function killGroup(pid: number, sig: NodeJS.Signals): void {
 }
 
 /**
- * Spawn one isolated detached process group and collect its output.
- * Runtime exits resolve as {@link SubprocessOutcome}; only spawn failures reject.
- * @param spec - fully resolved argv, cwd, limits, and cancellation.
- * @param internals - test-only spill-directory override.
- * @returns live process handle and outcome promise.
+ * Terminate one Windows process tree with `taskkill /T /F`. Contained like
+ * POSIX group signalling — delivery races tree exit, so an absent tree, a
+ * nonzero status, or a missing taskkill binary must not break idempotent
+ * teardown.
+ * @param pid - root process id; non-positive is a no-op.
+ */
+export function taskkillProcessTree(pid: number): void {
+  if (pid <= 0) return
+  // Outcome deliberately unchecked: an already-absent tree (status 128) and
+  // exit races are as tolerable here as ESRCH is for a POSIX group signal.
+  spawnSync('taskkill', ['/PID', String(pid), '/T', '/F'], { stdio: 'ignore' })
+}
+
+/**
+ * Signal a detached process tree with platform-correct semantics: POSIX
+ * signals the negative process-group id and falls back to the direct child
+ * when the group is gone; Windows terminates the tree via taskkill (any
+ * signal value force-terminates — Node maps signals to TerminateProcess).
+ */
+function signalTree(
+  platform: NodeJS.Platform,
+  pid: number,
+  sig: NodeJS.Signals,
+  child: ChildProcess,
+  taskkill: (pid: number) => void,
+): void {
+  if (platform === 'win32') {
+    taskkill(pid)
+    return
+  }
+  if (pid <= 0) return
+  try {
+    process.kill(-pid, sig)
+  } catch {
+    try {
+      child.kill(sig)
+    } catch {
+      // The direct child already exited; teardown remains idempotent.
+    }
+  }
+}
+
+/**
+ * Spawn one isolated detached process tree with the spec's per-stream stdio
+ * dispositions. Runtime exits resolve `done` as {@link SubprocessOutcome};
+ * only spawn failures reject.
+ * @param spec - fully resolved argv, cwd, stdio, grace, cancellation, environment.
+ * @param internals - test-only spill-directory, platform, and taskkill overrides.
+ * @returns live subprocess handle.
  */
-export function spawnProcess(spec: SubprocessSpawnSpec, internals: SpawnInternals = {}): SubprocessHandle {
+export function spawnSubprocess(spec: SubprocessSpawnSpec, internals: SpawnInternals = {}): SubprocessHandle {
   const spillDir = internals.spillDir ?? privateSpillDir()
+  const platform = internals.platform ?? process.platform
+  const taskkill = internals.taskkill ?? taskkillProcessTree
 
   if (spec.signal?.aborted) {
     throw new Error(`aborted before spawn: ${String(spec.signal.reason ?? 'aborted')}`)
@@ -252,41 +313,66 @@ export function spawnProcess(spec: SubprocessSpawnSpec, internals: SpawnInternal
     throw new Error('invalid argv: expected a non-empty program name at argv[0]')
   }
 
-  // Keep absent stdin as /dev/null; literal tuples preserve non-null output types.
+  const isCollect = (mode: SubprocessOutputMode): mode is SubprocessCollect =>
+    mode !== 'pipe' && mode !== 'inherit'
+  const outMode = spec.stdio.stdout
+  const errMode = spec.stdio.stderr
+  const stdinMode = spec.stdio.stdin
+
   const env = childEnv(spec.env, spec.dshEnv)
-  const child: ChildProcessByStdio<Writable | null, Readable, Readable> = spec.stdin !== undefined
-    ? spawn(program, args, { cwd: spec.cwd, env, stdio: ['pipe', 'pipe', 'pipe'], detached: true })
-    : spawn(program, args, { cwd: spec.cwd, env, stdio: ['ignore', 'pipe', 'pipe'], detached: true })
+  const child = spawn(program, args, {
+    cwd: spec.cwd,
+    env,
+    stdio: [
+      stdinMode === 'ignore' ? 'ignore' : 'pipe',
+      outMode === 'inherit' ? 'inherit' : 'pipe',
+      errMode === 'inherit' ? 'inherit' : 'pipe',
+    ],
+    // `detached` gives teardown a tree root on POSIX (its own process group);
+    // Windows terminates by root pid through taskkill /T instead.
+    detached: platform !== 'win32',
+  })
 
-  const stdout = new OutputCollector(spec.stdoutMaxBytes, spec.maxSpillBytes, 'stdout', spillDir)
-  const stderr = new OutputCollector(spec.stderrMaxBytes, spec.maxSpillBytes, 'stderr', spillDir)
-  child.stdout.on('data', (chunk: Buffer) => { stdout.push(chunk) })
-  child.stderr.on('data', (chunk: Buffer) => { stderr.push(chunk) })
+  const collectStream = (mode: SubprocessOutputMode, stream: Readable | null, label: string): OutputCollector | undefined => {
+    if (!isCollect(mode) || stream === null) return undefined
+    const collector = new OutputCollector(mode.maxBytes, mode.spill?.maxBytes, label, spillDir)
+    stream.on('data', (chunk: Buffer) => { collector.push(chunk) })
+    return collector
+  }
+  const stdoutCollector = collectStream(outMode, child.stdout, 'stdout')
+  const stderrCollector = collectStream(errMode, child.stderr, 'stderr')
 
   let graceTimer: NodeJS.Timeout | undefined
   let settled = false
 
-  // Failed spawns use pid -1 so kill remains a no-op.
+  // Failed spawns use pid -1 so signalling remains a no-op.
   const pid = child.pid ?? -1
 
-  const kill = (): void => {
+  const kill = (sig: NodeJS.Signals = 'SIGTERM'): void => {
+    // After settlement the tree is gone and the pid may be reused; callers
+    // commonly kill() in a finally, so this must not re-signal.
+    if (settled) return
+    signalTree(platform, pid, sig, child, taskkill)
+  }
+
+  const terminate = (): void => {
     if (graceTimer !== undefined) return // escalation already in flight
-    // After settlement the group is gone and the pid may be reused; callers
-    // commonly kill() in a finally, so this must not re-signal or start a
-    // timer that outlives the handle.
     if (settled) return
-    killGroup(pid, 'SIGTERM')
-    graceTimer = setTimeout(() => { killGroup(pid, 'SIGKILL') }, spec.graceMs)
+    signalTree(platform, pid, 'SIGTERM', child, taskkill)
+    graceTimer = setTimeout(() => {
+      if (!settled) signalTree(platform, pid, 'SIGKILL', child, taskkill)
+    }, spec.graceMs)
   }
 
   // The caller owns timeout classification; this layer only reacts to abort.
-  const onAbort = (): void => { kill() }
+  const onAbort = (): void => { terminate() }
   spec.signal?.addEventListener('abort', onAbort, { once: true })
 
-  // Stdin writes are best-effort; process exit and captured output remain authoritative.
-  if (child.stdin !== null) {
+  // Batch stdin is written and closed up front; process exit and captured
+  // output remain authoritative, so write errors (EPIPE) are best-effort.
+  if (typeof stdinMode === 'object' && child.stdin !== null) {
     child.stdin.on('error', () => { /* stdin write is best-effort; outcome rides on exit/output. */ })
-    child.stdin.end(spec.stdin)
+    child.stdin.end(stdinMode.data)
   }
 
   const done = new Promise<SubprocessOutcome>((resolve, reject) => {
@@ -294,15 +380,14 @@ export function spawnProcess(spec: SubprocessSpawnSpec, internals: SpawnInternal
     const settle = (exitCode: number | null, signal: NodeJS.Signals | null): void => {
       if (settled) return
       settled = true
-      child.stdout.destroy()
-      child.stderr.destroy()
+      // Only harness-collected pipes are force-closed at the drain boundary;
+      // a 'pipe'-mode stream belongs to the caller and closes with the child.
+      if (stdoutCollector !== undefined) child.stdout?.destroy()
+      if (stderrCollector !== undefined) child.stderr?.destroy()
+      stdoutCollector?.seal()
+      stderrCollector?.seal()
       cleanup()
-      resolve({
-        exitCode,
-        signal,
-        stdout: stdout.finalize(),
-        stderr: stderr.finalize(),
-      })
+      resolve({ exitCode, signal })
     }
     child.on('error', (error) => {
       // No meaningful close outcome follows a spawn failure.
@@ -311,6 +396,9 @@ export function spawnProcess(spec: SubprocessSpawnSpec, internals: SpawnInternal
       reject(error)
     })
     child.on('exit', (exitCode, signal) => {
+      // A surviving descendant that inherited a pipe must not hold the
+      // outcome open indefinitely: after exit, the same bounded grace that
+      // governs kills also bounds the close wait.
       pipeDrainTimer = setTimeout(() => { settle(exitCode, signal) }, spec.graceMs)
     })
     child.on('close', settle)
@@ -321,5 +409,82 @@ export function spawnProcess(spec: SubprocessSpawnSpec, internals: SpawnInternal
     }
   })
 
-  return { pid, stdout, stderr, done, kill }
+  /** Whether the detached tree's root (or POSIX group) is still alive. */
+  const treeAlive = (): boolean => {
+    if (pid <= 0) return false
+    if (platform === 'win32') {
+      // Windows has no group-liveness probe; the direct child's exit is the
+      // observable boundary (taskkill /T already took the tree with it).
+      return child.exitCode === null && child.signalCode === null
+    }
+    try {
+      process.kill(-pid, 0)
+      return true
+    } catch (error) {
+      const code = (error as NodeJS.ErrnoException).code
+      if (code === 'ESRCH') return false
+      /* v8 ignore start -- EPERM and non-POSIX negative-pid failures are platform defenses; CI runs
+         tree-lifecycle tests on POSIX hosts where absence reports ESRCH. */
+      if (code === 'EPERM') return true
+      return child.exitCode === null && child.signalCode === null
+      /* v8 ignore stop */
+    }
+  }
+
+  const waitForExit = async (signal?: AbortSignal): Promise<boolean> => {
+    while (treeAlive()) {
+      if (signal?.aborted) return false
+      await yieldToEventLoop()
+    }
+    return true
+  }
+
+  /** Race settlement against a timer without leaving listeners or live timers behind. */
+  const settlesWithin = async (ms: number): Promise<boolean> => {
+    if (settled) return true
+    let timer: NodeJS.Timeout | undefined
+    const timeout = new Promise<false>((resolve) => {
+      // `.unref()` so a pending grace timer never keeps the parent's loop alive.
+      timer = setTimeout(() => { resolve(false) }, ms)
+      timer.unref()
+    })
+    try {
+      return await Promise.race([done.then(() => true, () => true), timeout])
+    } finally {
+      if (timer !== undefined) clearTimeout(timer)
+    }
+  }
+
+  let disposal: Promise<void> | undefined
+  const dispose = (graces: SubprocessDisposeGraces): Promise<void> => (disposal ??= (async () => {
+    // 1. Close a piped stdin and allow cooperative teardown and flush.
+    if (stdinMode === 'pipe') child.stdin?.end()
+    if (await settlesWithin(graces.eofGraceMs)) return
+    // 2. POSIX gets a catchable graceful signal; Windows taskkill force-terminates.
+    if (platform !== 'win32') {
+      kill('SIGTERM')
+      if (await settlesWithin(graces.graceMs)) return
+    }
+    // 3. Force-kill the tree and await a bounded exit edge.
+    kill('SIGKILL')
+    if (!(await settlesWithin(graces.graceMs))) {
+      throw new Error(`child process did not exit within ${graces.graceMs}ms after forced termination`)
+    }
+  })())
+
+  return {
+    pid,
+    stdin: stdinMode === 'pipe' ? child.stdin ?? undefined : undefined,
+    stdout: outMode === 'pipe' ? child.stdout ?? undefined : undefined,
+    stderr: errMode === 'pipe' ? child.stderr ?? undefined : undefined,
+    collected: {
+      ...stdoutCollector !== undefined ? { stdout: stdoutCollector } : {},
+      ...stderrCollector !== undefined ? { stderr: stderrCollector } : {},
+    },
+    done,
+    kill,
+    terminate,
+    waitForExit,
+    dispose,
+  }
 }

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

@@ -7,9 +7,11 @@ function spec(command: string, overrides: Partial<SubprocessSpawnSpec> = {}): Su
   return {
     argv: ['bash', '-c', command],
     cwd: process.cwd(),
-    stdoutMaxBytes: 64_000,
-    stderrMaxBytes: 64_000,
-    maxSpillBytes: 64 * 1024 * 1024,
+    stdio: {
+      stdin: 'ignore',
+      stdout: { maxBytes: 64_000, spill: { maxBytes: 64 * 1024 * 1024 } },
+      stderr: { maxBytes: 64_000, spill: { maxBytes: 64 * 1024 * 1024 } },
+    },
     graceMs: 200,
     ...overrides,
   }
@@ -19,9 +21,10 @@ describe('LocalSubprocessService', () => {
   it('registers as ctx.subprocess and spawns managed handles', async () => {
     const ctx = new Context()
     const fiber = await ctx.plugin(LocalSubprocessService)
-    const result = await ctx.subprocess.spawn(spec('echo managed')).done
+    const handle = ctx.subprocess.spawn(spec('echo managed'))
+    const result = await handle.done
     expect(result.exitCode).toBe(0)
-    expect(result.stdout.text).toBe('managed\n')
+    expect(handle.collected.stdout!.readFrom(0).text).toBe('managed\n')
     await fiber.dispose()
   })
 

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

@@ -3,8 +3,8 @@ import { tmpdir } from 'node:os'
 import { dirname, join } from 'node:path'
 import { describe, expect, it, vi } from 'vitest'
 import type { DshEnvironment } from '@deepseek-ai/dsh-subprocess'
-import { killGroup, OutputCollector, spawnProcess } from '../src/spawn.ts'
-import type { SubprocessHandle } from '@deepseek-ai/dsh-subprocess'
+import { killGroup, OutputCollector, spawnSubprocess } from '../src/spawn.ts'
+import type { SubprocessHandle, SubprocessOutputReader } from '@deepseek-ai/dsh-subprocess'
 
 const { failNextClose, failNextUnlink } = vi.hoisted(() => ({
   failNextClose: { value: false },
@@ -33,15 +33,25 @@ vi.mock('node:fs', async (importOriginal) => {
 
 const spillDir = mkdtempSync(join(tmpdir(), 'dsh-subprocess-spec-'))
 
-function spec(command: string, overrides: Partial<Parameters<typeof spawnProcess>[0]> = {}) {
+type SpecOverrides = Partial<Parameters<typeof spawnSubprocess>[0]> & {
+  stdoutMaxBytes?: number
+  stderrMaxBytes?: number
+  maxSpillBytes?: number
+  stdin?: string
+}
+
+function spec(command: string, overrides: SpecOverrides = {}) {
+  const { stdoutMaxBytes = 64_000, stderrMaxBytes = 64_000, maxSpillBytes = 64 * 1024 * 1024, stdin, ...rest } = overrides
   return {
     argv: ['bash', '-c', command],
     cwd: process.cwd(),
-    stdoutMaxBytes: 64_000,
-    stderrMaxBytes: 64_000,
-    maxSpillBytes: 64 * 1024 * 1024,
+    stdio: {
+      stdin: stdin !== undefined ? { data: stdin } : 'ignore' as const,
+      stdout: { maxBytes: stdoutMaxBytes, spill: { maxBytes: maxSpillBytes } },
+      stderr: { maxBytes: stderrMaxBytes, spill: { maxBytes: maxSpillBytes } },
+    },
     graceMs: 3_000,
-    ...overrides,
+    ...rest,
   }
 }
 
@@ -62,12 +72,22 @@ async function waitGone(pid: number, timeoutMs = 5_000): Promise<void> {
 async function waitForStdout(running: SubprocessHandle, expected: string, timeoutMs = 5_000): Promise<void> {
   const deadline = Date.now() + timeoutMs
   while (Date.now() < deadline) {
-    if (running.stdout.readFrom(0).text.includes(expected)) return
+    if (running.collected.stdout!.readFrom(0).text.includes(expected)) return
     await new Promise(resolve => setTimeout(resolve, 20))
   }
   throw new Error(`stdout did not include ${JSON.stringify(expected)} after ${timeoutMs}ms`)
 }
 
+/** Await settlement and project both collected streams like a batch outcome. */
+async function finish(running: SubprocessHandle) {
+  const outcome = await running.done
+  const final = (reader: SubprocessOutputReader | undefined) => {
+    const read = reader!.readFrom(0)
+    return { text: read.text, truncated: read.lossy, ...read.spillPath !== undefined ? { spillPath: read.spillPath } : {} }
+  }
+  return { ...outcome, stdout: final(running.collected.stdout), stderr: final(running.collected.stderr) }
+}
+
 async function waitForPidFile(path: string, timeoutMs = 5_000): Promise<number> {
   const deadline = Date.now() + timeoutMs
   while (Date.now() < deadline) {
@@ -82,9 +102,9 @@ async function waitForPidFile(path: string, timeoutMs = 5_000): Promise<number>
   throw new Error(`pid file ${path} was not written after ${timeoutMs}ms`)
 }
 
-describe('spawnProcess', () => {
+describe('spawnSubprocess', () => {
   it('captures stdout on success', async () => {
-    const result = await spawnProcess(spec('echo hello')).done
+    const result = await finish(spawnSubprocess(spec('echo hello')))
     expect(result.exitCode).toBe(0)
     expect(result.signal).toBeNull()
     expect(result.stdout.text).toBe('hello\n')
@@ -93,33 +113,33 @@ describe('spawnProcess', () => {
   })
 
   it('captures stderr separately', async () => {
-    const result = await spawnProcess(spec('echo oops >&2')).done
+    const result = await finish(spawnSubprocess(spec('echo oops >&2')))
     expect(result.exitCode).toBe(0)
     expect(result.stdout.text).toBe('')
     expect(result.stderr.text).toBe('oops\n')
   })
 
   it('captures both streams', async () => {
-    const result = await spawnProcess(spec('echo out; echo err >&2')).done
+    const result = await finish(spawnSubprocess(spec('echo out; echo err >&2')))
     expect(result.stdout.text).toBe('out\n')
     expect(result.stderr.text).toBe('err\n')
   })
 
   it('reports non-zero exit codes', async () => {
-    const result = await spawnProcess(spec('exit 42')).done
+    const result = await finish(spawnSubprocess(spec('exit 42')))
     expect(result.exitCode).toBe(42)
     expect(result.signal).toBeNull()
   })
 
   it('passes the ambient TERM through untouched (terminal policy is the caller\'s)', async () => {
-    const result = await spawnProcess(spec('echo "${TERM:-unset}"', {
+    const result = await finish(spawnSubprocess(spec('echo "${TERM:-unset}"', {
       env: { TERM: 'callers-choice' },
-    })).done
+    })))
     expect(result.stdout.text).toBe('callers-choice\n')
   })
 
   it('runs in the requested cwd', async () => {
-    const result = await spawnProcess(spec('pwd', { cwd: '/tmp' })).done
+    const result = await finish(spawnSubprocess(spec('pwd', { cwd: '/tmp' })))
     expect(result.stdout.text.trim()).toMatch(/\/tmp$/)
   })
 
@@ -129,7 +149,7 @@ describe('spawnProcess', () => {
     // assert the kill itself lands as SIGTERM.
     const controller = new AbortController()
     const start = Date.now()
-    const running = spawnProcess(spec('sleep 60', { signal: controller.signal }))
+    const running = spawnSubprocess(spec('sleep 60', { signal: controller.signal }))
     setTimeout(() => { controller.abort('deadline') }, 100)
     const result = await running.done
     expect(Date.now() - start).toBeLessThan(5_000)
@@ -137,10 +157,21 @@ describe('spawnProcess', () => {
     expect(result.exitCode).toBeNull()
   })
 
-  it('escalates to SIGKILL when SIGTERM is trapped', async () => {
-    const running = spawnProcess(spec('trap \'\' TERM; echo ready; while :; do sleep 60 & wait $!; done', { graceMs: 200 }))
+  it('terminate() escalates to SIGKILL when SIGTERM is trapped', async () => {
+    const running = spawnSubprocess(spec('trap \'\' TERM; echo ready; while :; do sleep 60 & wait $!; done', { graceMs: 200 }))
     await waitForStdout(running, 'ready\n')
-    running.kill()
+    running.terminate()
+    const result = await running.done
+    expect(result.signal).toBe('SIGKILL')
+  })
+
+  it('kill() sends one signal Node-style, without escalation', async () => {
+    const running = spawnSubprocess(spec('trap \'\' TERM; echo armed; sleep 60', { graceMs: 100 }))
+    await waitForStdout(running, 'armed\n')
+    running.kill() // trapped SIGTERM, no SIGKILL follow-up
+    await new Promise(resolve => setTimeout(resolve, 400))
+    expect(running.collected.stdout).toBeDefined()
+    running.kill('SIGKILL') // explicit signal choice, still no timers
     const result = await running.done
     expect(result.signal).toBe('SIGKILL')
   })
@@ -149,7 +180,7 @@ describe('spawnProcess', () => {
     // The subshell writes the sleep's pid then waits on it; killing the
     // group must take the sleep down with bash.
     const pidFile = join(spillDir, `grandchild-${Date.now()}.pid`)
-    const running = spawnProcess(spec(`sleep 60 & echo $! > ${pidFile}; wait`))
+    const running = spawnSubprocess(spec(`sleep 60 & echo $! > ${pidFile}; wait`))
     const grandchild = await waitForPidFile(pidFile)
     expect(grandchild).toBeGreaterThan(0)
 
@@ -161,7 +192,7 @@ describe('spawnProcess', () => {
 
   it('aborts via AbortSignal mid-run', async () => {
     const controller = new AbortController()
-    const running = spawnProcess(spec('sleep 60', { signal: controller.signal }))
+    const running = spawnSubprocess(spec('sleep 60', { signal: controller.signal }))
     setTimeout(() => { controller.abort('user cancelled') }, 50)
     const result = await running.done
     expect(result.signal).toBe('SIGTERM')
@@ -170,19 +201,19 @@ describe('spawnProcess', () => {
   it('throws when the signal is already aborted before spawn', () => {
     const controller = new AbortController()
     controller.abort('too late')
-    expect(() => spawnProcess(spec('echo hi', { signal: controller.signal })))
+    expect(() => spawnSubprocess(spec('echo hi', { signal: controller.signal })))
       .toThrow(/aborted before spawn: too late/)
   })
 
   it('rejects with a spawn error for a nonexistent cwd', async () => {
-    await expect(spawnProcess(spec('echo hi', { cwd: '/nonexistent-dir-dsh-test' })).done)
+    await expect(spawnSubprocess(spec('echo hi', { cwd: '/nonexistent-dir-dsh-test' })).done)
       .rejects.toThrow(/ENOENT/)
   })
 
-  it('kill() is idempotent (second call does not restart escalation)', async () => {
-    const running = spawnProcess(spec('sleep 60'))
-    running.kill()
-    running.kill()
+  it('terminate() is idempotent (second call does not restart escalation)', async () => {
+    const running = spawnSubprocess(spec('sleep 60'))
+    running.terminate()
+    running.terminate()
     const result = await running.done
     expect(result.signal).toBe('SIGTERM')
   })
@@ -190,10 +221,10 @@ describe('spawnProcess', () => {
   it('bounds inherited-pipe draining after the shell exits', async () => {
     const pidFile = join(spillDir, `pipe-holder-${Date.now()}.pid`)
     const started = Date.now()
-    const running = spawnProcess(spec(`sleep 60 & echo $! > ${pidFile}; echo shell-done`, { graceMs: 100 }))
+    const running = spawnSubprocess(spec(`sleep 60 & echo $! > ${pidFile}; echo shell-done`, { graceMs: 100 }))
     const descendant = await waitForPidFile(pidFile)
     try {
-      const result = await running.done
+      const result = await finish(running)
       expect(Date.now() - started).toBeLessThan(1_000)
       expect(result.exitCode).toBe(0)
       expect(result.stdout.text).toBe('shell-done\n')
@@ -206,7 +237,7 @@ describe('spawnProcess', () => {
 
 describe('stdin and extra env (set by in-process plugins)', () => {
   it('writes stdin to the command and closes it', async () => {
-    const result = await spawnProcess(spec('cat', { stdin: 'hello from stdin\n' })).done
+    const result = await finish(spawnSubprocess(spec('cat', { stdin: 'hello from stdin\n' })))
     expect(result.exitCode).toBe(0)
     expect(result.stdout.text).toBe('hello from stdin\n')
   })
@@ -214,7 +245,7 @@ describe('stdin and extra env (set by in-process plugins)', () => {
   it('a command that reads stdin sees EOF when none is supplied', async () => {
     // No stdin → fd 0 is /dev/null, so `cat` reads EOF and exits 0 with no
     // output (it does NOT block).
-    const result = await spawnProcess(spec('cat')).done
+    const result = await finish(spawnSubprocess(spec('cat')))
     expect(result.exitCode).toBe(0)
     expect(result.stdout.text).toBe('')
   })
@@ -222,25 +253,25 @@ describe('stdin and extra env (set by in-process plugins)', () => {
   it('gives fd 0 the exact pre-seam type: /dev/null when no stdin, a pipe when supplied', async () => {
     // With no bytes, fd 0 remains the pre-seam `ignore` default (/dev/null, a character device).
     // Supplied bytes use Node's spawn pipe, which is an AF_UNIX socket rather than a FIFO.
-    const none = await spawnProcess(spec('test -c /dev/stdin && echo char || echo other')).done
+    const none = await finish(spawnSubprocess(spec('test -c /dev/stdin && echo char || echo other')))
     expect(none.stdout.text).toBe('char\n')
-    const piped = await spawnProcess(spec('test -S /dev/stdin && echo socket || echo other', { stdin: 'x' })).done
+    const piped = await finish(spawnSubprocess(spec('test -S /dev/stdin && echo socket || echo other', { stdin: 'x' })))
     expect(piped.stdout.text).toBe('socket\n')
   })
 
   it('merges ordinary extra env entries onto the scrubbed environment', async () => {
-    const result = await spawnProcess(spec('echo "$EXTRA_ONE/$EXTRA_TWO"', {
+    const result = await finish(spawnSubprocess(spec('echo "$EXTRA_ONE/$EXTRA_TWO"', {
       env: { EXTRA_ONE: 'alpha', EXTRA_TWO: 'beta' },
-    })).done
+    })))
     expect(result.stdout.text).toBe('alpha/beta\n')
   })
 
   it('an explicit extra env entry overrides the credential scrub', async () => {
     // EXPLICIT_OVERRIDE_KEY matches the credential scrub pattern, yet an explicit
     // entry is still honored — the scrub only drops AMBIENT process.env creds.
-    const result = await spawnProcess(spec('echo "$EXPLICIT_OVERRIDE_KEY"', {
+    const result = await finish(spawnSubprocess(spec('echo "$EXPLICIT_OVERRIDE_KEY"', {
       env: { EXPLICIT_OVERRIDE_KEY: 'explicit-wins' },
-    })).done
+    })))
     expect(result.stdout.text).toBe('explicit-wins\n')
   })
 
@@ -248,20 +279,20 @@ describe('stdin and extra env (set by in-process plugins)', () => {
     // The child exits without reading, so closing a stdin pipe holding ~1 MiB triggers EPIPE.
     // The handler swallows that write error and `done` reports the child's real exit.
     const big = 'x'.repeat(1024 * 1024)
-    const result = await spawnProcess(spec('exit 7', { stdin: big })).done
+    const result = await finish(spawnSubprocess(spec('exit 7', { stdin: big })))
     expect(result.exitCode).toBe(7)
   })
 })
 
 describe('output truncation and spill', () => {
   it('applies stdout and stderr caps independently', async () => {
-    const result = await spawnProcess(
+    const result = await finish(spawnSubprocess(
       spec('printf "%.0sx" $(seq 1 500); printf "%.0se" $(seq 1 500) >&2', {
         stdoutMaxBytes: 500,
         stderrMaxBytes: 100,
       }),
       { spillDir },
-    ).done
+    ))
     expect(result.stdout.truncated).toBe(false)
     expect(result.stdout.text).toBe('x'.repeat(500))
     expect(result.stderr.truncated).toBe(true)
@@ -270,10 +301,10 @@ describe('output truncation and spill', () => {
 
   it('keeps the tail and spills the full stream to disk', async () => {
     // 200 numbered lines of ~10 bytes; cap at 500 bytes keeps a late tail.
-    const result = await spawnProcess(
+    const result = await finish(spawnSubprocess(
       spec('for i in $(seq 1 200); do printf "line-%04d\\n" $i; done', { stdoutMaxBytes: 500, stderrMaxBytes: 500 }),
       { spillDir },
-    ).done
+    ))
     expect(result.stdout.truncated).toBe(true)
     expect(result.stdout.text.length).toBeLessThanOrEqual(500)
     expect(result.stdout.text).toContain('line-0200')
@@ -285,10 +316,10 @@ describe('output truncation and spill', () => {
   })
 
   it('does not truncate output exactly at the cap', async () => {
-    const result = await spawnProcess(
+    const result = await finish(spawnSubprocess(
       spec('printf "%.0sx" $(seq 1 500)', { stdoutMaxBytes: 500, stderrMaxBytes: 500 }),
       { spillDir },
-    ).done
+    ))
     expect(result.stdout.truncated).toBe(false)
     expect(result.stdout.text.length).toBe(500)
     expect(result.stdout.spillPath).toBeUndefined()
@@ -296,10 +327,10 @@ describe('output truncation and spill', () => {
 
   it('settles with the tail and no spill path when final spill close fails', async () => {
     failNextClose.value = true
-    const result = await spawnProcess(
+    const result = await finish(spawnSubprocess(
       spec('for i in $(seq 1 200); do printf "line-%04d\\n" $i; done', { stdoutMaxBytes: 500, stderrMaxBytes: 500 }),
       { spillDir },
-    ).done
+    ))
     expect(failNextClose.value).toBe(false)
     expect(result.exitCode).toBe(0)
     expect(result.stdout.truncated).toBe(true)
@@ -403,7 +434,7 @@ describe('killGroup', () => {
   })
 
   it('swallows ESRCH for vanished groups', async () => {
-    const running = spawnProcess(spec('true'))
+    const running = spawnSubprocess(spec('true'))
     await running.done
     expect(() => { killGroup(running.pid, 'SIGTERM') }).not.toThrow()
   })
@@ -412,7 +443,7 @@ describe('killGroup', () => {
     // Cleanup code commonly kills handles in a finally; after settlement the
     // group is gone and the pid may be reused, so a late kill must be inert
     // (no signal to a possibly-recycled pgid, no referenced timer delaying exit).
-    const running = spawnProcess(spec('true'))
+    const running = spawnSubprocess(spec('true'))
     await running.done
     const spy = vi.spyOn(process, 'kill')
     try {
@@ -424,17 +455,137 @@ describe('killGroup', () => {
   })
 })
 
+describe('stdio dispositions', () => {
+  it("'pipe' exposes raw streams for caller-owned protocol decoding", async () => {
+    const running = spawnSubprocess({
+      ...spec('cat'),
+      stdio: { stdin: 'pipe', stdout: 'pipe', stderr: { maxBytes: 1000 } },
+    })
+    expect(running.stdin).toBeDefined()
+    expect(running.stdout).toBeDefined()
+    expect(running.stderr).toBeUndefined()
+    expect(running.collected.stdout).toBeUndefined()
+    expect(running.collected.stderr).toBeDefined()
+
+    const echoed = new Promise<string>((resolve) => {
+      let text = ''
+      running.stdout!.on('data', (chunk: Buffer) => { text += chunk.toString('utf8') })
+      running.stdout!.on('end', () => { resolve(text) })
+    })
+    running.stdin!.end('through the pipe\n')
+    const outcome = await running.done
+    expect(outcome.exitCode).toBe(0)
+    expect(await echoed).toBe('through the pipe\n')
+  })
+
+  it('a collect mode without spill keeps only the in-memory tail (no file)', async () => {
+    const running = spawnSubprocess({
+      ...spec('for i in $(seq 1 200); do printf "line-%04d\\n" $i; done'),
+      stdio: { stdin: 'ignore', stdout: { maxBytes: 100 }, stderr: { maxBytes: 100 } },
+    }, { spillDir })
+    await running.done
+    const read = running.collected.stdout!.readFrom(0)
+    expect(read.lossy).toBe(true)
+    expect(read.text).toContain('line-0200')
+    expect(read.spillPath).toBeUndefined()
+  })
+})
+
+describe('dispose ladder', () => {
+  it('tier 1: a cooperative child exits on stdin EOF without any signal', async () => {
+    const running = spawnSubprocess({
+      ...spec('read -r line; exit 0'),
+      stdio: { stdin: 'pipe', stdout: { maxBytes: 1000 }, stderr: { maxBytes: 1000 } },
+    })
+    await running.dispose({ eofGraceMs: 5_000, graceMs: 200 })
+    const outcome = await running.done
+    expect(outcome.exitCode).toBe(0)
+    expect(outcome.signal).toBeNull()
+  })
+
+  it('tier 2: an EOF-deaf child dies by SIGTERM', async () => {
+    const running = spawnSubprocess({
+      ...spec('sleep 60'),
+      stdio: { stdin: 'pipe', stdout: { maxBytes: 1000 }, stderr: { maxBytes: 1000 } },
+    })
+    await running.dispose({ eofGraceMs: 100, graceMs: 5_000 })
+    const outcome = await running.done
+    expect(outcome.signal).toBe('SIGTERM')
+  })
+
+  it('tier 3: a TERM-trapping child dies by SIGKILL, and dispose() is idempotent', async () => {
+    const running = spawnSubprocess(spec('trap \'\' TERM; echo armed; sleep 60'))
+    await waitForStdout(running, 'armed\n')
+    const first = running.dispose({ eofGraceMs: 50, graceMs: 200 })
+    const second = running.dispose({ eofGraceMs: 50, graceMs: 200 })
+    expect(second).toBe(first)
+    await first
+    const outcome = await running.done
+    expect(outcome.signal).toBe('SIGKILL')
+  })
+})
+
+describe('windows tree semantics (injected platform)', () => {
+  it('kill and terminate route through taskkill by root pid', async () => {
+    const killed: number[] = []
+    const running = spawnSubprocess(spec('sleep 60', { graceMs: 100 }), {
+      spillDir,
+      platform: 'win32',
+      taskkill: (pid) => {
+        killed.push(pid)
+        // Simulate the forced tree termination taskkill performs.
+        try {
+          process.kill(pid, 'SIGKILL')
+        } catch {
+          // Already gone — matches taskkill's tolerated not-found status.
+        }
+      },
+    })
+    running.terminate()
+    const outcome = await running.done
+    expect(killed).toContain(running.pid)
+    expect(outcome.signal).toBe('SIGKILL')
+  })
+
+  it('waitForExit falls back to direct-child liveness where groups do not exist', async () => {
+    const running = spawnSubprocess(spec('true'), { spillDir, platform: 'win32', taskkill: () => {} })
+    await running.done
+    await expect(running.waitForExit()).resolves.toBe(true)
+  })
+})
+
+describe('waitForExit', () => {
+  it('waits for the whole detached tree, not just the shell', async () => {
+    const pidFile = join(spillDir, `tree-wait-${Date.now()}.pid`)
+    const running = spawnSubprocess(spec(`sleep 60 & echo $! > ${pidFile}; wait`))
+    const grandchild = await waitForPidFile(pidFile)
+    running.terminate()
+    await running.done
+    await expect(running.waitForExit()).resolves.toBe(true)
+    await expect(waitGone(grandchild, 100)).resolves.toBeUndefined()
+  })
+
+  it('an aborted wait reports false while the tree lives', async () => {
+    const running = spawnSubprocess(spec('sleep 60'))
+    const controller = new AbortController()
+    controller.abort()
+    await expect(running.waitForExit(controller.signal)).resolves.toBe(false)
+    running.terminate()
+    await running.done
+  })
+})
+
 describe('argv validation', () => {
   it('rejects an empty argv before spawning', () => {
-    expect(() => spawnProcess({ ...spec('true'), argv: [] })).toThrow(/non-empty program name/)
+    expect(() => spawnSubprocess({ ...spec('true'), argv: [] })).toThrow(/non-empty program name/)
   })
 
   it('rejects an empty program name before spawning', () => {
-    expect(() => spawnProcess({ ...spec('true'), argv: [''] })).toThrow(/non-empty program name/)
+    expect(() => spawnSubprocess({ ...spec('true'), argv: [''] })).toThrow(/non-empty program name/)
   })
 
   it('spawns argv verbatim without shell interpretation', async () => {
-    const result = await spawnProcess({ ...spec('unused'), argv: ['printf', '%s', '$HOME'] }).done
+    const result = await finish(spawnSubprocess({ ...spec('unused'), argv: ['printf', '%s', '$HOME'] }))
     expect(result.stdout.text).toBe('$HOME')
   })
 })
@@ -449,14 +600,14 @@ describe('abort edge cases', () => {
       addEventListener() {},
       removeEventListener() {},
     } as unknown as AbortSignal
-    expect(() => spawnProcess(spec('echo hi', { signal: bare })))
+    expect(() => spawnSubprocess(spec('echo hi', { signal: bare })))
       .toThrow(/aborted before spawn: aborted/)
   })
 
   it('reports the terminating signal of an externally self-killed command', async () => {
     // spawnProcess reports the raw signal; whether it counts as timeout/cancel is the
     // executor's classification (a self-kill is neither) — see executor.spec.ts.
-    const result = await spawnProcess(spec('kill -TERM $$')).done
+    const result = await finish(spawnSubprocess(spec('kill -TERM $$')))
     expect(result.signal).toBe('SIGTERM')
   })
 })
@@ -467,7 +618,7 @@ describe('environment and spill-file hardening', () => {
     process.env.DSH_TEST_TOKEN = 'also-secret'
     process.env.DSH_TEST_PLAIN = 'visible'
     try {
-      const result = await spawnProcess(spec('echo "[${DSH_TEST_API_KEY:-absent}|${DSH_TEST_TOKEN:-absent}|${DSH_TEST_PLAIN:-absent}]"')).done
+      const result = await finish(spawnSubprocess(spec('echo "[${DSH_TEST_API_KEY:-absent}|${DSH_TEST_TOKEN:-absent}|${DSH_TEST_PLAIN:-absent}]"')))
       expect(result.stdout.text.trim()).toBe('[absent|absent|absent]')
     } finally {
       delete process.env.DSH_TEST_API_KEY
@@ -479,9 +630,9 @@ describe('environment and spill-file hardening', () => {
   it('injects only the current trusted DSH environment after scrubbing ambient values', async () => {
     process.env.DSH_STALE = 'old-value'
     try {
-      const result = await spawnProcess(spec('echo "[${DSH_STALE:-absent}|$DSH_SHELL|$DSH_SESSION_ID]"', {
+      const result = await finish(spawnSubprocess(spec('echo "[${DSH_STALE:-absent}|$DSH_SHELL|$DSH_SESSION_ID]"', {
         dshEnv: { DSH_SHELL: '1', DSH_SESSION_ID: 'current-session' },
-      })).done
+      })))
       expect(result.stdout.text.trim()).toBe('[absent|1|current-session]')
     } finally {
       delete process.env.DSH_STALE
@@ -489,21 +640,21 @@ describe('environment and spill-file hardening', () => {
   })
 
   it('rejects DSH variables on the ordinary env channel', () => {
-    expect(() => spawnProcess(spec('true', { env: { DSH_WRONG_CHANNEL: 'bad' } })))
+    expect(() => spawnSubprocess(spec('true', { env: { DSH_WRONG_CHANNEL: 'bad' } })))
       .toThrow(/DSH_WRONG_CHANNEL.*dshEnv/)
   })
 
   it('rejects ordinary variables on the managed env channel', () => {
     const invalid = { PATH: '/wrong-channel' } as unknown as DshEnvironment
-    expect(() => spawnProcess(spec('true', { dshEnv: invalid })))
+    expect(() => spawnSubprocess(spec('true', { dshEnv: invalid })))
       .toThrow(/managed child env.*PATH.*use env/)
   })
 
   it('creates spill files with owner-only permissions and random names', async () => {
-    const result = await spawnProcess(
+    const result = await finish(spawnSubprocess(
       spec('for i in $(seq 1 200); do printf "line-%04d\\n" $i; done', { stdoutMaxBytes: 500, stderrMaxBytes: 500 }),
       { spillDir },
-    ).done
+    ))
     const path = result.stdout.spillPath!
     expect(path).toMatch(/dsh-subprocess-\d+-\d+-[0-9a-f]{12}-stdout\.log$/)
     const mode = statSync(path).mode & 0o777
@@ -511,9 +662,9 @@ describe('environment and spill-file hardening', () => {
   })
 
   it('defaults spills into a private per-process directory', async () => {
-    const result = await spawnProcess(
+    const result = await finish(spawnSubprocess(
       spec('for i in $(seq 1 200); do printf "line-%04d\\n" $i; done', { stdoutMaxBytes: 500, stderrMaxBytes: 500 }),
-    ).done
+    ))
     const dir = dirname(result.stdout.spillPath!)
     expect(dir).toMatch(/dsh-subprocess-/)
     const mode = statSync(dir).mode & 0o777
@@ -533,7 +684,7 @@ describe('environment and spill-file hardening', () => {
 
   it('honors AbortSignal on background-style runs (no timeout)', async () => {
     const controller = new AbortController()
-    const running = spawnProcess(spec('sleep 60', { signal: controller.signal }))
+    const running = spawnSubprocess(spec('sleep 60', { signal: controller.signal }))
     setTimeout(() => { controller.abort() }, 50)
     const result = await running.done
     expect(result.signal).toBe('SIGTERM')

packages/subprocess/subprocess/src/index.ts

@@ -1,14 +1,17 @@
 /**
- * The subprocess seam (`ctx.subprocess`): spawn fully-specified
- * commands into managed process groups with bounded, spill-backed output and
- * escalated kills. Command defaulting, shell semantics, deadlines, and
- * presentation belong to consumers — the bash executor seam is the owning
+ * The subprocess seam (`ctx.subprocess`): spawn fully-specified commands into
+ * managed process trees with Node-shaped stdio dispositions — raw pipes for
+ * protocol streams, inherit for diagnostics, bounded spill-backed collection
+ * for batch output — plus tree-scoped signalling and a cooperative dispose
+ * ladder. Command defaulting, shell semantics, deadlines, framing, and
+ * presentation belong to consumers; the bash executor seam is the owning
  * template. The local implementation lives in
  * `@deepseek-ai/dsh-subprocess-local`.
  * @module @deepseek-ai/dsh-subprocess
  */
 
 import { Context, Service } from 'cordis'
+import { DSH_ENV_PREFIX } from './types.ts'
 import type { SubprocessHandle, SubprocessSpawnSpec } from './types.ts'
 
 export { DSH_ENV_PREFIX } from './types.ts'
@@ -16,13 +19,48 @@ export type {
   CollectedOutput,
   DshEnvironment,
   DshEnvironmentKey,
+  SubprocessCollect,
+  SubprocessCollectedOutputs,
+  SubprocessDisposeGraces,
   SubprocessHandle,
   SubprocessOutcome,
+  SubprocessOutputMode,
   SubprocessOutputRead,
   SubprocessOutputReader,
   SubprocessSpawnSpec,
+  SubprocessStdinMode,
+  SubprocessStdio,
 } from './types.ts'
 
+/**
+ * Credential-shaped environment names are NOT forwarded to children (the
+ * harness's own `DEEPSEEK_API_KEY`/secrets must not leak into a spawned
+ * process implicitly). One heuristic for every in-repo spawner; a
+ * deliberately supplied entry survives because explicit env layers merge
+ * after the scrub.
+ */
+export const SENSITIVE_ENV_PATTERN = /KEY|SECRET|TOKEN/i
+
+/**
+ * The ambient parent environment minus credential-shaped names and minus all
+ * `DSH_*` names — the canonical base every harness child starts from. `PATH`,
+ * `HOME`, locale, and proxy variables survive, so child CLIs run normally;
+ * harness identity never leaks implicitly (a child that needs current `DSH_*`
+ * facts receives them through {@link SubprocessSpawnSpec.dshEnv}, and a
+ * deliberately forwarded credential goes through an explicit env layer, which
+ * merges after this scrub). Exported as a plain function so spawners that
+ * cannot route through the service (node-pty backends, SDK-managed
+ * transports) share the one scrub definition.
+ * @returns a fresh environment object safe to hand to a child spawn.
+ */
+export function scrubbedParentEnv(): Record<string, string> {
+  const env: Record<string, string> = {}
+  for (const [key, value] of Object.entries(process.env)) {
+    if (value !== undefined && !SENSITIVE_ENV_PATTERN.test(key) && !key.startsWith(DSH_ENV_PREFIX)) env[key] = value
+  }
+  return env
+}
+
 declare module 'cordis' {
   interface Context {
     subprocess: SubprocessService
@@ -37,13 +75,17 @@ declare module 'cordis' {
  *
  * Implementations must honor these semantics:
  * - {@link spawn} returns immediately with a live handle; `done` resolves at
- *   process close and rejects only for spawn-level failures.
- * - Output readers are offset-based and non-consuming, so independent readers
- *   never consume one another's output; lossy reads report truncation and the
- *   spill file holding the complete stream when one exists.
- * - {@link SubprocessHandle.kill} and the spec's abort signal escalate
- *   SIGTERM→grace→SIGKILL across the whole process group.
- * - Disposal kills all still-running managed processes and awaits their exit.
+ *   process close with exit facts and rejects only for spawn-level failures.
+ * - Collect-mode readers are offset-based and non-consuming, so independent
+ *   readers never consume one another's output; lossy reads report truncation
+ *   and the spill file holding the complete stream when one exists. Piped
+ *   streams are handed to the caller raw and never buffered here.
+ * - {@link SubprocessHandle.kill} signals without escalation,
+ *   {@link SubprocessHandle.terminate} (and the spec's abort signal) escalates
+ *   SIGTERM→grace→SIGKILL, and {@link SubprocessHandle.dispose} runs the
+ *   cooperative EOF-first ladder — all tree-scoped on every platform.
+ * - Disposal of the service terminates all still-running managed processes
+ *   and awaits their exit.
  */
 export abstract class SubprocessService extends Service {
   constructor(ctx: Context) {
@@ -53,8 +95,8 @@ export abstract class SubprocessService extends Service {
   /**
    * Start one managed child process from a fully-specified spec; this seam
    * applies no defaults.
-   * @param spec - argv, directory, limits, grace, cancellation, and environment.
-   * @returns the live process handle (readers, kill, outcome promise).
+   * @param spec - argv, directory, stdio dispositions, grace, cancellation, and environment.
+   * @returns the live process handle (streams/readers, signalling, outcome promise).
    */
   abstract spawn(spec: SubprocessSpawnSpec): SubprocessHandle
 }

packages/subprocess/subprocess/src/types.ts

@@ -1,11 +1,14 @@
 /**
- * Vocabulary for the subprocess seam: fully-specified spawn requests,
- * bounded output with spill recovery, and live process handles. Command
- * defaulting, shell semantics, and presentation belong to consumers such as
- * the bash executor seam.
+ * Vocabulary for the subprocess seam: fully-specified spawn requests with
+ * Node-shaped per-stream stdio modes, bounded collected output with spill
+ * recovery, raw piped streams, and tree-scoped termination. Command
+ * defaulting, shell semantics, protocol framing, and presentation belong to
+ * consumers such as the bash executor seam.
  * @module dsh-subprocess/types
  */
 
+import type { Readable, Writable } from 'node:stream'
+
 /** Namespace prefix reserved for DeepSeek Harness-managed child environment facts. */
 export const DSH_ENV_PREFIX = 'DSH_' as const
 
@@ -26,60 +29,97 @@ export interface CollectedOutput {
 }
 
 /**
- * A fully-specified spawn request. This seam applies no defaults: every limit
- * and directory is explicit, so the caller's own config — not a hidden
- * subprocess-service default — decides them (the `dsh-bash` request/spec split
- * is the owning template).
+ * stdin disposition. `'ignore'` leaves fd 0 on `/dev/null`; `'pipe'` exposes
+ * {@link SubprocessHandle.stdin} for the caller's ongoing protocol writes;
+ * `{ data }` writes the bytes and closes (the batch shape).
+ */
+export type SubprocessStdinMode = 'ignore' | 'pipe' | { readonly data: string }
+
+/**
+ * Bounded in-memory collection for one output stream, with an optional
+ * full-stream spill file. Omitting `spill` keeps only the in-memory tail —
+ * the diagnostic-tail shape (a language server's stderr); including it makes
+ * the complete stream recoverable up to its cap (the bash tool shape).
+ */
+export interface SubprocessCollect {
+  /** In-memory cap in bytes; overflow keeps the TAIL. */
+  maxBytes: number
+  /** Full-stream spill file; absent disables spilling entirely. */
+  spill?: {
+    /** Whole-stream byte cap; a larger stream discards its now-incomplete spill. */
+    maxBytes: number
+  }
+}
+
+/**
+ * stdout/stderr disposition. `'pipe'` exposes the raw `Readable` for the
+ * caller's protocol decoding; `'inherit'` passes the parent's descriptor
+ * through (child diagnostics land on the harness's own stream); a
+ * {@link SubprocessCollect} object buffers boundedly with offset-based reads.
+ */
+export type SubprocessOutputMode = 'pipe' | 'inherit' | SubprocessCollect
+
+/** Per-stream stdio dispositions, all explicit — this seam applies no defaults. */
+export interface SubprocessStdio {
+  stdin: SubprocessStdinMode
+  stdout: SubprocessOutputMode
+  stderr: SubprocessOutputMode
+}
+
+/**
+ * A fully-specified spawn request. This seam applies no defaults: every
+ * disposition, limit, and directory is explicit, so the caller's own config —
+ * not a hidden subprocess-service default — decides them (the `dsh-bash`
+ * request/spec split is the owning template).
  */
 export interface SubprocessSpawnSpec {
   /** Executable and arguments; `argv[0]` is the program. Never shell-interpreted here. */
   argv: readonly string[]
   /** Working directory for the child. */
   cwd: string
-  /** Stdout in-memory cap; overflow spills to disk (tail kept in memory). */
-  stdoutMaxBytes: number
-  /** Stderr in-memory cap; overflow spills to disk (tail kept in memory). */
-  stderrMaxBytes: number
-  /** Per-stream spill-file cap; larger streams retain only their in-memory tail. */
-  maxSpillBytes: number
-  /** Grace period for kill escalation and for inherited pipes after process exit. */
-  graceMs: number
+  /** Per-stream stdio dispositions. */
+  stdio: SubprocessStdio
   /**
-   * Abort signal — kills the process group when it fires. The caller owns
-   * deadlines and cause classification; this seam only reacts to the abort.
+   * Grace period in milliseconds for the {@link SubprocessHandle.terminate}
+   * escalation and for draining still-open collected pipes after the process
+   * exits (an inherited descriptor held by a surviving descendant cannot hold
+   * the outcome open indefinitely).
    */
-  signal?: AbortSignal | undefined
+  graceMs: number
   /**
-   * Bytes to write to the child's stdin, then close it. Absent (or empty)
-   * leaves stdin closed/empty.
+   * Abort signal — starts the terminate escalation on the process tree when
+   * it fires. The caller owns deadlines and cause classification; this seam
+   * only reacts to the abort.
    */
-  stdin?: string | undefined
+  signal?: AbortSignal | undefined
   /**
-   * Ordinary environment entries merged after the implementation's credential
-   * scrub. `DSH_*` names are rejected and belong in {@link dshEnv}.
+   * Ordinary environment entries merged onto the implementation's scrubbed
+   * parent base (see `scrubbedParentEnv`). `DSH_*` names are rejected and
+   * belong in {@link dshEnv}; a deliberately forwarded credential-shaped
+   * entry survives because this layer merges after the scrub.
    */
   env?: Record<string, string> | undefined
   /**
-   * Harness-owned `DSH_*` variables for this execution. Implementations
-   * discard ambient `DSH_*` entries before merging this snapshot, so an
-   * unavailable current fact cannot inherit a stale value from the harness
-   * process, and reject non-`DSH_*` names supplied through this channel.
+   * Harness-owned `DSH_*` variables for this execution. The scrubbed base has
+   * already discarded ambient `DSH_*` entries, so an unavailable current fact
+   * cannot inherit a stale value from the harness process; non-`DSH_*` names
+   * on this channel are rejected.
    */
   dshEnv?: DshEnvironment | undefined
 }
 
 /**
- * Raw outcome of one closed process. Deliberately carries NO timeout or
- * cancellation classification: the service kills on abort but does not decide
- * why — the caller reads the signal it owns to classify causes.
+ * Exit facts of one closed process — Node's `close`-event vocabulary.
+ * Deliberately carries NO timeout or cancellation classification (the caller
+ * reads the signal it owns to classify causes) and NO output: collected
+ * streams stay readable through {@link SubprocessHandle.collected} after
+ * settlement, so batch and streaming callers share one access path.
  */
 export interface SubprocessOutcome {
   /** Exit code; null when the process died from a signal. */
   exitCode: number | null
   /** Terminating signal (e.g. 'SIGTERM'); null on normal exit. */
   signal: NodeJS.Signals | null
-  stdout: CollectedOutput
-  stderr: CollectedOutput
 }
 
 /** One incremental {@link SubprocessOutputReader.readFrom} read. */
@@ -95,9 +135,11 @@ export interface SubprocessOutputRead {
 }
 
 /**
- * Cursor-free incremental access to one live output stream. Offsets are
+ * Cursor-free incremental access to one collected output stream. Offsets are
  * whole-stream byte coordinates owned by the caller, so independent readers
- * cannot consume one another's output.
+ * cannot consume one another's output; `readFrom(0)` after settlement is the
+ * batch result (`lossy` then means the in-memory tail lost its head — the
+ * {@link CollectedOutput.truncated} fact).
  */
 export interface SubprocessOutputReader {
   /**
@@ -110,19 +152,87 @@ export interface SubprocessOutputReader {
   readFrom(fromByte: number): SubprocessOutputRead
 }
 
+/** Offset-based readers for the streams spawned in collect mode. */
+export interface SubprocessCollectedOutputs {
+  /** Present iff stdout is a {@link SubprocessCollect}. */
+  readonly stdout?: SubprocessOutputReader
+  /** Present iff stderr is a {@link SubprocessCollect}. */
+  readonly stderr?: SubprocessOutputReader
+}
+
+/**
+ * The two grace periods of the cooperative dispose ladder
+ * ({@link SubprocessHandle.dispose}). Consumers carry them as defaulted,
+ * validated Config fields, so teardown timing is deployment-tunable and this
+ * seam hardcodes nothing.
+ */
+export interface SubprocessDisposeGraces {
+  /**
+   * Tier-1 window (ms): after stdin EOF, how long the child gets to quiesce
+   * ON ITS OWN — flush durable state, tear down its own descendants — before
+   * escalation to platform termination. Usually WIDER than
+   * {@link SubprocessDisposeGraces.graceMs}: a cooperative child's EOF-driven
+   * teardown may itself wait on a signal-trapping grandchild plus a final
+   * flush.
+   */
+  eofGraceMs: number
+  /**
+   * Termination confirmation window (ms): POSIX applies it after `SIGTERM`
+   * and again after `SIGKILL`; Windows applies it after the forced tree
+   * termination.
+   */
+  graceMs: number
+}
+
 /**
- * A live child process. `kill()` starts the group SIGTERM→grace→SIGKILL
- * escalation; buffered output remains readable after exit.
+ * A live child process rooted in its own process tree. Collected output
+ * remains readable after exit; piped streams belong to the caller.
+ *
+ * Termination is tree-scoped everywhere: POSIX signals the detached process
+ * group (falling back to the direct child when the group is gone), Windows
+ * terminates the tree via `taskkill /T`, so helper processes cannot outlive
+ * the handle unnoticed.
  */
 export interface SubprocessHandle {
-  /** Process id (group leader); -1 when the spawn itself failed. */
+  /** Process id (tree root); -1 when the spawn itself failed. */
   readonly pid: number
-  /** Live stdout reader (also readable after exit). */
-  readonly stdout: SubprocessOutputReader
-  /** Live stderr reader (also readable after exit). */
-  readonly stderr: SubprocessOutputReader
-  /** Resolves when the process closes; rejects only for spawn-level failures. */
+  /** The child's stdin, present iff spawned with `stdin: 'pipe'`. */
+  readonly stdin: Writable | undefined
+  /** The child's raw stdout, present iff spawned with `stdout: 'pipe'`. */
+  readonly stdout: Readable | undefined
+  /** The child's raw stderr, present iff spawned with `stderr: 'pipe'`. */
+  readonly stderr: Readable | undefined
+  /** Offset-based readers for collect-mode streams (also readable after exit). */
+  readonly collected: SubprocessCollectedOutputs
+  /** Resolves at process close with exit facts; rejects only for spawn-level failures. */
   readonly done: Promise<SubprocessOutcome>
-  /** Begin SIGTERM→grace→SIGKILL on the process group. Idempotent. */
-  kill(): void
+  /**
+   * Send one signal to the process tree, Node-style — no escalation, no
+   * timers. A no-op after the outcome has settled (the pid may be reused).
+   * @param signal - the signal to deliver (default `SIGTERM`; Windows
+   *   force-terminates the tree for any value).
+   */
+  kill(signal?: NodeJS.Signals): void
+  /**
+   * Begin the SIGTERM → `graceMs` → SIGKILL escalation on the process tree
+   * (Windows force-terminates immediately). Idempotent; also triggered by the
+   * spec's abort signal.
+   */
+  terminate(): void
+  /**
+   * Wait until the process tree has exited — the tree, not just the direct
+   * child, so a still-running helper is observable before teardown returns.
+   * @param signal - optional bound for the wait.
+   * @returns `true` when the tree exited, `false` when the signal aborted first.
+   */
+  waitForExit(signal?: AbortSignal): Promise<boolean>
+  /**
+   * Tear the child down to quiescence, resolving only after exit: close stdin
+   * (when this handle owns a piped one) and allow cooperative flush for
+   * `eofGraceMs`, then SIGTERM with a `graceMs` window (POSIX), then forced
+   * tree termination with a final bounded `graceMs` wait.
+   * @param graces - the ladder's two windows, from the consumer's Config.
+   * @throws when the child still has not exited `graceMs` after the forced tier.
+   */
+  dispose(graces: SubprocessDisposeGraces): Promise<void>
 }

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

@@ -1,7 +1,7 @@
 import { describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
-import { SubprocessService } from '@deepseek-ai/dsh-subprocess'
-import type { SubprocessHandle, SubprocessOutputRead, SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
+import { scrubbedParentEnv, SubprocessService } from '@deepseek-ai/dsh-subprocess'
+import type { SubprocessDisposeGraces, SubprocessHandle, SubprocessOutputRead, SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
 
 /**
  * Minimal concrete service: a hand-built handle. The seam is spawn-only —
@@ -11,18 +11,20 @@ import type { SubprocessHandle, SubprocessOutputRead, SubprocessSpawnSpec } from
 class StubSubprocessService extends SubprocessService {
   spawn(spec: SubprocessSpawnSpec): SubprocessHandle {
     const read: SubprocessOutputRead = { text: '', nextOffset: 0, lossy: false }
-    let killed = false
+    const collected = spec.stdio.stdout !== 'pipe' && spec.stdio.stdout !== 'inherit'
+      ? { stdout: { readFrom: () => read } }
+      : {}
     return {
       pid: spec.argv.length,
-      stdout: { readFrom: () => read },
-      stderr: { readFrom: () => read },
-      done: Promise.resolve({
-        exitCode: killed ? null : 0,
-        signal: null,
-        stdout: { text: 'ok', truncated: false },
-        stderr: { text: '', truncated: false },
-      }),
-      kill: () => { killed = true },
+      stdin: undefined,
+      stdout: undefined,
+      stderr: undefined,
+      collected,
+      done: Promise.resolve({ exitCode: 0, signal: null }),
+      kill: () => {},
+      terminate: () => {},
+      waitForExit: () => Promise.resolve(true),
+      dispose: (_graces: SubprocessDisposeGraces) => Promise.resolve(),
     }
   }
 }
@@ -34,22 +36,40 @@ describe('SubprocessService seam', () => {
     const handle = ctx.subprocess.spawn({
       argv: ['true'],
       cwd: '/stub',
-      stdoutMaxBytes: 1,
-      stderrMaxBytes: 1,
-      maxSpillBytes: 1,
+      stdio: { stdin: 'ignore', stdout: { maxBytes: 1 }, stderr: 'inherit' },
       graceMs: 1,
     })
     expect(handle.pid).toBe(1)
-    expect(handle.stdout.readFrom(0)).toEqual({ text: '', nextOffset: 0, lossy: false })
+    expect(handle.collected.stdout!.readFrom(0)).toEqual({ text: '', nextOffset: 0, lossy: false })
     handle.kill()
+    handle.terminate()
+    await expect(handle.waitForExit()).resolves.toBe(true)
+    await expect(handle.dispose({ eofGraceMs: 1, graceMs: 1 })).resolves.toBeUndefined()
     const outcome = await handle.done
-    expect(outcome.stdout.text).toBe('ok')
+    expect(outcome.exitCode).toBe(0)
   })
 
-  it('loading a second implementation throws (one processes service per context — cordis standard)', async () => {
+  it('loading a second implementation throws (one subprocess service per context — cordis standard)', async () => {
     const ctx = new Context()
     await ctx.plugin(StubSubprocessService)
-    class SecondManager extends StubSubprocessService {}
-    await expect(ctx.plugin(SecondManager)).rejects.toThrow(/service "subprocess" has been registered/)
+    class SecondService extends StubSubprocessService {}
+    await expect(ctx.plugin(SecondService)).rejects.toThrow(/service "subprocess" has been registered/)
+  })
+
+  it('scrubbedParentEnv drops credential-shaped and DSH_ names but keeps PATH', () => {
+    process.env.DSH_SCRUB_PROBE = 'stale'
+    process.env.SCRUB_PROBE_TOKEN = 'secret'
+    process.env.SCRUB_PROBE_PLAIN = 'visible'
+    try {
+      const env = scrubbedParentEnv()
+      expect(env.DSH_SCRUB_PROBE).toBeUndefined()
+      expect(env.SCRUB_PROBE_TOKEN).toBeUndefined()
+      expect(env.SCRUB_PROBE_PLAIN).toBe('visible')
+      expect(env.PATH).toBeDefined()
+    } finally {
+      delete process.env.DSH_SCRUB_PROBE
+      delete process.env.SCRUB_PROBE_TOKEN
+      delete process.env.SCRUB_PROBE_PLAIN
+    }
   })
 })