From 12a7e384174d34195a30bb8083bba2d14f117e51 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Sun, 26 Jul 2026 14:07:42 +0800 Subject: [PATCH] feat(subprocess): reshape the seam Node-ward for multi-consumer use MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 | 58 ++- .../subprocess/subprocess-local/src/index.ts | 30 +- .../subprocess/subprocess-local/src/spawn.ts | 339 +++++++++++++----- .../subprocess-local/tests/local.spec.ts | 13 +- .../subprocess-local/tests/spawn.spec.ts | 279 ++++++++++---- packages/subprocess/subprocess/src/index.ts | 68 +++- packages/subprocess/subprocess/src/types.ts | 200 ++++++++--- .../subprocess/tests/service.spec.ts | 60 ++-- 8 files changed, 786 insertions(+), 261 deletions(-) diff --git a/packages/bash/bash-local/src/index.ts b/packages/bash/bash-local/src/index.ts index 463e7a96d4..9d9ed676e2 100644 --- a/packages/bash/bash-local/src/index.ts +++ b/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> & Pick +/** 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 { // 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 }, } diff --git a/packages/subprocess/subprocess-local/src/index.ts b/packages/subprocess/subprocess-local/src/index.ts index a5256f325b..d76e5cf410 100644 --- a/packages/subprocess/subprocess-local/src/index.ts +++ b/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() - /** 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[] = [] 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) }, diff --git a/packages/subprocess/subprocess-local/src/spawn.ts b/packages/subprocess/subprocess-local/src/spawn.ts index 95b391ea4b..f939b9e774 100644 --- a/packages/subprocess/subprocess-local/src/spawn.ts +++ b/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>, 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 /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 spawnProcess(spec: SubprocessSpawnSpec, internals: SpawnInternals = {}): SubprocessHandle { +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 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 env = childEnv(spec.env, spec.dshEnv) - const child: ChildProcessByStdio = 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 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 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 env = childEnv(spec.env, spec.dshEnv) + 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 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 => { - 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. + 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 - killGroup(pid, 'SIGTERM') - graceTimer = setTimeout(() => { killGroup(pid, 'SIGKILL') }, spec.graceMs) + signalTree(platform, pid, sig, child, taskkill) + } + + const terminate = (): void => { + if (graceTimer !== undefined) return // escalation already in flight + if (settled) return + 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((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 => { + 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 => { + if (settled) return true + let timer: NodeJS.Timeout | undefined + const timeout = new Promise((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 | undefined + const dispose = (graces: SubprocessDisposeGraces): Promise => (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, + } } diff --git a/packages/subprocess/subprocess-local/tests/local.spec.ts b/packages/subprocess/subprocess-local/tests/local.spec.ts index d6f81042b5..ca723fdb87 100644 --- a/packages/subprocess/subprocess-local/tests/local.spec.ts +++ b/packages/subprocess/subprocess-local/tests/local.spec.ts @@ -7,9 +7,11 @@ function spec(command: string, overrides: Partial = {}): 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() }) diff --git a/packages/subprocess/subprocess-local/tests/spawn.spec.ts b/packages/subprocess/subprocess-local/tests/spawn.spec.ts index 1bc2b27498..2f48b0fef3 100644 --- a/packages/subprocess/subprocess-local/tests/spawn.spec.ts +++ b/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[0]> = {}) { +type SpecOverrides = Partial[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 { async function waitForStdout(running: SubprocessHandle, expected: string, timeoutMs = 5_000): Promise { 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 { const deadline = Date.now() + timeoutMs while (Date.now() < deadline) { @@ -82,9 +102,9 @@ async function waitForPidFile(path: string, timeoutMs = 5_000): Promise 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((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') diff --git a/packages/subprocess/subprocess/src/index.ts b/packages/subprocess/subprocess/src/index.ts index 890cf74e61..bbcca90133 100644 --- a/packages/subprocess/subprocess/src/index.ts +++ b/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 { + const env: Record = {} + 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 } diff --git a/packages/subprocess/subprocess/src/types.ts b/packages/subprocess/subprocess/src/types.ts index a5ec7f4e55..4c667919c7 100644 --- a/packages/subprocess/subprocess/src/types.ts +++ b/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. */ + /** Per-stream stdio dispositions. */ + stdio: SubprocessStdio + /** + * 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). + */ graceMs: number /** - * Abort signal — kills the process group when it fires. The caller owns - * deadlines and cause classification; this seam only reacts to the abort. + * 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. */ signal?: AbortSignal | undefined /** - * Bytes to write to the child's stdin, then close it. Absent (or empty) - * leaves stdin closed/empty. - */ - stdin?: string | 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 | 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 +} + /** - * A live child process. `kill()` starts the group SIGTERM→grace→SIGKILL - * escalation; buffered output remains readable after exit. + * 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 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 - /** 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 + /** + * 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 } diff --git a/packages/subprocess/subprocess/tests/service.spec.ts b/packages/subprocess/subprocess/tests/service.spec.ts index 630aa35f1a..d033e24a31 100644 --- a/packages/subprocess/subprocess/tests/service.spec.ts +++ b/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 + } }) })