feat(subprocess): migrate lsp-local, subagent-acp, and the env scrubs onto the seam

Review direction (tianyicui, PR #660): in a stacked PR, change all other
process-running places to use the new service.

- lsp-local: LspConnection spawns through ctx.subprocess (piped protocol
  streams + a no-spill collected stderr tail); its private process-tree
  helpers (POSIX group signalling, Windows taskkill, liveness polling) are
  deleted in favor of the seam's handle verbs, and its buildChildEnv now
  rides scrubbedParentEnv (LSP children also stop inheriting stale DSH_*).
  The plugin injects 'subprocess'; compositions/tests mount
  dsh-subprocess-local.
- subagent-acp: the ACP child spawns through the seam (piped ndjson streams,
  inherited stderr); spawn failure surfaces through done-rejection into the
  same startup race; disposal is handle.dispose with the plugin's configured
  graces. dsh-subagent-subprocess is DELETED — its dispose ladder and scrub
  are the seam's, and the isolated-config-dir helper had no consumer.
- mcp-client, pty-local, sdk-helper: adopt scrubbedParentEnv as the one
  scrub definition (their spawns stay put by ownership: the MCP SDK and
  node-pty own those calls; the SDK wizard runs outside any composition).
- Coverage: per-file 100% over every touched src file, with each v8 ignore
  carrying a platform or contract reason; new suites cover stdio
  dispositions, the dispose ladder tiers, injected-win32 tree semantics,
  waitForExit, settled-kill/terminate no-ops, and spawn-failure disposal.
- Docs: consumer-migration Agent Note (en; zh follows in this PR), seam note
  updated in place, subprocess.md rewritten for the reshaped vocabulary
  (type-equiv re-registered), READMEs and SERVICE_ROLES updated, taskkill
  added to knip ignoreBinaries.

packages/lsp/lsp-local/package.json

@@ -31,6 +31,7 @@
     "@deepseek-ai/dsh-invariants": "^0.0.1",
     "@deepseek-ai/dsh-llm": "^0.0.1",
     "@deepseek-ai/dsh-lsp": "^0.0.1",
+    "@deepseek-ai/dsh-subprocess": "^0.0.1",
     "@deepseek-ai/dsh-timeout": "^0.0.1",
     "cordis": "^4.0.0-rc.7"
   },
@@ -42,6 +43,8 @@
     "@deepseek-ai/dsh-invariants": "workspace:^",
     "@deepseek-ai/dsh-llm": "workspace:^",
     "@deepseek-ai/dsh-lsp": "workspace:^",
+    "@deepseek-ai/dsh-subprocess": "workspace:^",
+    "@deepseek-ai/dsh-subprocess-local": "workspace:^",
     "@deepseek-ai/dsh-timeout": "workspace:^",
     "cordis": "^4.0.0-rc.7",
     "typescript": "^6.0.3",

packages/lsp/lsp-local/src/connection.ts

@@ -1,16 +1,17 @@
 /**
- * A JSON-RPC endpoint over one spawned language server's stdio. Owns id correlation, outbound
- * requests/notifications, and inbound server→client requests: it answers `workspace/configuration`
- * from static config, and rejects `workspace/applyEdit` (this host never applies edits or runs
- * commands). It caps stderr, surfaces framing/decoder failures as a fatal close, and exposes the
- * child handle so the instance owns process-signal teardown.
+ * A JSON-RPC endpoint over one language server spawned through the subprocess
+ * seam. Owns id correlation, outbound requests/notifications, and inbound
+ * server→client requests: it answers `workspace/configuration` from static
+ * config, and rejects `workspace/applyEdit` (this host never applies edits or
+ * runs commands). It caps stderr, surfaces framing/decoder failures as a
+ * fatal close, and exposes tree-scoped termination through the handle so the
+ * instance owns teardown; group/tree mechanics live in the seam's
+ * implementation.
  * @module @deepseek-ai/dsh-lsp-local/connection
  */
 
-import type { ChildProcessByStdio } from 'node:child_process'
-import { spawn, spawnSync } from 'node:child_process'
-import type { Readable, Writable } from 'node:stream'
-import { setImmediate as yieldToEventLoop } from 'node:timers/promises'
+import type { Writable } from 'node:stream'
+import type { SubprocessHandle, SubprocessSpawnSpec } from '@deepseek-ai/dsh-subprocess'
 import { encodeMessage, MessageDecoder } from './framing.ts'
 
 /** How to launch the server and answer its config requests. */
@@ -27,6 +28,12 @@ export interface ConnectionSpec {
   readonly maxMessageBytes: number
   /** Largest stderr tail retained for diagnostics. */
   readonly maxStderrBytes: number
+  /**
+   * Bound (ms) for draining pipes a surviving helper still holds after the
+   * server exits; the instance passes its kill grace so exit observation is
+   * never slower than the escalation it feeds.
+   */
+  readonly pipeDrainGraceMs: number
   /** Static answer to every `workspace/configuration` item. */
   readonly configuration: unknown
 }
@@ -48,178 +55,89 @@ export type ConnectionWriter = (
   done: (error?: Error | null) => void,
 ) => void
 
-/** Host operations used to signal a detached process tree. */
-export interface ProcessTreeOperations {
-  /** Signal a POSIX process group. */
-  readonly signal: (target: number, signal: NodeJS.Signals) => void
-  /** Signal the direct child when POSIX group signaling is unavailable. */
-  readonly killChild: (signal: NodeJS.Signals) => void
-  /** Terminate a Windows process tree by root pid. */
-  readonly taskkill: (pid: number) => void
-}
-
-/** Narrow taskkill runner result used by the Windows process-tree adapter. */
-export interface TaskkillResult {
-  /** Process exit status, or null when spawning failed. */
-  readonly status: number | null
-  /** Spawn failure, when the executable could not run. */
-  readonly error?: Error
-}
-
-/** Invoke a command synchronously for the Windows taskkill adapter. */
-export type TaskkillRunner = (
-  command: string,
-  args: string[],
-  options: { stdio: 'ignore' },
-) => TaskkillResult
-
-/** Invoke the host process-signal primitive for a POSIX process group. */
-export type ProcessSignalRunner = (target: number, signal: NodeJS.Signals) => boolean
-
-const processSignalRunner: ProcessSignalRunner = process.kill.bind(process)
-
-/** taskkill status for "process not found": the requested process tree is already absent. */
-const TASKKILL_TREE_NOT_FOUND_STATUS = 128
+/** Spawn one subprocess for this connection (the provider passes `ctx.subprocess.spawn`). */
+export type ConnectionSpawner = (spec: SubprocessSpawnSpec) => SubprocessHandle
 
 const writeConnectionMessage: ConnectionWriter = (stdin, message, done) => {
   stdin.write(encodeMessage(message), done)
 }
 
-/**
- * Terminate one Windows process tree and wait for taskkill to finish.
- * @param pid - root process id.
- * @param run - command runner; tests inject results without requiring Windows.
- */
-export function taskkillProcessTree(
-  pid: number,
-  run: TaskkillRunner = spawnSync,
-): void {
-  const result = run('taskkill', ['/PID', String(pid), '/T', '/F'], { stdio: 'ignore' })
-  if (result.error !== undefined) throw result.error
-  if (result.status === TASKKILL_TREE_NOT_FOUND_STATUS) return
-  if (result.status !== 0) throw new Error(`taskkill exited with status ${String(result.status)}`)
-}
-
-/**
- * Signal one POSIX process group through an injectable host primitive.
- * @param target - negative process-group id.
- * @param signal - requested signal.
- * @param run - host signal runner; tests inject it without touching real processes.
- */
-export function signalProcessGroup(
-  target: number,
-  signal: NodeJS.Signals,
-  run: ProcessSignalRunner = processSignalRunner,
-): void {
-  run(target, signal)
-}
-
-/**
- * Wait until a process-tree liveness probe reports exit.
- * @param isAlive - process-tree liveness probe.
- * @param signal - optional bound for the wait.
- * @param yieldNow - event-loop yield primitive.
- * @returns `true` when the tree exited, or `false` when the signal aborted first.
- */
-export async function waitForTreeExit(
-  isAlive: () => boolean,
-  signal?: AbortSignal,
-  yieldNow: () => Promise<unknown> = yieldToEventLoop,
-): Promise<boolean> {
-  while (isAlive()) {
-    if (signal?.aborted) return false
-    await yieldNow()
-  }
-  return true
-}
-
-/**
- * Signal a detached process tree with platform-correct semantics. POSIX falls back to the direct
- * child; Windows requires taskkill to reach the full tree.
- * @param platform - host platform.
- * @param pid - detached root process id.
- * @param signal - requested termination signal.
- * @param operations - host operations.
- */
-export function signalProcessTree(
-  platform: NodeJS.Platform,
-  pid: number,
-  signal: NodeJS.Signals,
-  operations: ProcessTreeOperations,
-): void {
-  if (platform === 'win32') {
-    operations.taskkill(pid)
-    return
-  }
-  try {
-    operations.signal(-pid, signal)
-  } catch {
-    try {
-      operations.killChild(signal)
-    } catch {
-      // The direct child already exited; teardown remains idempotent.
-    }
-  }
-}
-
 /** A live JSON-RPC endpoint bound to one child process. */
 export class LspConnection {
-  private readonly child: ChildProcessByStdio<Writable, Readable, Readable>
+  private readonly handle: SubprocessHandle
+  private readonly stdin: Writable
   private readonly decoder: MessageDecoder
   private readonly pending = new Map<number, Pending>()
   private nextId = 1
-  private stderr = Buffer.alloc(0)
   private closeReason: Error | undefined
   /** Set once the process has fully exited; the instance awaits it during teardown. */
   readonly closed: Promise<void>
 
   /**
    * @param spec - how to launch the server and answer its config requests.
+   * @param spawner - the subprocess seam's spawn (the provider passes `ctx.subprocess.spawn`).
    * @param onServerRequest - answers a server→client request; rejects to send an error response.
    * @param writer - message writer; tests inject callback failures without relying on OS pipe races.
    */
   constructor(
-    private readonly spec: ConnectionSpec,
+    spec: ConnectionSpec,
+    spawner: ConnectionSpawner,
     private readonly onServerRequest: (method: string, params: unknown) => Promise<unknown>,
     private readonly writer: ConnectionWriter = writeConnectionMessage,
   ) {
     this.decoder = new MessageDecoder(spec.maxMessageBytes)
-    // `detached` gives teardown a process-tree root: POSIX signals its negative process-group id,
-    // while Windows passes the root pid to taskkill /T so helpers such as tsserver cannot outlive it.
-    this.child = spawn(spec.command, [...spec.args], {
+    // stdin/stdout are piped protocol streams this endpoint frames itself;
+    // stderr is a collected diagnostic tail (no spill — the bounded tail IS
+    // the contract). The seam owns detachment and tree-scoped signalling.
+    this.handle = spawner({
+      argv: [spec.command, ...spec.args],
       cwd: spec.cwd,
+      stdio: {
+        stdin: 'pipe',
+        stdout: 'pipe',
+        stderr: { maxBytes: spec.maxStderrBytes },
+      },
+      graceMs: spec.pipeDrainGraceMs,
       env: spec.env,
-      stdio: ['pipe', 'pipe', 'pipe'],
-      detached: true,
     })
+    /* v8 ignore start -- 'pipe' dispositions expose both streams by the seam contract; defensive. */
+    if (this.handle.stdin === undefined || this.handle.stdout === undefined) {
+      throw new Error('lsp-local: subprocess implementation dropped a piped protocol stream')
+    }
+    /* v8 ignore stop */
+    this.stdin = this.handle.stdin
     this.closed = new Promise<void>((resolve) => {
-      this.child.on('close', () => {
+      const close = (): void => {
         const reason = this.closeReason ?? new Error(this.exitMessage())
         // Record the reason so any request issued AFTER close rejects immediately instead of hanging
         // (a closed process sends no further responses).
         this.closeReason = reason
         this.failAll(reason)
         resolve()
+      }
+      this.handle.done.then(close, (error: unknown) => {
+        // A spawn-level failure never produces a close event; the rejection is
+        // the fatal cause and the close boundary at once.
+        this.fail(asError(error))
+        close()
       })
     })
-    this.child.on('error', (error) => { this.fail(error) })
     // Child stdin can fail while the process itself remains alive (for example, a server closes fd
     // 0). Treat that as a fatal connection error so pending requests reject immediately instead of
     // waiting for a process-close event that may never arrive.
-    this.child.stdin.on('error', (error) => { this.fail(error) })
-    this.child.stdout.on('data', (chunk: Buffer) => { this.onStdout(chunk) })
-    this.child.stderr.on('data', (chunk: Buffer) => { this.onStderr(chunk) })
+    this.stdin.on('error', (error) => { this.fail(error) })
+    this.handle.stdout.on('data', (chunk: Buffer) => { this.onStdout(chunk) })
   }
 
   /** The child's pid, or `-1` when the spawn produced no pid (so signalling is a no-op). */
   get pid(): number {
-    /* v8 ignore next -- the `-1` fallback only applies to a spawn that produced no pid; defensive. */
-    return this.child.pid ?? -1
+    return this.handle.pid
   }
 
   /** The retained stderr tail, for diagnostics on a failed server. */
   get stderrTail(): string {
-    return this.stderr.toString('utf8')
+    /* v8 ignore next -- the collect disposition always exposes a stderr reader; defensive. */
+    return this.handle.collected.stderr?.readFrom(0).text ?? ''
   }
 
   /** Whether the transport has failed even if the child close event has not arrived yet. */
@@ -289,14 +207,14 @@ export class LspConnection {
     return this.nextId
   }
 
-  /** Request termination of the server's process tree. */
+  /** Request termination of the server's process tree (SIGTERM, no escalation). */
   terminate(): void {
-    this.signalTree('SIGTERM')
+    this.handle.kill('SIGTERM')
   }
 
   /** Force termination of the server's process tree. */
   kill(): void {
-    this.signalTree('SIGKILL')
+    this.handle.kill('SIGKILL')
   }
 
   /**
@@ -305,39 +223,7 @@ export class LspConnection {
    * @returns `true` when the tree exited, or `false` when the signal aborted first.
    */
   async waitForProcessTreeExit(signal?: AbortSignal): Promise<boolean> {
-    return await waitForTreeExit(this.processTreeAlive.bind(this), signal)
-  }
-
-  /** Signal the whole process tree. */
-  private signalTree(sig: NodeJS.Signals): void {
-    const pid = this.child.pid
-    if (pid === undefined) return
-    signalProcessTree(process.platform, pid, sig, {
-      signal: signalProcessGroup,
-      killChild: this.child.kill.bind(this.child),
-      taskkill: taskkillProcessTree,
-    })
-  }
-
-  /** Whether the detached tree's root or POSIX process group is still alive. */
-  private processTreeAlive(): boolean {
-    const pid = this.child.pid
-    /* v8 ignore next -- only an asynchronous spawn failure omits pid; its close path owns cleanup. */
-    if (pid === undefined) return false
-    try {
-      process.kill(-pid, 0)
-      return true
-    } catch (error) {
-      const code = (error as NodeJS.ErrnoException).code
-      /* v8 ignore next -- POSIX reports an absent group as ESRCH, but child-reaping timing makes
-         whether lifecycle tests observe this branch platform-dependent. */
-      if (code === 'ESRCH') return false
-      /* v8 ignore start -- EPERM and non-POSIX negative-pid failures are platform defenses; CI runs
-         process-group lifecycle tests on POSIX hosts where absence reports ESRCH. */
-      if (code === 'EPERM') return true
-      return this.child.exitCode === null && this.child.signalCode === null
-      /* v8 ignore stop */
-    }
+    return await this.handle.waitForExit(signal)
   }
 
   private onStdout(chunk: Buffer): void {
@@ -348,28 +234,12 @@ export class LspConnection {
       // A framing/JSON failure corrupts the stream position irrecoverably: fail the instance and
       // SIGKILL the whole group so helper processes don't outlive the leader.
       this.fail(asError(error))
-      this.signalTree('SIGKILL')
+      this.handle.kill('SIGKILL')
       return
     }
     for (const message of messages) this.dispatch(message)
   }
 
-  private onStderr(chunk: Buffer): void {
-    // Retain the TAIL, not the prefix: a language server's fatal diagnostic usually appears just
-    // before it exits, so the final bounded segment is the useful one.
-    const cap = this.spec.maxStderrBytes
-    if (chunk.length >= cap) {
-      // Copy the bounded suffix so retaining it does not pin an arbitrarily large incoming buffer.
-      this.stderr = Buffer.from(chunk.subarray(chunk.length - cap))
-      return
-    }
-    const retainedBytes = Math.min(this.stderr.length, cap - chunk.length)
-    this.stderr = Buffer.concat([
-      this.stderr.subarray(this.stderr.length - retainedBytes),
-      chunk,
-    ], retainedBytes + chunk.length)
-  }
-
   private dispatch(message: unknown): void {
     if (message === null || typeof message !== 'object') return
     const frame = message as Record<string, unknown>
@@ -423,7 +293,7 @@ export class LspConnection {
         reject(error)
       }
       try {
-        this.writer(this.child.stdin, message, done)
+        this.writer(this.stdin, message, done)
       /* v8 ignore start -- Node stream write failures are callback-delivered; this guards a
          nonconforming Writable implementation throwing synchronously. */
       } catch (error) {

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

@@ -25,6 +25,8 @@ import { MAX_TIMER_DELAY_MS } from '@deepseek-ai/dsh-timeout'
 import { abortable, abortError } from './abort.ts'
 import { canonicalizeWorkspace, readHostSource } from './host.ts'
 import { LspInstance } from './instance.ts'
+import type { ConnectionSpawner } from './connection.ts'
+import { scrubbedParentEnv } from '@deepseek-ai/dsh-subprocess'
 import type { InstanceSpec } from './instance.ts'
 
 export { canonicalizeWorkspace, readHostSource } from './host.ts'
@@ -44,10 +46,10 @@ export { LspConnection } from './connection.ts'
 export const name = 'lsp-local'
 
 /** Services required by this plugin. */
-export const inject = ['lsp']
+export const inject = ['lsp', 'subprocess']
 
 /** Credential-shaped ambient env vars are NOT forwarded to the child by default. */
-const SENSITIVE_ENV_PATTERN = /KEY|SECRET|TOKEN/i
+
 
 const DEFAULT_MAX_MESSAGE_BYTES = 16_000_000
 const DEFAULT_MAX_STDERR_BYTES = 1_000_000
@@ -127,7 +129,7 @@ export function apply(ctx: Context, config: Config): void {
     validateServerConfig(providerId, resolved)
     const childEnv = buildChildEnv(resolved.env)
     const executable = resolveExecutable(resolved.command, childEnv)
-    return new LocalLspProvider(providerId, resolved, childEnv, executable)
+    return new LocalLspProvider(providerId, resolved, childEnv, executable, spec => ctx.subprocess.spawn(spec))
   })
 
   ctx.effect(() => {
@@ -189,6 +191,7 @@ class LocalLspProvider implements LspProvider {
     private readonly config: ResolvedServerConfig,
     private readonly childEnv: Record<string, string>,
     private readonly executable: string,
+    private readonly spawner: ConnectionSpawner,
   ) {
     this.id = LspProviderId(providerId)
     this.extensionToLanguage = config.extensionToLanguage
@@ -282,10 +285,12 @@ class LocalLspProvider implements LspProvider {
       initializationOptions: this.config.initializationOptions,
       maxMessageBytes: this.config.maxMessageBytes,
       maxStderrBytes: this.config.maxStderrBytes,
+      // Exit observation must never be slower than the escalation it feeds.
+      pipeDrainGraceMs: this.config.killGraceMs,
       shutdownTimeoutMs: this.config.shutdownTimeoutMs,
       killGraceMs: this.config.killGraceMs,
     }
-    return new LspInstance(spec)
+    return new LspInstance(spec, this.spawner)
   }
 
   /** Dispose every live instance and block further queries. */
@@ -302,12 +307,9 @@ class LocalLspProvider implements LspProvider {
   }
 }
 
-/** The ambient env minus credential-shaped vars, plus the config's explicit env. */
+/** The seam's scrubbed parent env (credential-shaped and DSH_* names dropped), plus the config's explicit env. */
 function buildChildEnv(extra: Record<string, string>): Record<string, string> {
-  const scrubbed = Object.entries(process.env).filter(
-    ([key, value]) => value !== undefined && !SENSITIVE_ENV_PATTERN.test(key),
-  ) as [string, string][]
-  return { ...Object.fromEntries(scrubbed), ...extra }
+  return { ...scrubbedParentEnv(), ...extra }
 }
 
 /**

packages/lsp/lsp-local/src/instance.ts

@@ -17,7 +17,7 @@ import type {
 import { deadline } from '@deepseek-ai/dsh-timeout'
 import { abortable, abortError } from './abort.ts'
 import { LspConnection } from './connection.ts'
-import type { ConnectionSpec, ConnectionWriter } from './connection.ts'
+import type { ConnectionSpawner, ConnectionSpec, ConnectionWriter } from './connection.ts'
 import type { HostSource } from './host.ts'
 import type { WireInitializeResult, WireServerCapabilities } from './protocol.ts'
 import {
@@ -67,10 +67,11 @@ export class LspInstance {
 
   /**
    * @param spec - the launch, initialize, and teardown parameters.
+   * @param spawner - the subprocess seam's spawn function.
    * @param writer - optional connection writer used by transport conformance tests.
    */
-  constructor(private readonly spec: InstanceSpec, writer?: ConnectionWriter) {
-    this.connection = new LspConnection(spec, (method, params) => this.answerServerRequest(method, params), writer)
+  constructor(private readonly spec: InstanceSpec, spawner: ConnectionSpawner, writer?: ConnectionWriter) {
+    this.connection = new LspConnection(spec, spawner, (method, params) => this.answerServerRequest(method, params), writer)
     this.ready = this.initialize()
     // A handshake rejection must not surface as an unhandled rejection before the first query awaits
     // it; queries attach the real handler.

packages/lsp/lsp-local/tests/built-lib.e2e.ts

@@ -16,7 +16,8 @@ import { afterAll, beforeAll, describe, expect, it } from 'vitest'
 
 const pkgDir = fileURLToPath(new URL('..', import.meta.url))
 const seamLib = join(pkgDir, '../lsp/lib/index.js')
-const built = existsSync(join(pkgDir, 'lib/index.js')) && existsSync(seamLib)
+const subprocessLib = join(pkgDir, '../../subprocess/subprocess-local/lib/index.js')
+const built = existsSync(join(pkgDir, 'lib/index.js')) && existsSync(seamLib) && existsSync(subprocessLib)
 
 const fixtureServer = fileURLToPath(new URL('./fixture-server.ts', import.meta.url))
 
@@ -41,8 +42,10 @@ describe.skipIf(!built)('built lib real load path (plain node)', () => {
       const { Context } = await import('cordis')
       const { default: Lsp } = await import('@deepseek-ai/dsh-lsp')
       const LspLocal = await import('@deepseek-ai/dsh-lsp-local')
+      const { default: LocalSubprocessService } = await import('@deepseek-ai/dsh-subprocess-local')
       const ctx = new Context()
       await ctx.plugin(Lsp)
+      await ctx.plugin(LocalSubprocessService)
       await ctx.plugin(LspLocal, {
         servers: {
           fake: {

packages/lsp/lsp-local/tests/connection.spec.ts

@@ -1,18 +1,8 @@
-import { afterEach, describe, expect, it, vi } from 'vitest'
+import { afterEach, describe, expect, it } from 'vitest'
 import { fileURLToPath } from 'node:url'
 import { LspConnection } from '@deepseek-ai/dsh-lsp-local'
-import {
-  signalProcessGroup,
-  signalProcessTree,
-  taskkillProcessTree,
-  waitForTreeExit,
-} from '@deepseek-ai/dsh-lsp-local/src/connection.ts'
-import type {
-  ConnectionWriter,
-  ProcessSignalRunner,
-  ProcessTreeOperations,
-  TaskkillRunner,
-} from '@deepseek-ai/dsh-lsp-local/src/connection.ts'
+import type { ConnectionWriter } from '@deepseek-ai/dsh-lsp-local/src/connection.ts'
+import { spawnSubprocess } from '@deepseek-ai/dsh-subprocess-local/src/spawn.ts'
 
 const fixtureServer = fileURLToPath(new URL('./fixture-server.ts', import.meta.url))
 
@@ -42,8 +32,9 @@ function connect(
     env: { ...process.env as Record<string, string>, ...env },
     maxMessageBytes: 16_000_000,
     maxStderrBytes: 100_000,
+    pipeDrainGraceMs: 3_000,
     configuration: { setting: 42 },
-  }, (method, params) => {
+  }, spawnSubprocess, (method, params) => {
     seen?.push({ method, params })
     return onServerRequest(method, params)
   })
@@ -151,8 +142,9 @@ function connectScript(script: string, maxStderrBytes = 100_000, writer?: Connec
     env: { ...process.env as Record<string, string> },
     maxMessageBytes: 16_000_000,
     maxStderrBytes,
+    pipeDrainGraceMs: 3_000,
     configuration: null,
-  }, () => Promise.resolve(null), writer)
+  }, spawnSubprocess, () => Promise.resolve(null), writer)
   open.push(conn)
   return conn
 }
@@ -166,8 +158,9 @@ describe('LspConnection edge behavior', () => {
       env: {},
       maxMessageBytes: 1000,
       maxStderrBytes: 1000,
+      pipeDrainGraceMs: 3_000,
       configuration: null,
-    }, () => Promise.resolve(null))
+    }, spawnSubprocess, () => Promise.resolve(null))
     open.push(conn)
     await expect(conn.request('initialize', {})).rejects.toThrow()
   })
@@ -248,72 +241,6 @@ describe('LspConnection edge behavior', () => {
   })
 })
 
-describe('process-tree signaling', () => {
-  it('forwards POSIX process-group signals through the host runner', () => {
-    const run: ProcessSignalRunner = vi.fn(() => true)
-    signalProcessGroup(-42, 'SIGKILL', run)
-    expect(run).toHaveBeenCalledWith(-42, 'SIGKILL')
-  })
-
-  it('waits for tree exit and stops when its bound aborts', async () => {
-    const isAlive = vi.fn()
-      .mockReturnValueOnce(true)
-      .mockReturnValue(false)
-    const yieldNow = vi.fn(() => Promise.resolve())
-    await expect(waitForTreeExit(isAlive, undefined, yieldNow)).resolves.toBe(true)
-    expect(yieldNow).toHaveBeenCalledOnce()
-
-    const controller = new AbortController()
-    controller.abort()
-    await expect(waitForTreeExit(() => true, controller.signal, yieldNow)).resolves.toBe(false)
-  })
-
-  it('uses taskkill for a Windows tree and a negative pid for a POSIX group', () => {
-    const operations = fakeProcessTreeOperations()
-    signalProcessTree('win32', 42, 'SIGTERM', operations)
-    expect(operations.taskkill).toHaveBeenCalledWith(42)
-    expect(operations.signal).not.toHaveBeenCalled()
-
-    signalProcessTree('linux', 42, 'SIGKILL', operations)
-    expect(operations.signal).toHaveBeenCalledWith(-42, 'SIGKILL')
-  })
-
-  it('surfaces a Windows taskkill failure without downgrading to the direct child', () => {
-    const fallback = fakeProcessTreeOperations()
-    vi.mocked(fallback.taskkill).mockImplementation(() => { throw new Error('taskkill unavailable') })
-    expect(() => { signalProcessTree('win32', 42, 'SIGTERM', fallback) }).toThrow(/taskkill unavailable/)
-    expect(fallback.killChild).not.toHaveBeenCalled()
-  })
-
-  it('tolerates a POSIX tree-signaling race after the direct child is already gone', () => {
-    const posixGone = fakeProcessTreeOperations()
-    vi.mocked(posixGone.signal).mockImplementation(() => { throw new Error('group gone') })
-    vi.mocked(posixGone.killChild).mockImplementation(() => { throw new Error('child gone') })
-    expect(() => { signalProcessTree('linux', 42, 'SIGKILL', posixGone) }).not.toThrow()
-  })
-
-  it('runs taskkill for the full tree, accepts an absent tree, and rejects command failures', () => {
-    const success: TaskkillRunner = vi.fn(() => ({ status: 0 }))
-    taskkillProcessTree(42, success)
-    expect(success).toHaveBeenCalledWith('taskkill', ['/PID', '42', '/T', '/F'], { stdio: 'ignore' })
-
-    expect(() => { taskkillProcessTree(42, () => ({ status: 128 })) }).not.toThrow()
-
-    const spawnFailure = new Error('cannot spawn taskkill')
-    expect(() => { taskkillProcessTree(42, () => ({ status: null, error: spawnFailure })) }).toThrow(spawnFailure)
-    expect(() => { taskkillProcessTree(42, () => ({ status: 1 })) }).toThrow(/status 1/)
-  })
-})
-
-/** Create observable process-tree operations without touching host processes. */
-function fakeProcessTreeOperations(): ProcessTreeOperations {
-  return {
-    signal: vi.fn(),
-    killChild: vi.fn(),
-    taskkill: vi.fn(),
-  }
-}
-
 /** Poll a predicate until it holds or a deadline elapses. */
 async function waitFor(predicate: () => boolean, timeoutMs = 3000): Promise<void> {
   const start = Date.now()

packages/lsp/lsp-local/tests/instance.spec.ts

@@ -9,6 +9,7 @@ import type { ConnectionWriter } from '@deepseek-ai/dsh-lsp-local/src/connection
 import { escalateProcessTree } from '@deepseek-ai/dsh-lsp-local/src/instance.ts'
 import type { InstanceSpec } from '@deepseek-ai/dsh-lsp-local/src/instance.ts'
 import type { LspProviderQuery, LspQueryResult } from '@deepseek-ai/dsh-lsp'
+import { spawnSubprocess } from '@deepseek-ai/dsh-subprocess-local/src/spawn.ts'
 
 const fixtureServer = fileURLToPath(new URL('./fixture-server.ts', import.meta.url))
 
@@ -43,10 +44,11 @@ function makeInstance(
     initializationOptions: { init: true },
     maxMessageBytes: 16_000_000,
     maxStderrBytes: 100_000,
+    pipeDrainGraceMs: 200,
     shutdownTimeoutMs: 200,
     killGraceMs: 200,
     ...overrides,
-  }, writer)
+  }, spawnSubprocess, writer)
   live.push(instance)
   return instance
 }
@@ -72,10 +74,11 @@ function scriptInstance(script: string, overrides: Partial<InstanceSpec> = {}):
     initializationOptions: null,
     maxMessageBytes: 16_000_000,
     maxStderrBytes: 100_000,
+    pipeDrainGraceMs: 150,
     shutdownTimeoutMs: 150,
     killGraceMs: 150,
     ...overrides,
-  })
+  }, spawnSubprocess)
   live.push(instance)
   return instance
 }

packages/lsp/lsp-local/tests/lifecycle.spec.ts

@@ -7,6 +7,7 @@ import { pathToFileURL, fileURLToPath } from 'node:url'
 import { Context } from 'cordis'
 import Lsp, { type LspProvider, type LspQueryRequest, type LspQueryResult } from '@deepseek-ai/dsh-lsp'
 import { deadline } from '@deepseek-ai/dsh-timeout'
+import LocalSubprocessService from '@deepseek-ai/dsh-subprocess-local'
 import * as LspLocal from '@deepseek-ai/dsh-lsp-local'
 import type { LspLocalServerConfig } from '@deepseek-ai/dsh-lsp-local'
 
@@ -45,6 +46,7 @@ async function mount(
 ): Promise<Context> {
   const ctx = new Context()
   await ctx.plugin(Lsp)
+  await ctx.plugin(LocalSubprocessService)
   const register = ctx.lsp.registerProvider.bind(ctx.lsp)
   const registrationSpy = captureProvider === undefined
     ? undefined
@@ -76,6 +78,7 @@ describe('lsp-local end to end over a fake server', () => {
     await writeFile(join(ws, 'a.py'), 'x = 1\n')
     const ctx = new Context()
     await ctx.plugin(Lsp)
+    await ctx.plugin(LocalSubprocessService)
     await ctx.plugin(LspLocal, {
       servers: {
         typescript: fakeServer({ LSP_FAKE_HOVER: JSON.stringify({ contents: 'ts' }) }),
@@ -318,6 +321,7 @@ describe('lsp-local end to end over a fake server', () => {
   it('rejects at load when the command is not found', async () => {
     const ctx = new Context()
     await ctx.plugin(Lsp)
+    await ctx.plugin(LocalSubprocessService)
     await expect(ctx.plugin(LspLocal, {
       servers: {
         missing: {

packages/lsp/lsp-local/tests/provider.spec.ts

@@ -3,6 +3,7 @@ import { chmod, mkdtemp, mkdir, rm, writeFile, realpath } from 'node:fs/promises
 import { tmpdir } from 'node:os'
 import { delimiter, join } from 'node:path'
 import { Context } from 'cordis'
+import LocalSubprocessService from '@deepseek-ai/dsh-subprocess-local'
 import Lsp, { type LspQueryRequest } from '@deepseek-ai/dsh-lsp'
 import * as LspLocal from '@deepseek-ai/dsh-lsp-local'
 import type { Config, LspLocalServerConfig } from '@deepseek-ai/dsh-lsp-local'
@@ -42,6 +43,7 @@ describe('lsp-local provider resolution', () => {
 
     const ctx = new Context()
     await ctx.plugin(Lsp)
+    await ctx.plugin(LocalSubprocessService)
     await expect(ctx.plugin(LspLocal, config('onpath', {
       command: 'fake-lsp',
       args: [],
@@ -54,6 +56,7 @@ describe('lsp-local provider resolution', () => {
   it('skips empty PATH segments and fails when the command is absent', async () => {
     const ctx = new Context()
     await ctx.plugin(Lsp)
+    await ctx.plugin(LocalSubprocessService)
     await expect(ctx.plugin(LspLocal, config('nope', {
       command: 'fake-lsp',
       args: [],
@@ -67,6 +70,7 @@ describe('lsp-local provider resolution', () => {
     // Use a server that never emits results and dispose the plugin, then confirm queries are refused.
     const ctx = new Context()
     await ctx.plugin(Lsp)
+    await ctx.plugin(LocalSubprocessService)
     // Grab the provider instance by registering, then dispose the whole plugin fiber.
     const lsp = ctx.lsp
     const fiber = await ctx.plugin(LspLocal, config('disp', {
@@ -83,6 +87,7 @@ describe('lsp-local provider resolution', () => {
   it('rejects a nonpositive teardown budget at load', async () => {
     const ctx = new Context()
     await ctx.plugin(Lsp)
+    await ctx.plugin(LocalSubprocessService)
     await expect(ctx.plugin(LspLocal, config('bad-budget', {
       command: process.execPath,
       args: ['-e', ''],
@@ -95,6 +100,7 @@ describe('lsp-local provider resolution', () => {
   it('rejects a nonpositive byte cap at load', async () => {
     const ctx = new Context()
     await ctx.plugin(Lsp)
+    await ctx.plugin(LocalSubprocessService)
     await expect(ctx.plugin(LspLocal, config('bad-cap', {
       command: process.execPath,
       args: ['-e', ''],
@@ -107,6 +113,7 @@ describe('lsp-local provider resolution', () => {
   it.each(['shutdownTimeoutMs', 'killGraceMs'] as const)('rejects %s above Node timer range at load', async (name) => {
     const ctx = new Context()
     await ctx.plugin(Lsp)
+    await ctx.plugin(LocalSubprocessService)
     await expect(ctx.plugin(LspLocal, config('bad-timer', {
       command: process.execPath,
       args: ['-e', ''],
@@ -122,6 +129,7 @@ describe('lsp-local provider resolution', () => {
     await writeFile(notExe, 'plain text, not executable')
     const ctx = new Context()
     await ctx.plugin(Lsp)
+    await ctx.plugin(LocalSubprocessService)
     await expect(ctx.plugin(LspLocal, config('abs-bad', {
       command: notExe,
       args: [],
@@ -133,6 +141,7 @@ describe('lsp-local provider resolution', () => {
   it('rejects an executable directory as a command at load', async () => {
     const ctx = new Context()
     await ctx.plugin(Lsp)
+    await ctx.plugin(LocalSubprocessService)
     await expect(ctx.plugin(LspLocal, config('abs-directory', {
       command: ws,
       args: [],
@@ -144,6 +153,7 @@ describe('lsp-local provider resolution', () => {
   it('rejects an empty server table at load', async () => {
     const ctx = new Context()
     await ctx.plugin(Lsp)
+    await ctx.plugin(LocalSubprocessService)
     await expect(ctx.plugin(LspLocal, { servers: {} })).rejects.toThrow(/servers must contain at least one server/)
     await ctx.fiber.dispose()
   })
@@ -151,6 +161,7 @@ describe('lsp-local provider resolution', () => {
   it('rejects an empty server id at load', async () => {
     const ctx = new Context()
     await ctx.plugin(Lsp)
+    await ctx.plugin(LocalSubprocessService)
     await expect(ctx.plugin(LspLocal, config('', {
       command: process.execPath,
       extensionToLanguage: { '.ts': 'typescript' },
@@ -161,6 +172,7 @@ describe('lsp-local provider resolution', () => {
   it('resolves every executable before publishing any provider', async () => {
     const ctx = new Context()
     await ctx.plugin(Lsp)
+    await ctx.plugin(LocalSubprocessService)
     await expect(ctx.plugin(LspLocal, {
       servers: {
         valid: { command: process.execPath, extensionToLanguage: { '.ts': 'typescript' } },
@@ -174,6 +186,7 @@ describe('lsp-local provider resolution', () => {
   it('rolls back earlier registrations when a later server conflicts', async () => {
     const ctx = new Context()
     await ctx.plugin(Lsp)
+    await ctx.plugin(LocalSubprocessService)
     await expect(ctx.plugin(LspLocal, {
       servers: {
         first: { command: process.execPath, extensionToLanguage: { '.ts': 'typescript' } },

packages/lsp/lsp-local/tests/typescript-server.e2e.ts

@@ -10,6 +10,7 @@ import { mkdtemp, mkdir, rm, writeFile, realpath } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { Context } from 'cordis'
+import LocalSubprocessService from '@deepseek-ai/dsh-subprocess-local'
 import Lsp, { type LspQueryRequest, type LspQueryResult } from '@deepseek-ai/dsh-lsp'
 import * as LspLocal from '@deepseek-ai/dsh-lsp-local'
 
@@ -52,6 +53,7 @@ beforeAll(async () => {
 
   ctx = new Context()
   await ctx.plugin(Lsp)
+  await ctx.plugin(LocalSubprocessService)
   await ctx.plugin(LspLocal, {
     servers: {
       typescript: {

packages/lsp/lsp-local/tsconfig.json

@@ -29,6 +29,9 @@
     {
       "path": "../lsp"
     },
+    {
+      "path": "../../subprocess/subprocess"
+    },
     {
       "path": "../../support/invariants"
     }

packages/lsp/tool-lsp/tests/integration.spec.ts

@@ -6,6 +6,7 @@ import { pathToFileURL } from 'node:url'
 import { Context } from 'cordis'
 import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
 import ToolRegistry from '@deepseek-ai/dsh-tools'
+import LocalSubprocessService from '@deepseek-ai/dsh-subprocess-local'
 import Lsp from '@deepseek-ai/dsh-lsp'
 import * as LspLocal from '@deepseek-ai/dsh-lsp-local'
 import * as TimeoutPolicy from '@deepseek-ai/dsh-timeout-policy'
@@ -49,6 +50,7 @@ async function mount(hang: boolean, timeoutMs?: number): Promise<Context> {
   await ctx.plugin(SystemPrompt)
   await ctx.plugin(ToolRegistry)
   await ctx.plugin(Lsp)
+  await ctx.plugin(LocalSubprocessService)
   await ctx.plugin(LspLocal, {
     servers: {
       inline: {