From b2852bbece7e9e4dba9b61a0cdb2e0978323b595 Mon Sep 17 00:00:00 2001 From: Stephen Hellicar Date: Mon, 3 Aug 2026 01:18:00 +1000 Subject: [PATCH 01/11] Join ExecV3 pipeline stages with real OS pipes A pipeline cancelled mid-flight left its run promise unsettled, so the CLI hung with every process already dead. The stages were joined by in-process streams that simulated backpressure and SIGPIPE. The kernel now provides both. --- packages/claude-sdk-tools/CHANGELOG.md | 1 + packages/claude-sdk-tools/changes.jsonl | 1 + .../src/ExecV3/runPipeline.ts | 118 ++++------- packages/claude-sdk-tools/test/Exec.spec.ts | 7 +- packages/claude-sdk-tools/test/ExecV3.spec.ts | 3 +- .../test/ExecV3/duration.spec.ts | 20 +- .../claude-sdk-tools/test/FakeExecutor.ts | 32 ++- .../test/integration/AzSessionCache.spec.ts | 7 +- .../integration/pipeline-teardown.spec.ts | 41 ++++ packages/exec-core/CHANGELOG.md | 6 + packages/exec-core/changes.jsonl | 3 + packages/exec-core/src/Executor.ts | 200 ++++++++++++++---- packages/exec-core/src/entry/index.ts | 7 +- packages/exec-core/src/types.ts | 28 +++ 14 files changed, 343 insertions(+), 131 deletions(-) diff --git a/packages/claude-sdk-tools/CHANGELOG.md b/packages/claude-sdk-tools/CHANGELOG.md index 97f7842d..155c69e9 100644 --- a/packages/claude-sdk-tools/CHANGELOG.md +++ b/packages/claude-sdk-tools/CHANGELOG.md @@ -96,6 +96,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - AzureDevOps.PullRequest.* tools accept an account field, matching AzCli/EscalatedAzCli - Binary files are blocked from text reads when the format is recognised; unrecognised formats are still treated as text - ExecV3 and Memory import defineTool, ToolCancelledError, ToolRefusedError, and pathSchema from their own claude-sdk subpaths instead of the barrel, so a consumer bundling this package no longer pulls in the whole SDK module graph +- ExecV3 pipelines now run over real OS pipes, so a cancelled or timed-out pipe returns instead of hanging the caller - Find tool follows symlinks with cycle detection - Fix version metadata - GitHub_PullRequest_AutoMerge takes a required strategy (merge, squash, rebase) when enabling, so it can queue a specific merge method instead of only accepting the repo default diff --git a/packages/claude-sdk-tools/changes.jsonl b/packages/claude-sdk-tools/changes.jsonl index b1d1b383..4dd58627 100644 --- a/packages/claude-sdk-tools/changes.jsonl +++ b/packages/claude-sdk-tools/changes.jsonl @@ -88,3 +88,4 @@ {"description":"An interactive az identity no longer gets a silent, unattended background relogin; the browser/MFA prompt only ever appears attached to a real caller's call","category":"fixed"} {"description":"The az session's own login and command env now strips the same ambient Azure credential vars ExecV3 strips, so the CLI's own environment can no longer steer a login it believes it fully controls","category":"security"} {"description":"NodeFileSystem implements the new IFileSystem members: the real OS temp directory, the process user id, a recursive create that honours an explicit mode, a symlink-preserving lstat, and a readlinkSync that answers null rather than throwing when there is nothing to follow","category":"added"} +{"description":"ExecV3 pipelines now run over real OS pipes, so a cancelled or timed-out pipe returns instead of hanging the caller","category":"fixed"} diff --git a/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts b/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts index 581d2c93..8427ede6 100644 --- a/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts +++ b/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts @@ -1,48 +1,48 @@ import { resolve } from 'node:path'; import { PassThrough, Readable, type Writable } from 'node:stream'; import type { IFileSystem } from '@shellicar/claude-core/fs/interfaces'; -import { fromStream, PipeConsumerGone } from '@shellicar/exec-core'; +import { fromStream, type PipelineStage } from '@shellicar/exec-core'; import type { EngineContext } from './engine'; import type { Command, CommandResult } from './types'; interface StageSinks { - stdout: Writable; - stderr: Writable; + stdout?: Writable; + stderr?: Writable; stdoutCapture?: PassThrough; stderrCapture?: PassThrough; } /** - * Resolve a single stage's sinks under V3's redirect model ({ stdout?, stderr? } with - * stderr "&1" = merge). `downstream` is the bridge a non-terminal stage feeds; when - * present and no stdout redirect diverts it, stdout flows onward and is NOT captured. + * Resolve one stage's sinks under V3's redirect model ({ stdout?, stderr? } with stderr "&1" + * = merge). A non-terminal stage has no stdout sink at all: its stdout is an OS pipe into the + * next stage, so those bytes never reach this process and are not captured. Merged stderr is + * left unset too, and follows stdout wherever it goes — into the pipe on a non-terminal stage. */ -function resolveStageSinks(cmd: Command, downstream: Writable | undefined, cwd: string, fs: IFileSystem): StageSinks { +function resolveStageSinks(cmd: Command, isLast: boolean, cwd: string, fs: IFileSystem): StageSinks { const redirect = cmd.redirect; const mergeStderr = redirect?.stderr === '&1'; - let stdout: Writable; + let stdout: Writable | undefined; let stdoutCapture: PassThrough | undefined; if (redirect?.stdout != null) { + // A non-terminal stage with a stdout redirect is rejected at validation (R4), so this + // is only reached on a terminal stage. const file = fs.createWriteStream(resolve(cwd, redirect.stdout), { flags: 'w' }); file.on('error', () => { // Redirect write errors should not crash the run. }); stdout = file; - // a terminal stage with a stdout redirect captures nothing; a non-terminal stage - // with op "|" + stdout redirect is rejected at validation (R4), so this branch is - // only reached on a terminal stage. - } else if (downstream != null) { - stdout = downstream; - } else { + } else if (isLast) { stdoutCapture = new PassThrough(); stdout = stdoutCapture; } - let stderr: Writable; + // Merged stderr gets no sink of its own — it follows stdout, which for a non-terminal + // stage means straight into the pipe. + let stderr: Writable | undefined; let stderrCapture: PassThrough | undefined; if (mergeStderr) { - stderr = stdout; + // no sink } else if (redirect?.stderr != null) { const file = fs.createWriteStream(resolve(cwd, redirect.stderr), { flags: 'w' }); file.on('error', () => { @@ -60,77 +60,37 @@ function resolveStageSinks(cmd: Command, downstream: Writable | undefined, cwd: /** Execute a pipeline (length ≥ 1), one CommandResult per stage. */ export async function runPipeline(commands: Command[], ctx: EngineContext): Promise { const n = commands.length; - const bridges = Array.from({ length: n - 1 }, () => { - const bridge = new PassThrough(); - // Teardown destroys a bridge while its producer may still be piping into it; that - // write-after-destroy emits 'error', and an unhandled stream 'error' would crash the - // process. Swallow it — the producer is being killed anyway. - bridge.on('error', () => {}); - return bridge; - }); + const sinks = commands.map((cmd, i) => resolveStageSinks(cmd, i === n - 1, cmd.cwd ?? ctx.cwd, ctx.fs)); - // Each stage gets its own teardown controller. When a stage settles, the upstream - // feeding it has nowhere left to send output, so we abort that upstream's controller, - // driving Executor.run's existing abort → group-kill path. The kill makes the upstream - // settle in turn, so teardown cascades one hop at a time all the way up the pipe. This - // is the SIGPIPE analogue: without it `find | head -1` hangs, the producer blocked on - // backpressure with no consumer. - const controllers = commands.map(() => new AbortController()); - const settled = new Array(n).fill(false); + const stages: PipelineStage[] = commands.map((cmd, i) => ({ + cmd: { program: cmd.program, args: cmd.args, cwd: cmd.cwd ?? ctx.cwd, env: ctx.envProvider.buildEnv(cmd.env) }, + stdout: sinks[i].stdout, + stderr: sinks[i].stderr, + mergeStderr: cmd.redirect?.stderr === '&1', + })); - const teardownUpstreamOf = (i: number): void => { - // An external cancel (timeout / ESC) already aborts every stage; that is not a - // consumer-exit teardown, so it must not tear an upstream down or double-abort. - if (ctx.signal?.aborted) { - return; - } - const up = i - 1; - // Nothing to tear down at the head; and never re-tear a stage that already exited on - // its own, which keeps its natural exit as-is instead of a teardown SIGPIPE. - if (up < 0 || settled[up]) { - return; - } - // Destroy the bridge feeding this stage. The consumer has gone, so nothing drains the - // bridge; the upstream is blocked on backpressure, and Executor.run's teardown awaits - // `finished()` on that bridge before it resolves. An orphaned bridge never emits the - // readable 'end' `finished()` waits for, so the killed producer would hang in its own - // teardown. Destroying it forces 'close', which settles `finished()`, and unblocks the - // producer's write so the group-kill below can take it down. - bridges[up].destroy(); - // Abort with the PipeConsumerGone reason: Executor.run maps it to a real SIGPIPE - // kill, so the producer dies from signal 13 and closes with `signal: 'SIGPIPE'` — - // the honest broken-pipe death, not a SIGTERM we later relabel. - controllers[up].abort(PipeConsumerGone); - }; + // Only the head can take a caller-supplied stdin; validation rejects it on a pipe target (NE2). + const stdin = commands[0].stdin != null ? Readable.from(commands[0].stdin) : undefined; - const runs: Promise[] = commands.map((cmd, i) => { - const isLast = i === n - 1; - const downstream = isLast ? undefined : bridges[i]; - const stageCwd = cmd.cwd ?? ctx.cwd; - const { stdout, stderr, stdoutCapture, stderrCapture } = resolveStageSinks(cmd, downstream, stageCwd, ctx.fs); - const stdin: Readable | undefined = i === 0 ? (cmd.stdin != null ? Readable.from(cmd.stdin) : undefined) : bridges[i - 1]; - // Combine the external cancel with this stage's own teardown controller. Either one - // aborting kills the stage; Executor.run honours a single signal, so merge them. - const signal = ctx.signal ? AbortSignal.any([ctx.signal, controllers[i].signal]) : controllers[i].signal; + const runs = ctx.executor.runPipeline(stages, { stdin, signal: ctx.signal }); + // Every stage starts together, so each start is read before any of them can settle. That is + // what makes a pipe's durations overlap rather than sum. + const startedAt = runs.map(() => ctx.now()); - const startedAt = ctx.now(); - return Promise.all([ctx.executor.run({ program: cmd.program, args: cmd.args, cwd: stageCwd, env: ctx.envProvider.buildEnv(cmd.env) }, { stdin, stdout, stderr, signal }), stdoutCapture ? fromStream(stdoutCapture) : Promise.resolve(''), stderrCapture ? fromStream(stderrCapture) : Promise.resolve('')]).then( - ([status, out, err]): CommandResult => { - settled[i] = true; - teardownUpstreamOf(i); // this stage is the consumer that just settled → stop its producer - // A producer torn down because its consumer left really died from SIGPIPE, so it - // closes with `signal: 'SIGPIPE'`. Report the stage's real exit as-is — the kill is - // honest, so no relabelling is needed. + return Promise.all( + runs.map((run, i) => { + const { stdoutCapture, stderrCapture } = sinks[i]; + return Promise.all([run, stdoutCapture ? fromStream(stdoutCapture) : Promise.resolve(''), stderrCapture ? fromStream(stderrCapture) : Promise.resolve('')]).then(([status, out, err]): CommandResult => { + // A producer whose consumer exited dies from a kernel SIGPIPE, so its real exit is + // already the honest broken-pipe death. Report it as-is. return { stdout: out, stderr: err, exitCode: status.exitCode, signal: status.signal, - durationMs: Math.round(ctx.now() - startedAt), + durationMs: Math.round(ctx.now() - startedAt[i]), }; - }, - ); - }); - - return Promise.all(runs); + }); + }), + ); } diff --git a/packages/claude-sdk-tools/test/Exec.spec.ts b/packages/claude-sdk-tools/test/Exec.spec.ts index 1e17f4f1..ecf2b2d1 100644 --- a/packages/claude-sdk-tools/test/Exec.spec.ts +++ b/packages/claude-sdk-tools/test/Exec.spec.ts @@ -1,5 +1,5 @@ import { ToolCancelledError } from '@shellicar/claude-sdk'; -import type { CommandSpec, ExitStatus, IExecutor, SpawnOpts } from '@shellicar/exec-core'; +import type { CommandSpec, ExitStatus, IExecutor, PipelineStage, SpawnOpts } from '@shellicar/exec-core'; import { describe, expect, it } from 'vitest'; import type { z } from 'zod'; import { createExec } from '../src/Exec/Exec'; @@ -73,7 +73,10 @@ function createInterleavingExecutor(expected: number): StubExecutor { return { exitCode: 0, signal: null }; }; - return { executor: { run }, maxInFlight: () => peak }; + // Exec (V1) only ever calls run(); one call per stage is enough for this double. + const runPipeline = (stages: PipelineStage[]): Promise[] => stages.map((stage) => run(stage.cmd, { stdout: stage.stdout, stderr: stage.stderr })); + + return { executor: { run, runPipeline }, maxInFlight: () => peak }; } describe('Exec — basic execution', () => { diff --git a/packages/claude-sdk-tools/test/ExecV3.spec.ts b/packages/claude-sdk-tools/test/ExecV3.spec.ts index a3728716..f16c50f1 100644 --- a/packages/claude-sdk-tools/test/ExecV3.spec.ts +++ b/packages/claude-sdk-tools/test/ExecV3.spec.ts @@ -1,5 +1,5 @@ import { ToolRefusedError } from '@shellicar/claude-sdk'; -import type { CommandSpec, ExitStatus, IExecutor, SpawnOpts } from '@shellicar/exec-core'; +import type { CommandSpec, ExitStatus, IExecutor, PipelineStage, SpawnOpts } from '@shellicar/exec-core'; import { describe, expect, it } from 'vitest'; import { StaticRulesConfigProvider } from '../src/Exec/IRulesConfigProvider'; import { createExecV3 } from '../src/ExecV3/ExecV3'; @@ -12,6 +12,7 @@ const echoExecutor: IExecutor = { opts?.stderr?.end(); return { exitCode: 0, signal: null }; }, + runPipeline: (stages: PipelineStage[]): Promise[] => stages.map((stage) => echoExecutor.run(stage.cmd, { stdout: stage.stdout, stderr: stage.stderr })), }; describe('ExecV3 — configured blocklist', () => { diff --git a/packages/claude-sdk-tools/test/ExecV3/duration.spec.ts b/packages/claude-sdk-tools/test/ExecV3/duration.spec.ts index fb43d765..5fa1375e 100644 --- a/packages/claude-sdk-tools/test/ExecV3/duration.spec.ts +++ b/packages/claude-sdk-tools/test/ExecV3/duration.spec.ts @@ -1,4 +1,4 @@ -import type { CommandSpec, ExitStatus, IExecutor, SpawnOpts } from '@shellicar/exec-core'; +import type { CommandSpec, ExitStatus, IExecutor, PipelineStage, SpawnOpts } from '@shellicar/exec-core'; import { describe, expect, it } from 'vitest'; import { StaticRulesConfigProvider } from '../../src/Exec/IRulesConfigProvider'; import { createExecV3 } from '../../src/ExecV3/ExecV3'; @@ -11,6 +11,12 @@ const echoExecutor: IExecutor = { opts?.stderr?.end(); return { exitCode: 0, signal: null }; }, + runPipeline: (stages: PipelineStage[]): Promise[] => + stages.map((stage) => { + stage.stdout?.end(stage.cmd.program); + stage.stderr?.end(); + return Promise.resolve({ exitCode: 0, signal: null }); + }), }; // The clock is injected (EngineContext.now / createExecV3's `now` param), so durationMs is @@ -33,9 +39,9 @@ describe('ExecV3 — durationMs uses the injected clock', () => { // No real spawn and no real elapsed time: durationMs is computed entirely from the injected // clock, so a pipe's "overlap, not addition" arithmetic can be proven with a fixed clock // sequence and manually-resolved ("spy") promises standing in for the two stages — nothing -// needs to actually take any wall-clock time. Both stages "start" on the same tick (runPipeline -// calls ctx.now() for each stage synchronously, back to back, before either's run() resolves), -// so which one settles first does not affect the assertion. +// needs to actually take any wall-clock time. Both stages "start" on the same tick (the stages +// are spawned as one unit, and ctx.now() is read for each of them back to back before any can +// settle), so which one settles first does not affect the assertion. describe('ExecV3 — pipe durationMs reflects overlap, not addition', () => { it('top-level durationMs is less than the sum of the per-stage durationMs', async () => { // top-start, stage0-start, stage1-start, first-stage-end, second-stage-end, top-end @@ -57,6 +63,12 @@ describe('ExecV3 — pipe durationMs reflects overlap, not addition', () => { opts?.stderr?.end(); return cmd.program === 'producer' ? producerDone : consumerDone; }, + runPipeline: (stages: PipelineStage[]): Promise[] => + stages.map((stage) => { + stage.stdout?.end(); + stage.stderr?.end(); + return stage.cmd.program === 'producer' ? producerDone : consumerDone; + }), }; const tool = createExecV3(new MemoryFileSystem(), spyExecutor, passthroughEnvProvider, new StaticRulesConfigProvider(), now); diff --git a/packages/claude-sdk-tools/test/FakeExecutor.ts b/packages/claude-sdk-tools/test/FakeExecutor.ts index cc4e4a20..08978da2 100644 --- a/packages/claude-sdk-tools/test/FakeExecutor.ts +++ b/packages/claude-sdk-tools/test/FakeExecutor.ts @@ -1,4 +1,4 @@ -import type { CommandSpec, ExitStatus, IExecutor, SpawnOpts } from '@shellicar/exec-core'; +import type { CommandSpec, ExitStatus, IExecutor, PipelineOpts, PipelineStage, SpawnOpts } from '@shellicar/exec-core'; export type FakeResponse = { stdout?: string; @@ -46,6 +46,36 @@ export class FakeExecutor implements IExecutor { return { exitCode: 'exitCode' in response ? (response.exitCode ?? null) : 0, signal: response.signal ?? null }; } + + /** The real executor joins stages with OS pipes; with no processes to join, the fake carries + * each stage's stdout string forward as the next stage's stdin. */ + public runPipeline(stages: PipelineStage[], opts: PipelineOpts = {}): Promise[] { + let upstream = drain(opts.stdin); + + return stages.map((stage, i) => { + const isLast = i === stages.length - 1; + const ran = upstream.then((stdin) => { + this.calls.push(stage.cmd); + const response = this.respond(stage.cmd, stdin); + + // Only a terminal stage has a stdout sink: a non-terminal stage's stdout is the pipe, + // so it is carried to the next stage instead of being written anywhere. + if (isLast && response.stdout != null) { + stage.stdout?.write(response.stdout); + } + if (response.stderr != null) { + (stage.mergeStderr ? stage.stdout : stage.stderr)?.write(response.stderr); + } + stage.stdout?.end(); + stage.stderr?.end(); + + return { stdout: response.stdout ?? '', status: { exitCode: 'exitCode' in response ? (response.exitCode ?? null) : 0, signal: response.signal ?? null } }; + }); + + upstream = ran.then((outcome) => outcome.stdout); + return ran.then((outcome) => outcome.status); + }); + } } /** A FakeResponder covering the shell-ish invocations these test suites lean on: diff --git a/packages/claude-sdk-tools/test/integration/AzSessionCache.spec.ts b/packages/claude-sdk-tools/test/integration/AzSessionCache.spec.ts index 2175530c..228ff93c 100644 --- a/packages/claude-sdk-tools/test/integration/AzSessionCache.spec.ts +++ b/packages/claude-sdk-tools/test/integration/AzSessionCache.spec.ts @@ -1,7 +1,7 @@ import { rm } from 'node:fs/promises'; import type { PassThrough } from 'node:stream'; import { Clock, Instant, ZoneOffset } from '@js-joda/core'; -import type { CommandSpec, ExitStatus, IExecutor, SpawnOpts } from '@shellicar/exec-core'; +import type { CommandSpec, ExitStatus, IExecutor, PipelineStage, SpawnOpts } from '@shellicar/exec-core'; import { afterEach, describe, expect, it } from 'vitest'; import { AzSessionCache } from '../../src/Az/AzSessionCache'; import type { AzDeps } from '../../src/Az/runAz'; @@ -55,6 +55,11 @@ class ControllableExecutor implements IExecutor { }); } + // az never pipes; one call per stage is the whole of what this double needs to model. + public runPipeline(stages: PipelineStage[]): Promise[] { + return stages.map((stage) => this.run(stage.cmd, { stdout: stage.stdout, stderr: stage.stderr })); + } + public async resolve(index: number, stdout = ''): Promise { const call = this.calls[index]; const out = call?.opts?.stdout as PassThrough | undefined; diff --git a/packages/claude-sdk-tools/test/integration/pipeline-teardown.spec.ts b/packages/claude-sdk-tools/test/integration/pipeline-teardown.spec.ts index 080a4657..9fe3cd1b 100644 --- a/packages/claude-sdk-tools/test/integration/pipeline-teardown.spec.ts +++ b/packages/claude-sdk-tools/test/integration/pipeline-teardown.spec.ts @@ -149,6 +149,47 @@ describe('SIGPIPE death — yes | head -n 1', () => { }); }); +// --------------------------------------------------------------------------- +// external cancel of a live pipe — bash: sleep 5 | cat, then ESC +// --------------------------------------------------------------------------- +// +// A cancel (ESC, or the default 30s timeout) that lands while every stage is still alive is +// a different path from a consumer exiting early, and it went untested for months while it +// hung: the stages died, but the run promise never settled, so the CLI wedged with nothing +// left running. `sleep 5 | cat` holds both stages open and moves no data, so the cancel +// lands on a live pipe. The requirement is only that the call comes back at all. + +describe('external cancel — sleep 5 | cat cancelled mid-flight', () => { + const input = { + intent: 'hold a two-stage pipe open so a cancel lands while both stages are alive', + commands: [{ program: 'sleep', args: ['5'], op: '|' as const }, { program: 'cat' }], + }; + + it('settles rather than hanging after every stage is killed', async () => { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), 100); + const hung = Symbol('hung'); + const guard = new Promise((resolve) => { + setTimeout(() => resolve(hung), BOUND_MS); + }); + + try { + // Cancelling makes the handler throw ToolCancelledError; either settlement proves it came + // back, so only a hang can fail this. + const settled = ExecV3.handler(ExecV3InputSchema.parse(input), controller.signal).then( + () => 'settled' as const, + () => 'settled' as const, + ); + const outcome = await Promise.race([settled, guard]); + const expected = 'settled'; + const actual = outcome === hung ? 'hung' : outcome; + expect(actual).toBe(expected); + } finally { + clearTimeout(timer); + } + }); +}); + // --------------------------------------------------------------------------- // middle-consumer exit — bash: find ~ -type f | head -n 1 | sleep 500 // --------------------------------------------------------------------------- diff --git a/packages/exec-core/CHANGELOG.md b/packages/exec-core/CHANGELOG.md index ed731088..1b55f671 100644 --- a/packages/exec-core/CHANGELOG.md +++ b/packages/exec-core/CHANGELOG.md @@ -12,9 +12,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Add a README describing the package and pointing to the main documentation - Allow killing a process with a chosen signal - Merge a child's stderr into its stdout by routing both to the same stream +- Run a pipeline as one unit, joining stages with real OS pipes so the kernel provides backpressure and delivers SIGPIPE when a consumer exits - Spawned children are detached from the operator's tty and the process group is killed on abort (SIGTERM, then SIGKILL after a grace period) - Stream-based interface for spawning a single process, with stdin, stdout, and stderr wired as streams +### Removed + +- Removed the PipeConsumerGone abort reason, which existed only to simulate a broken pipe in userland + ### Fixed +- A cancelled multi-stage pipeline now returns instead of hanging after its processes are killed - Fix version metadata diff --git a/packages/exec-core/changes.jsonl b/packages/exec-core/changes.jsonl index c6195978..5f78e237 100644 --- a/packages/exec-core/changes.jsonl +++ b/packages/exec-core/changes.jsonl @@ -4,3 +4,6 @@ {"description":"Add a README describing the package and pointing to the main documentation","category":"added"} {"description":"Allow killing a process with a chosen signal","category":"added"} {"description":"Fix version metadata","category":"fixed"} +{"description":"Run a pipeline as one unit, joining stages with real OS pipes so the kernel provides backpressure and delivers SIGPIPE when a consumer exits","category":"added"} +{"description":"A cancelled multi-stage pipeline now returns instead of hanging after its processes are killed","category":"fixed"} +{"description":"Removed the PipeConsumerGone abort reason, which existed only to simulate a broken pipe in userland","category":"removed"} diff --git a/packages/exec-core/src/Executor.ts b/packages/exec-core/src/Executor.ts index 1deaa121..48394140 100644 --- a/packages/exec-core/src/Executor.ts +++ b/packages/exec-core/src/Executor.ts @@ -1,39 +1,25 @@ -import { spawn } from 'node:child_process'; +import { type ChildProcess, spawn } from 'node:child_process'; import { existsSync } from 'node:fs'; import type { Writable } from 'node:stream'; import { finished } from 'node:stream/promises'; -import { PipeConsumerGone } from './reasons.js'; -import type { CommandSpec, ExitStatus, IExecutor, SpawnOpts } from './types.js'; - -// The kill signal for a teardown depends on why it fired: a producer whose pipe -// consumer has gone dies from SIGPIPE (so it closes with `signal: 'SIGPIPE'`, the honest -// broken-pipe death); every other abort (cancel, timeout) uses SIGTERM. The orchestrator states the reason -// on the abort; the mapping lives here because exec-core owns the kill. -function killSignal(reason: unknown): NodeJS.Signals { - return reason === PipeConsumerGone ? 'SIGPIPE' : 'SIGTERM'; -} - -// The distinct output sinks of a run — stdout and stderr may be the same Writable -// (merge), so de-dupe before acting on them. -function distinctSinks(opts: SpawnOpts): Writable[] { - const seen = new Set(); - for (const sink of [opts.stdout, opts.stderr]) { - if (sink) { - seen.add(sink); - } - } - return [...seen]; -} +import type { CommandSpec, ExitStatus, IExecutor, PipelineOpts, PipelineStage, SpawnOpts } from './types.js'; // End each distinct output sink and wait for it to finish flushing. Ending and // waiting are one operation: resolving only once every sink has finished is the // ordering contract a caller reading a redirect file depends on, so a caller must -// never be able to end a sink without then waiting for it. The promise form of -// `finished` resolves on finish and rejects on error; swallow the rejection so a -// broken sink cannot hang or fail the await. -async function closeSinks(opts: SpawnOpts): Promise { +// never be able to end a sink without then waiting for it. Sinks are de-duped because +// stdout and stderr may be the same Writable (merge). The promise form of `finished` +// resolves on finish and rejects on error; swallow the rejection so a broken sink +// cannot hang or fail the await. +async function closeSinks(sinks: (Writable | undefined)[]): Promise { + const distinct = new Set(); + for (const sink of sinks) { + if (sink) { + distinct.add(sink); + } + } await Promise.all( - distinctSinks(opts).map((sink) => { + [...distinct].map((sink) => { sink.end(); return finished(sink).catch(() => {}); }), @@ -61,15 +47,15 @@ export class Executor implements IExecutor { // An already-aborted signal never fires 'abort', so the listener below would // not catch it. Without this guard a chained command that inherits the // aborted signal still spawns — defeating ESC-cancel. Return the same killed - // status the group-kill path produces (the reason-mapped signal, no exit code). + // status the group-kill path produces (no exit code). if (opts.signal?.aborted) { - await closeSinks(opts); - return { exitCode: null, signal: killSignal(opts.signal.reason) }; + await closeSinks([opts.stdout, opts.stderr]); + return { exitCode: null, signal: 'SIGTERM' }; } if (!existsSync(cmd.cwd)) { opts.stderr?.write(`Working directory not found: ${cmd.cwd}`); - await closeSinks(opts); + await closeSinks([opts.stdout, opts.stderr]); return { exitCode: 126, signal: null }; } @@ -111,7 +97,7 @@ export class Executor implements IExecutor { const onAbort = () => { if (child.pid != null) { - this.#groupKill(child.pid, killSignal(opts.signal?.reason)); + this.#groupKill(child.pid); } }; opts.signal?.addEventListener('abort', onAbort, { once: true }); @@ -127,7 +113,7 @@ export class Executor implements IExecutor { this.#pids.delete(child.pid); } opts.signal?.removeEventListener('abort', onAbort); - await closeSinks(opts); + await closeSinks([opts.stdout, opts.stderr]); resolve(status); }; @@ -143,15 +129,151 @@ export class Executor implements IExecutor { }); } - #groupKill(pid: number, signal: NodeJS.Signals = 'SIGTERM'): void { + /** + * Stages are joined by real OS pipes: stage i is spawned with stage i+1's stdin fd as its + * own stdout, so the two children are connected in the kernel and this process never sees + * the bytes. That is what makes the pipe behave: the kernel applies backpressure, and it + * delivers SIGPIPE to a producer the instant its consumer exits. Nothing here has to + * notice a consumer leaving, because nothing here is in the middle. + */ + public runPipeline(stages: PipelineStage[], opts: PipelineOpts = {}): Promise[] { + const n = stages.length; + if (n === 1) { + const only = stages[0]; + return [this.run(only.cmd, { stdin: opts.stdin, stdout: only.stdout, stderr: only.mergeStderr ? only.stdout : only.stderr, signal: opts.signal })]; + } + + const settle = new Array<(status: ExitStatus) => void>(n); + const settled = new Array(n).fill(false); + const results = stages.map( + (_, i) => + new Promise((resolve) => { + settle[i] = resolve; + }), + ); + + // A stage's own sinks are closed when its process closes. Every sink here is either a + // capture the caller drains or a file, so this can never wait on a reader that has gone. + const finish = async (i: number, status: ExitStatus): Promise => { + if (settled[i]) { + return; + } + settled[i] = true; + await closeSinks([stages[i].stdout, stages[i].stderr]); + settle[i](status); + }; + + if (opts.signal?.aborted) { + for (let i = 0; i < n; i++) { + void finish(i, { exitCode: null, signal: 'SIGTERM' }); + } + return results; + } + + const children = new Array(n); + + // Spawn from the tail, because a stage's stdout IS its consumer's stdin fd and that fd + // only exists once the consumer has been spawned. + for (let i = n - 1; i >= 0; i--) { + const stage = stages[i]; + if (!existsSync(stage.cmd.cwd)) { + (stage.mergeStderr ? stage.stdout : stage.stderr)?.write(`Working directory not found: ${stage.cmd.cwd}`); + void finish(i, { exitCode: 126, signal: null }); + continue; + } + // A terminal stage's stdout comes back to the parent to be captured or redirected. A + // non-terminal stage writes into its consumer, or to nowhere if that consumer never started. + const stdout: 'pipe' | 'ignore' | Writable = i === n - 1 ? 'pipe' : (children[i + 1]?.stdin ?? 'ignore'); + const child = spawn(stage.cmd.program, stage.cmd.args ?? [], { + cwd: stage.cmd.cwd, + env: stage.cmd.env, + detached: true, + stdio: ['pipe', stdout, stage.mergeStderr ? stdout : 'pipe'], + }); + children[i] = child; + if (child.pid != null) { + this.#pids.add(child.pid); + } + } + + // Drop this process's copy of each write end now that the producer holds its own. While + // the parent still holds one, the pipe has a writer here and the consumer never sees EOF. + for (let i = 1; i < n; i++) { + children[i]?.stdin?.destroy(); + } + + const head = children[0]; + if (head?.stdin) { + if (opts.stdin) { + opts.stdin.pipe(head.stdin); + head.stdin.on('error', () => { + // Expected when the child exits before the input finishes writing. + }); + } else { + head.stdin.end(); + } + } + + for (let i = 0; i < n; i++) { + const child = children[i]; + if (!child) { + continue; + } + const stage = stages[i]; + const isLast = i === n - 1; + + // Only a terminal stage has a parent-side stdout; a non-terminal one wrote straight + // into the next child. A merged non-terminal stage has no parent-side stderr either. + if (child.stdout) { + const sink = isLast ? stage.stdout : undefined; + if (sink) { + child.stdout.pipe(sink, { end: false }); + } else { + child.stdout.resume(); + } + } + if (child.stderr) { + const sink = stage.mergeStderr ? stage.stdout : stage.stderr; + if (sink) { + child.stderr.pipe(sink, { end: false }); + } else { + child.stderr.resume(); + } + } + + child.on('close', (code, sig) => { + if (child.pid != null) { + this.#pids.delete(child.pid); + } + void finish(i, { exitCode: code, signal: sig ?? null }); + }); + + child.on('error', (err: NodeJS.ErrnoException) => { + (stage.mergeStderr ? stage.stdout : stage.stderr)?.write(err.code === 'ENOENT' ? `Command not found: ${stage.cmd.program}` : err.message); + void finish(i, { exitCode: err.code === 'ENOENT' ? 127 : 1, signal: null }); + }); + } + + const onAbort = () => { + for (const child of children) { + if (child?.pid != null) { + this.#groupKill(child.pid); + } + } + }; + opts.signal?.addEventListener('abort', onAbort, { once: true }); + void Promise.all(results).then(() => opts.signal?.removeEventListener('abort', onAbort)); + + return results; + } + + #groupKill(pid: number): void { try { - process.kill(-pid, signal); + process.kill(-pid, 'SIGTERM'); } catch { return; } - // If the process ignores the signal (a producer that handles SIGPIPE), the SIGKILL - // below reaps it after the grace period, and it then reports SIGKILL, not SIGPIPE. - // That is honest: a program that chose to handle the broken pipe did not die of it. + // A process that ignores SIGTERM is reaped by the SIGKILL below after the grace period. setTimeout(() => { try { process.kill(-pid, 'SIGKILL'); diff --git a/packages/exec-core/src/entry/index.ts b/packages/exec-core/src/entry/index.ts index 2ad903f3..84e3ce6b 100644 --- a/packages/exec-core/src/entry/index.ts +++ b/packages/exec-core/src/entry/index.ts @@ -1,7 +1,6 @@ import { Executor } from '../Executor.js'; import { fromStream } from '../fromStream.js'; -import { PipeConsumerGone } from '../reasons.js'; -import type { CommandSpec, ExitStatus, IExecutor, SpawnOpts } from '../types.js'; +import type { CommandSpec, ExitStatus, IExecutor, PipelineOpts, PipelineStage, SpawnOpts } from '../types.js'; -export type { CommandSpec, ExitStatus, IExecutor, SpawnOpts }; -export { Executor, fromStream, PipeConsumerGone }; +export type { CommandSpec, ExitStatus, IExecutor, PipelineOpts, PipelineStage, SpawnOpts }; +export { Executor, fromStream }; diff --git a/packages/exec-core/src/types.ts b/packages/exec-core/src/types.ts index d0d43590..f369f018 100644 --- a/packages/exec-core/src/types.ts +++ b/packages/exec-core/src/types.ts @@ -31,7 +31,35 @@ export interface SpawnOpts { signal?: AbortSignal; } +/** + * One stage of a pipeline. A non-terminal stage has no `stdout` sink: its stdout is the + * write end of a real pipe into the next stage, so the parent never sees those bytes. + */ +export interface PipelineStage { + cmd: CommandSpec; + /** Destination for this stage's stdout. Only a terminal stage has one. */ + stdout?: Writable; + /** Destination for this stage's stderr. Absent → drained, or merged when `mergeStderr`. */ + stderr?: Writable; + /** 2>&1 — stderr goes wherever stdout goes, including into the pipe. */ + mergeStderr?: boolean; +} + +/** Options for the pipeline as a whole. Individual destinations belong to each stage. */ +export interface PipelineOpts { + /** Source piped into the first stage's stdin. Absent → stdin is closed immediately. */ + stdin?: Readable; + /** When aborted, every stage's process group is killed. */ + signal?: AbortSignal; +} + /** The contract the tool layer depends on. Executor is one implementation. */ export interface IExecutor { run(cmd: CommandSpec, opts?: SpawnOpts): Promise; + /** + * Run stages as one pipeline over real OS pipes, stdout[i] feeding stdin[i+1]. Returns one + * promise per stage, in stage order, each settling when that stage's process closes. + * Returning the promises rather than awaiting them lets the caller time each stage itself. + */ + runPipeline(stages: PipelineStage[], opts?: PipelineOpts): Promise[]; } From 3e7ef81ab8633d22b5b7aabdbd47cc0e0b32999b Mon Sep 17 00:00:00 2001 From: Stephen Hellicar Date: Mon, 3 Aug 2026 02:11:59 +1000 Subject: [PATCH 02/11] Report why a pipeline stage could not start, and stop its producer A stage that fails to start has no child and no second file descriptor, so 2>&1 had nothing to apply to and the reason was discarded. Its consumer never opened the pipe either, leaving the producer writing into nothing until the run was cancelled. --- .../src/ExecV3/runPipeline.ts | 12 ++- .../integration/executorConformance.spec.ts | 68 +++++++++++++++- .../integration/pipeline-teardown.spec.ts | 32 ++++++++ packages/exec-core/CHANGELOG.md | 4 + packages/exec-core/changes.jsonl | 1 + packages/exec-core/src/Executor.ts | 32 +++++--- packages/exec-core/src/reasons.ts | 8 -- packages/exec-core/src/types.ts | 8 +- .../test/integration/Executor.spec.ts | 77 +++++++++++++++++++ 9 files changed, 214 insertions(+), 28 deletions(-) delete mode 100644 packages/exec-core/src/reasons.ts diff --git a/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts b/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts index 8427ede6..ee4d79f3 100644 --- a/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts +++ b/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts @@ -15,8 +15,7 @@ interface StageSinks { /** * Resolve one stage's sinks under V3's redirect model ({ stdout?, stderr? } with stderr "&1" * = merge). A non-terminal stage has no stdout sink at all: its stdout is an OS pipe into the - * next stage, so those bytes never reach this process and are not captured. Merged stderr is - * left unset too, and follows stdout wherever it goes — into the pipe on a non-terminal stage. + * next stage, so those bytes never reach this process and are not captured. */ function resolveStageSinks(cmd: Command, isLast: boolean, cwd: string, fs: IFileSystem): StageSinks { const redirect = cmd.redirect; @@ -37,13 +36,12 @@ function resolveStageSinks(cmd: Command, isLast: boolean, cwd: string, fs: IFile stdout = stdoutCapture; } - // Merged stderr gets no sink of its own — it follows stdout, which for a non-terminal - // stage means straight into the pipe. + // A merged stage still gets a capture. Its child's stderr goes to stdout, but a stage that + // never starts has no child and no fd 2 to merge, and the executor's account of why still + // has to reach the caller on the stage that failed. let stderr: Writable | undefined; let stderrCapture: PassThrough | undefined; - if (mergeStderr) { - // no sink - } else if (redirect?.stderr != null) { + if (!mergeStderr && redirect?.stderr != null) { const file = fs.createWriteStream(resolve(cwd, redirect.stderr), { flags: 'w' }); file.on('error', () => { // Redirect write errors should not crash the run. diff --git a/packages/claude-sdk-tools/test/integration/executorConformance.spec.ts b/packages/claude-sdk-tools/test/integration/executorConformance.spec.ts index 0db0c856..21e310ad 100644 --- a/packages/claude-sdk-tools/test/integration/executorConformance.spec.ts +++ b/packages/claude-sdk-tools/test/integration/executorConformance.spec.ts @@ -1,5 +1,5 @@ import { PassThrough, Readable } from 'node:stream'; -import { Executor, type IExecutor } from '@shellicar/exec-core'; +import { Executor, type IExecutor, type PipelineStage } from '@shellicar/exec-core'; import { describe, expect, it } from 'vitest'; import { FakeExecutor, shellLikeResponder } from '../FakeExecutor'; @@ -89,3 +89,69 @@ describe.each(executors)('%s', (_name, executor) => { }); } }); + +// The same pinning for pipelines. FakeExecutor cannot join processes, so it carries each +// stage's output forward as a string instead; these cases are what hold that stand-in to what +// the real Executor does, and without them every fake-backed pipe test rests on nothing. + +type PipeCase = { + name: string; + stages: { program: string; args?: string[] }[]; + expect: { + terminalStdout?: string; + /** Used instead of `terminalStdout` when exact whitespace isn't part of the contract. */ + terminalStdoutTrimmed?: string; + exitCodes: (number | null)[]; + }; +}; + +const pipeCases: PipeCase[] = [ + { name: 'carries stdout into the next stage', stages: [{ program: 'echo', args: ['a', 'b'] }, { program: 'cat' }], expect: { terminalStdout: 'a b\n', exitCodes: [0, 0] } }, + { + name: 'carries stdout through three stages', + stages: [ + { program: 'printf', args: ['a\nb\nc\n'] }, + { program: 'grep', args: ['b'] }, + { program: 'wc', args: ['-l'] }, + ], + expect: { terminalStdoutTrimmed: '1', exitCodes: [0, 0, 0] }, + }, + { name: 'reports each stage its own exit code', stages: [{ program: 'false' }, { program: 'cat' }], expect: { terminalStdout: '', exitCodes: [1, 0] } }, +]; + +async function runPipe(executor: IExecutor, c: PipeCase): Promise<{ terminalStdout: string; exitCodes: (number | null)[] }> { + const terminal = new PassThrough(); + let terminalStdout = ''; + terminal.on('data', (chunk) => { + terminalStdout += chunk.toString(); + }); + + const stages: PipelineStage[] = c.stages.map((stage, i) => ({ + cmd: { program: stage.program, args: stage.args ?? [], cwd: process.cwd(), env: process.env }, + stdout: i === c.stages.length - 1 ? terminal : undefined, + stderr: new PassThrough().resume(), + })); + + const statuses = await Promise.all(executor.runPipeline(stages)); + return { terminalStdout, exitCodes: statuses.map((s) => s.exitCode) }; +} + +describe.each(executors)('%s pipelines', (_name, executor) => { + for (const c of pipeCases) { + it(c.name, async () => { + const result = await runPipe(executor, c); + + if (c.expect.terminalStdout != null) { + expect(result.terminalStdout).toBe(c.expect.terminalStdout); + } + if (c.expect.terminalStdoutTrimmed != null) { + expect(result.terminalStdout.trim()).toBe(c.expect.terminalStdoutTrimmed); + } + }); + + it(`${c.name} — exit codes`, async () => { + const result = await runPipe(executor, c); + expect(result.exitCodes).toEqual(c.expect.exitCodes); + }); + } +}); diff --git a/packages/claude-sdk-tools/test/integration/pipeline-teardown.spec.ts b/packages/claude-sdk-tools/test/integration/pipeline-teardown.spec.ts index 9fe3cd1b..040e76e1 100644 --- a/packages/claude-sdk-tools/test/integration/pipeline-teardown.spec.ts +++ b/packages/claude-sdk-tools/test/integration/pipeline-teardown.spec.ts @@ -149,6 +149,38 @@ describe('SIGPIPE death — yes | head -n 1', () => { }); }); +// --------------------------------------------------------------------------- +// consumer that never starts — bash: yes | +// --------------------------------------------------------------------------- +// +// The consumer fails before it can read anything, so the pipe has no reader at all. The +// producer must take the same broken-pipe death it gets from a consumer that started and +// exited, rather than writing into nothing until the run times out. + +describe('consumer that never starts — yes | a stage with a missing cwd', () => { + const input = { + intent: 'pipe an endless producer into a stage that cannot start', + commands: [ + { program: 'yes', op: '|' as const }, + { program: 'cat', cwd: '/nonexistent/xyzzy-teardown' }, + ], + }; + + it('the producer dies from SIGPIPE rather than running on', async () => { + const outcome = await runBounded(input); + const expected = 'SIGPIPE'; + const actual = outcome.timedOut ? null : outcome.output.results[0]?.signal; + expect(actual).toBe(expected); + }); + + it('the stage that could not start reports why', async () => { + const outcome = await runBounded(input); + const expected = true; + const actual = outcome.timedOut ? false : (outcome.output.results[1]?.stderr.includes('Working directory not found') ?? false); + expect(actual).toBe(expected); + }); +}); + // --------------------------------------------------------------------------- // external cancel of a live pipe — bash: sleep 5 | cat, then ESC // --------------------------------------------------------------------------- diff --git a/packages/exec-core/CHANGELOG.md b/packages/exec-core/CHANGELOG.md index 1b55f671..e37cf79d 100644 --- a/packages/exec-core/CHANGELOG.md +++ b/packages/exec-core/CHANGELOG.md @@ -16,6 +16,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Spawned children are detached from the operator's tty and the process group is killed on abort (SIGTERM, then SIGKILL after a grace period) - Stream-based interface for spawning a single process, with stdin, stdout, and stderr wired as streams +### Changed + +- IExecutor now requires a runPipeline method, so an existing implementation of the interface must add one + ### Removed - Removed the PipeConsumerGone abort reason, which existed only to simulate a broken pipe in userland diff --git a/packages/exec-core/changes.jsonl b/packages/exec-core/changes.jsonl index 5f78e237..169f7f02 100644 --- a/packages/exec-core/changes.jsonl +++ b/packages/exec-core/changes.jsonl @@ -7,3 +7,4 @@ {"description":"Run a pipeline as one unit, joining stages with real OS pipes so the kernel provides backpressure and delivers SIGPIPE when a consumer exits","category":"added"} {"description":"A cancelled multi-stage pipeline now returns instead of hanging after its processes are killed","category":"fixed"} {"description":"Removed the PipeConsumerGone abort reason, which existed only to simulate a broken pipe in userland","category":"removed"} +{"description":"IExecutor now requires a runPipeline method, so an existing implementation of the interface must add one","category":"changed"} diff --git a/packages/exec-core/src/Executor.ts b/packages/exec-core/src/Executor.ts index 48394140..02b6ba21 100644 --- a/packages/exec-core/src/Executor.ts +++ b/packages/exec-core/src/Executor.ts @@ -177,13 +177,15 @@ export class Executor implements IExecutor { for (let i = n - 1; i >= 0; i--) { const stage = stages[i]; if (!existsSync(stage.cmd.cwd)) { - (stage.mergeStderr ? stage.stdout : stage.stderr)?.write(`Working directory not found: ${stage.cmd.cwd}`); + stage.stderr?.write(`Working directory not found: ${stage.cmd.cwd}`); void finish(i, { exitCode: 126, signal: null }); continue; } // A terminal stage's stdout comes back to the parent to be captured or redirected. A - // non-terminal stage writes into its consumer, or to nowhere if that consumer never started. - const stdout: 'pipe' | 'ignore' | Writable = i === n - 1 ? 'pipe' : (children[i + 1]?.stdin ?? 'ignore'); + // non-terminal stage writes into its consumer's stdin. Where that consumer never started, + // the parent takes the read end itself only to close it below, so the producer meets a + // pipe with no reader instead of writing into nothing until the run is cancelled. + const stdout: 'pipe' | Writable = i === n - 1 ? 'pipe' : (children[i + 1]?.stdin ?? 'pipe'); const child = spawn(stage.cmd.program, stage.cmd.args ?? [], { cwd: stage.cmd.cwd, env: stage.cmd.env, @@ -222,14 +224,18 @@ export class Executor implements IExecutor { const stage = stages[i]; const isLast = i === n - 1; - // Only a terminal stage has a parent-side stdout; a non-terminal one wrote straight - // into the next child. A merged non-terminal stage has no parent-side stderr either. if (child.stdout) { - const sink = isLast ? stage.stdout : undefined; - if (sink) { - child.stdout.pipe(sink, { end: false }); + if (isLast) { + if (stage.stdout) { + child.stdout.pipe(stage.stdout, { end: false }); + } else { + child.stdout.resume(); + } } else { - child.stdout.resume(); + // A non-terminal stage only has a parent-side stdout when its consumer never + // started. Closing the sole read end gives it the broken pipe it would have got + // from a consumer that started and died. + child.stdout.destroy(); } } if (child.stderr) { @@ -245,11 +251,17 @@ export class Executor implements IExecutor { if (child.pid != null) { this.#pids.delete(child.pid); } + // Forget the child so a later abort cannot signal a process group this pid no longer + // names. Nothing else reads `children` after the wiring above. + children[i] = undefined; void finish(i, { exitCode: code, signal: sig ?? null }); }); child.on('error', (err: NodeJS.ErrnoException) => { - (stage.mergeStderr ? stage.stdout : stage.stderr)?.write(err.code === 'ENOENT' ? `Command not found: ${stage.cmd.program}` : err.message); + if (settled[i]) { + return; + } + stage.stderr?.write(err.code === 'ENOENT' ? `Command not found: ${stage.cmd.program}` : err.message); void finish(i, { exitCode: err.code === 'ENOENT' ? 127 : 1, signal: null }); }); } diff --git a/packages/exec-core/src/reasons.ts b/packages/exec-core/src/reasons.ts deleted file mode 100644 index d18dec86..00000000 --- a/packages/exec-core/src/reasons.ts +++ /dev/null @@ -1,8 +0,0 @@ -/** - * Abort reason marking a teardown caused by a pipe consumer exiting. The orchestrator - * aborts a producer's signal with this reason; Executor maps it to a SIGPIPE kill, so the - * producer dies from signal 13 and reports `signal: 'SIGPIPE'` honestly, rather than being - * killed with SIGTERM and relabelled. The reason vocabulary lives here so the tool imports it from - * exec-core, keeping the dependency pointing into the core rather than out of it. - */ -export const PipeConsumerGone = Symbol('PipeConsumerGone'); diff --git a/packages/exec-core/src/types.ts b/packages/exec-core/src/types.ts index f369f018..2f5b5b46 100644 --- a/packages/exec-core/src/types.ts +++ b/packages/exec-core/src/types.ts @@ -39,9 +39,13 @@ export interface PipelineStage { cmd: CommandSpec; /** Destination for this stage's stdout. Only a terminal stage has one. */ stdout?: Writable; - /** Destination for this stage's stderr. Absent → drained, or merged when `mergeStderr`. */ + /** + * Destination for this stage's stderr, and for the diagnostics the executor writes itself + * when a stage cannot be started. `mergeStderr` diverts the child's own stderr away from + * here; it never diverts the executor's, which describe a child that does not exist. + */ stderr?: Writable; - /** 2>&1 — stderr goes wherever stdout goes, including into the pipe. */ + /** 2>&1 — the child's stderr goes wherever its stdout goes, including into the pipe. */ mergeStderr?: boolean; } diff --git a/packages/exec-core/test/integration/Executor.spec.ts b/packages/exec-core/test/integration/Executor.spec.ts index 36096cdb..b68a9136 100644 --- a/packages/exec-core/test/integration/Executor.spec.ts +++ b/packages/exec-core/test/integration/Executor.spec.ts @@ -56,3 +56,80 @@ describe('Executor.run already-aborted signal', () => { expect(actual).toBe(expected); }); }); + +describe('Executor.runPipeline', () => { + const spec = (program: string, args: string[] = []) => ({ program, args, cwd: process.cwd(), env: process.env }); + + const collector = (): { sink: Writable; read: () => string } => { + let captured = ''; + const sink = new Writable({ + write(chunk, _encoding, callback) { + captured += chunk.toString(); + callback(); + }, + }); + return { sink, read: () => captured }; + }; + + it('carries a stage\u2019s stdout into the next stage', async () => { + using executor = new Executor(); + const terminal = collector(); + + await Promise.all(executor.runPipeline([{ cmd: spec('echo', ['piped']) }, { cmd: spec('cat'), stdout: terminal.sink }])); + + const expected = 'piped\n'; + const actual = terminal.read(); + expect(actual).toBe(expected); + }); + + // Same ordering contract as run(): a caller reading a redirect file the moment the promise + // resolves must not race the flush. + it('resolves a stage only after its output sink has finished', async () => { + using executor = new Executor(); + let finished = false; + const sink = new Writable({ + write(_chunk, _encoding, callback) { + callback(); + }, + final(callback) { + setTimeout(callback, 50); + }, + }); + sink.on('finish', () => { + finished = true; + }); + + const runs = executor.runPipeline([{ cmd: spec('echo', ['hi']) }, { cmd: spec('cat'), stdout: sink }]); + await runs[runs.length - 1]; + + const expected = true; + const actual = finished; + expect(actual).toBe(expected); + }); + + it('spawns nothing when the signal is already aborted', async () => { + using executor = new Executor(); + const controller = new AbortController(); + controller.abort(); + const terminal = collector(); + + await Promise.all(executor.runPipeline([{ cmd: spec('echo', ['hi']) }, { cmd: spec('cat'), stdout: terminal.sink }], { signal: controller.signal })); + + const expected = ''; + const actual = terminal.read(); + expect(actual).toBe(expected); + }); + + // The kernel, not this package, is what stops the producer: closing the read end is the + // whole mechanism, so this goes red if the fd handoff is ever routed back through the parent. + it('kills a producer with SIGPIPE when its consumer exits early', async () => { + using executor = new Executor(); + const terminal = collector(); + + const [producer] = await Promise.all(executor.runPipeline([{ cmd: spec('yes') }, { cmd: spec('head', ['-n', '1']), stdout: terminal.sink }])); + + const expected = 'SIGPIPE'; + const actual = producer.signal; + expect(actual).toBe(expected); + }); +}); From 54f3adfb66b284482721724bb54150f9eaa74e2e Mon Sep 17 00:00:00 2001 From: Stephen Hellicar Date: Mon, 3 Aug 2026 20:04:24 +1000 Subject: [PATCH 03/11] Stop the executor waiting on sinks it does not own A lone stage was delegated to run(), which takes one stderr sink where a stage has two destinations for it, so a merged command's capture was never ended and whoever drained it waited forever. Closing a sink then waiting for it has the same shape wherever the sink is a duplex: the reader on the other end is the caller, who is waiting for the executor. --- .../integration/executorConformance.spec.ts | 111 +++++++++++++++--- .../test/integration/redirect-file.spec.ts | 35 ++++++ packages/exec-core/CHANGELOG.md | 1 + packages/exec-core/changes.jsonl | 1 + packages/exec-core/src/Executor.ts | 31 +++-- .../test/integration/Executor.spec.ts | 83 ++++++++++++- 6 files changed, 231 insertions(+), 31 deletions(-) create mode 100644 packages/claude-sdk-tools/test/integration/redirect-file.spec.ts diff --git a/packages/claude-sdk-tools/test/integration/executorConformance.spec.ts b/packages/claude-sdk-tools/test/integration/executorConformance.spec.ts index 21e310ad..bae97ff8 100644 --- a/packages/claude-sdk-tools/test/integration/executorConformance.spec.ts +++ b/packages/claude-sdk-tools/test/integration/executorConformance.spec.ts @@ -1,5 +1,5 @@ import { PassThrough, Readable } from 'node:stream'; -import { Executor, type IExecutor, type PipelineStage } from '@shellicar/exec-core'; +import { Executor, fromStream, type IExecutor, type PipelineStage } from '@shellicar/exec-core'; import { describe, expect, it } from 'vitest'; import { FakeExecutor, shellLikeResponder } from '../FakeExecutor'; @@ -66,26 +66,59 @@ const executors: [string, IExecutor][] = [ ['Executor (real)', new Executor()], ]; +type Outcome = { stdout: string; stderr: string; exitCode: number | null }; + +function expectCase(result: Outcome, c: Case): void { + if (c.expect.stdout != null) { + expect(result.stdout).toBe(c.expect.stdout); + } + if (c.expect.stdoutTrimmed != null) { + expect(result.stdout.trim()).toBe(c.expect.stdoutTrimmed); + } + if (c.expect.stderr != null) { + expect(result.stderr).toBe(c.expect.stderr); + } + if (c.expect.stderrIncludes != null) { + expect(result.stderr).toContain(c.expect.stderrIncludes); + } + if (c.expect.exitCode !== undefined) { + expect(result.exitCode).toBe(c.expect.exitCode); + } +} + describe.each(executors)('%s', (_name, executor) => { for (const c of cases) { it(c.name, async () => { - const result = await run(executor, c); + expectCase(await run(executor, c), c); + }); + } +}); - if (c.expect.stdout != null) { - expect(result.stdout).toBe(c.expect.stdout); - } - if (c.expect.stdoutTrimmed != null) { - expect(result.stdout.trim()).toBe(c.expect.stdoutTrimmed); - } - if (c.expect.stderr != null) { - expect(result.stderr).toBe(c.expect.stderr); - } - if (c.expect.stderrIncludes != null) { - expect(result.stderr).toContain(c.expect.stderrIncludes); - } - if (c.expect.exitCode !== undefined) { - expect(result.exitCode).toBe(c.expect.exitCode); - } +// The same table again, driven as a one-stage pipeline. A single command is the commonest +// thing ExecV3 runs, and it reaches the executor through runPipeline, not run — so without +// these, every behaviour above is pinned on a route the tool does not take. + +async function runAsSingleStage(executor: IExecutor, c: Case): Promise { + const stdout = new PassThrough(); + const stderr = new PassThrough(); + let stdoutText = ''; + let stderrText = ''; + stdout.on('data', (chunk) => { + stdoutText += chunk.toString(); + }); + stderr.on('data', (chunk) => { + stderrText += chunk.toString(); + }); + + const [status] = await Promise.all(executor.runPipeline([{ cmd: { program: c.program, args: c.args ?? [], cwd: c.cwd ?? process.cwd(), env: process.env }, stdout, stderr }], { stdin: c.stdin != null ? Readable.from(c.stdin) : undefined })); + + return { stdout: stdoutText, stderr: stderrText, exitCode: status.exitCode }; +} + +describe.each(executors)('%s as a one-stage pipeline', (_name, executor) => { + for (const c of cases) { + it(c.name, async () => { + expectCase(await runAsSingleStage(executor, c), c); }); } }); @@ -155,3 +188,47 @@ describe.each(executors)('%s pipelines', (_name, executor) => { }); } }); + +// Merging is the axis the pipe cases above leave untested, and it is the one where the fake +// and the real executor can drift apart without anything noticing: the fake ends every sink a +// stage was given, so a real executor that leaves one open still looks correct from any +// fake-backed test. A caller drains these sinks to strings, so an unended one is not a lost +// message but a result that never arrives. + +const DRAIN_BOUND_MS = 2000; + +// Drains each stage's stderr to a string exactly as ExecV3 does, so "the sink was ended" and +// "the caller's read completes" are the same event. An unended sink leaves that read pending +// forever, so the bound is what turns a hang into a value the assertion can compare. +async function drainsEveryStderr(executor: IExecutor, stageCount: number): Promise { + const cmd = { program: 'echo', args: ['hi'], cwd: process.cwd(), env: process.env }; + const stages: PipelineStage[] = Array.from({ length: stageCount }, (_, i) => ({ + cmd: i === 0 ? cmd : { program: 'cat', args: [], cwd: cmd.cwd, env: cmd.env }, + stdout: i === stageCount - 1 ? new PassThrough().resume() : undefined, + stderr: new PassThrough(), + mergeStderr: true, + })); + + const runs = executor.runPipeline(stages); + const drains = stages.map((stage) => fromStream(stage.stderr as PassThrough)); + const done = Promise.all([...runs, ...drains]).then(() => true); + const bound = new Promise((resolve) => { + setTimeout(() => resolve(false), DRAIN_BOUND_MS); + }); + + return Promise.race([done, bound]); +} + +describe.each(executors)('%s merged stages', (_name, executor) => { + it('ends the stderr sink of a single merged stage', async () => { + const expected = true; + const actual = await drainsEveryStderr(executor, 1); + expect(actual).toBe(expected); + }); + + it('ends the stderr sink of every stage in a merged pipeline', async () => { + const expected = true; + const actual = await drainsEveryStderr(executor, 2); + expect(actual).toBe(expected); + }); +}); diff --git a/packages/claude-sdk-tools/test/integration/redirect-file.spec.ts b/packages/claude-sdk-tools/test/integration/redirect-file.spec.ts new file mode 100644 index 00000000..42da27e8 --- /dev/null +++ b/packages/claude-sdk-tools/test/integration/redirect-file.spec.ts @@ -0,0 +1,35 @@ +import { readFileSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, describe, expect, it } from 'vitest'; +import { ExecV3 } from '../../src/entry/ExecV3'; +import { call } from '../helpers'; + +// A stdout redirect is the one shape where the caller reads the result off disk instead of out +// of the tool's own output, so it rests entirely on the executor having flushed and closed the +// file before the call returns. A single command is also the arity with the least coverage +// here: every other redirect test in the repo runs against the fake, which writes no files. + +const target = join(tmpdir(), 'execv3-redirect-file.log'); + +describe('a single command redirecting stdout to a file', () => { + afterEach(() => { + rmSync(target, { force: true }); + }); + + it('has written the file by the time the call returns', async () => { + await call(ExecV3, { intent: 'write stdout to a file', commands: [{ program: 'echo', args: ['written'], redirect: { stdout: target } }] }); + + const expected = 'written\n'; + const actual = readFileSync(target, 'utf8'); + expect(actual).toBe(expected); + }); + + it('captures nothing itself, because the output went to the file', async () => { + const result = await call(ExecV3, { intent: 'write stdout to a file', commands: [{ program: 'echo', args: ['written'], redirect: { stdout: target } }] }); + + const expected = ''; + const actual = result.results[0]?.stdout; + expect(actual).toBe(expected); + }); +}); diff --git a/packages/exec-core/CHANGELOG.md b/packages/exec-core/CHANGELOG.md index e37cf79d..282c4173 100644 --- a/packages/exec-core/CHANGELOG.md +++ b/packages/exec-core/CHANGELOG.md @@ -27,4 +27,5 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - A cancelled multi-stage pipeline now returns instead of hanging after its processes are killed +- A command no longer hangs when an output sink it was given is drained slowly or not at all - Fix version metadata diff --git a/packages/exec-core/changes.jsonl b/packages/exec-core/changes.jsonl index 169f7f02..5ea86a6e 100644 --- a/packages/exec-core/changes.jsonl +++ b/packages/exec-core/changes.jsonl @@ -8,3 +8,4 @@ {"description":"A cancelled multi-stage pipeline now returns instead of hanging after its processes are killed","category":"fixed"} {"description":"Removed the PipeConsumerGone abort reason, which existed only to simulate a broken pipe in userland","category":"removed"} {"description":"IExecutor now requires a runPipeline method, so an existing implementation of the interface must add one","category":"changed"} +{"description":"A command no longer hangs when an output sink it was given is drained slowly or not at all","category":"fixed"} diff --git a/packages/exec-core/src/Executor.ts b/packages/exec-core/src/Executor.ts index 02b6ba21..e3a1c6d6 100644 --- a/packages/exec-core/src/Executor.ts +++ b/packages/exec-core/src/Executor.ts @@ -1,16 +1,22 @@ import { type ChildProcess, spawn } from 'node:child_process'; import { existsSync } from 'node:fs'; -import type { Writable } from 'node:stream'; +import { Duplex, type Writable } from 'node:stream'; import { finished } from 'node:stream/promises'; import type { CommandSpec, ExitStatus, IExecutor, PipelineOpts, PipelineStage, SpawnOpts } from './types.js'; -// End each distinct output sink and wait for it to finish flushing. Ending and -// waiting are one operation: resolving only once every sink has finished is the -// ordering contract a caller reading a redirect file depends on, so a caller must -// never be able to end a sink without then waiting for it. Sinks are de-duped because -// stdout and stderr may be the same Writable (merge). The promise form of `finished` -// resolves on finish and rejects on error; swallow the rejection so a broken sink -// cannot hang or fail the await. +// End each distinct output sink, and wait for the ones whose flush is ours to complete. +// +// A write-only sink is a destination this process owns the whole of: a file, where 'finish' +// means the bytes are out and a caller reading that file straight after is safe. Waiting is +// the ordering contract, so a caller must never be able to end such a sink without waiting. +// +// A duplex sink is not that. Something else reads its other end, and it cannot finish until +// that reader consumes it, so waiting here is waiting for the caller while the caller waits +// for us. Ending it is the whole of the obligation, and the caller's own read is the join. +// Which of the two a sink is is decided here, from the sink itself, so no caller has to know. +// +// Sinks are de-duped because stdout and stderr may be the same Writable (merge). `finished` +// rejects on a stream error; swallow it so a broken sink cannot fail the await. async function closeSinks(sinks: (Writable | undefined)[]): Promise { const distinct = new Set(); for (const sink of sinks) { @@ -21,7 +27,7 @@ async function closeSinks(sinks: (Writable | undefined)[]): Promise { await Promise.all( [...distinct].map((sink) => { sink.end(); - return finished(sink).catch(() => {}); + return sink instanceof Duplex ? undefined : finished(sink).catch(() => {}); }), ); } @@ -137,11 +143,10 @@ export class Executor implements IExecutor { * notice a consumer leaving, because nothing here is in the middle. */ public runPipeline(stages: PipelineStage[], opts: PipelineOpts = {}): Promise[] { + // Every arity goes down the same path, one stage included. Delegating a lone stage to run() + // looks like reuse, but run() takes one stderr sink where a stage has two destinations for + // it, so a merged stage's sink was left open and whoever drained it waited forever. const n = stages.length; - if (n === 1) { - const only = stages[0]; - return [this.run(only.cmd, { stdin: opts.stdin, stdout: only.stdout, stderr: only.mergeStderr ? only.stdout : only.stderr, signal: opts.signal })]; - } const settle = new Array<(status: ExitStatus) => void>(n); const settled = new Array(n).fill(false); diff --git a/packages/exec-core/test/integration/Executor.spec.ts b/packages/exec-core/test/integration/Executor.spec.ts index b68a9136..ee7d896a 100644 --- a/packages/exec-core/test/integration/Executor.spec.ts +++ b/packages/exec-core/test/integration/Executor.spec.ts @@ -1,6 +1,7 @@ -import { Writable } from 'node:stream'; +import { PassThrough, Writable } from 'node:stream'; import { describe, expect, it } from 'vitest'; import { Executor } from '../../src/Executor.js'; +import { fromStream } from '../../src/fromStream.js'; describe('Executor.run output-sink flush', () => { // Ordering contract: run must not resolve until its output sinks have finished @@ -82,6 +83,86 @@ describe('Executor.runPipeline', () => { expect(actual).toBe(expected); }); + // The sink-closing contract holds for every sink a stage was given, not only the ones its + // output happens to reach. A merged stage sends its child's stderr to stdout, so nothing is + // ever written to the stderr sink, and it must still be ended: a caller draining that sink + // to a string waits on an end that would otherwise never come, and never learns the stage + // finished at all. + it('ends a merged stage’s stderr sink, which its output never reaches', async () => { + using executor = new Executor(); + const terminal = collector(); + const unusedStderr = new PassThrough(); + + // Draining the sink is both how a caller uses it and how this test observes it: the read + // completes only if the sink was ended. The bound turns the failure into a value, since an + // unended sink leaves the read pending rather than rejecting. + const runs = executor.runPipeline([{ cmd: spec('echo', ['hi']), stdout: terminal.sink, stderr: unusedStderr, mergeStderr: true }]); + const drained = Promise.all([...runs, fromStream(unusedStderr)]).then(() => 'drained' as const); + const bound = new Promise<'pending'>((resolve) => { + setTimeout(() => resolve('pending'), 2000); + }); + + const expected = 'drained'; + const actual = await Promise.race([drained, bound]); + expect(actual).toBe(expected); + }); + + // Nothing the executor waits for should be something the caller has to do first. A capture + // is a duplex: the executor writes one end and the caller reads the other, so waiting for it + // to finish is waiting for the caller, who is waiting for the executor. A caller that drains + // late, slowly, or not at all is then a hang rather than a mistake with a consequence. + it('settles even when the caller never drains a sink it was given', async () => { + using executor = new Executor(); + const neverRead = new PassThrough(); + + const settled = Promise.all(executor.runPipeline([{ cmd: spec('echo', ['hi']), stdout: neverRead }])).then(() => 'settled' as const); + const bound = new Promise<'stalled'>((resolve) => { + setTimeout(() => resolve('stalled'), 2000); + }); + + const expected = 'settled'; + const actual = await Promise.race([settled, bound]); + expect(actual).toBe(expected); + }); + + // Both invariants below are ones run() already guarantees, pinned here for a single stage + // because that is the arity whose route through the executor is the least obvious and the + // most used: every plain command, and every link of an && chain, is a pipeline of one. + it('resolves a single stage only after its output sink has finished', async () => { + using executor = new Executor(); + let finished = false; + const sink = new Writable({ + write(_chunk, _encoding, callback) { + callback(); + }, + final(callback) { + setTimeout(callback, 50); + }, + }); + sink.on('finish', () => { + finished = true; + }); + + await Promise.all(executor.runPipeline([{ cmd: spec('echo', ['hi']), stdout: sink }])); + + const expected = true; + const actual = finished; + expect(actual).toBe(expected); + }); + + it('spawns nothing for a single stage when the signal is already aborted', async () => { + using executor = new Executor(); + const controller = new AbortController(); + controller.abort(); + const terminal = collector(); + + await Promise.all(executor.runPipeline([{ cmd: spec('echo', ['hi']), stdout: terminal.sink }], { signal: controller.signal })); + + const expected = ''; + const actual = terminal.read(); + expect(actual).toBe(expected); + }); + // Same ordering contract as run(): a caller reading a redirect file the moment the promise // resolves must not race the flush. it('resolves a stage only after its output sink has finished', async () => { From b6b69bdef7ad773d064b790754d051b8d5e69c62 Mon Sep 17 00:00:00 2001 From: Stephen Hellicar Date: Mon, 3 Aug 2026 21:05:16 +1000 Subject: [PATCH 04/11] Assert that a torn-down producer stopped, not how it died Node's stdio is a socketpair rather than a pipe, so a consumer that exits leaving unread bytes makes the kernel reset the connection: the producer's write fails with ECONNRESET and it exits non-zero, where a clean close would have raised SIGPIPE. Over 40 rounds that was SIGPIPE every time on macOS and about one in three on Linux, so a test naming the signal is asserting a coin toss. --- CLAUDE.md | 7 ++ .../integration/pipeline-diagnostics.spec.ts | 44 ++++++++++++ .../integration/pipeline-teardown.spec.ts | 68 +++++++++++-------- .../integration/single-stage-merge.spec.ts | 52 ++++++++++++++ packages/exec-core/CHANGELOG.md | 2 +- packages/exec-core/changes.jsonl | 2 +- .../test/integration/Executor.spec.ts | 15 ++-- 7 files changed, 153 insertions(+), 37 deletions(-) create mode 100644 packages/claude-sdk-tools/test/integration/pipeline-diagnostics.spec.ts create mode 100644 packages/claude-sdk-tools/test/integration/single-stage-merge.spec.ts diff --git a/CLAUDE.md b/CLAUDE.md index fec73702..0de7e14d 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -309,6 +309,13 @@ This is a local dev-loop gap only, not a pipeline defect: CI always builds `keyc 3. **Slash commands are string-matched** — no command registry 4. **Context thresholds hardcoded** — 85%/90% tool disable thresholds not configurable 5. **AppLayout combines View + Controller** — separation planned +6. **A broken pipe reports two different ways** — ExecV3 joins pipeline stages at the file descriptor, but Node's stdio is a *socketpair*, not a pipe. A consumer that exits leaving unread bytes makes the kernel reset the connection, so the producer's write fails with `ECONNRESET`, it exits non-zero, and it prints its own error (`yes: standard output: Connection reset by peer`). A clean close instead raises SIGPIPE and the producer dies silently. Which one happens is timing: measured over 40 rounds, SIGPIPE every time on macOS and about one in three on Linux. The hang fix is unaffected — the producer always stops — and the blast radius is only a consumer that exits early, never a full-drain pipe. + + The cost worth caring about is not the `signal` field, which nothing reads. It is the stderr line: a plausible-looking error the model may act on, for a command that did exactly what was asked. + + **Do not "fix" this by reaching for a shell.** Bash is deterministic here because it calls `pipe(2)`; ExecV3 has no shell on purpose, so this nondeterminism is a cost of that decision, not a defect to route around. Two real options if it ever matters: a FIFO per link (rejected — Node has no `mkfifo` either, so it means spawning a process per link or native code anyway, and it puts a predictable path in a shared tmpdir inside the one tool that runs arbitrary commands), or a native `pipe(2)` binding (the clean answer, following the `keychain-native` N-API pattern, but it turns `exec-core` from pure JS into a native package needing per-platform prebuilds). Revisit only if Linux becomes a platform we ship to rather than one we test on. + + Do not assert a specific signal for a torn-down producer in a test. `pipeline-teardown.spec.ts` asserts it stopped and did not succeed, which holds on both. ## Az Auth Hardening diff --git a/packages/claude-sdk-tools/test/integration/pipeline-diagnostics.spec.ts b/packages/claude-sdk-tools/test/integration/pipeline-diagnostics.spec.ts new file mode 100644 index 00000000..161d2be2 --- /dev/null +++ b/packages/claude-sdk-tools/test/integration/pipeline-diagnostics.spec.ts @@ -0,0 +1,44 @@ +import { describe, expect, it } from 'vitest'; +import type { z } from 'zod'; +import { ExecV3, ExecV3InputSchema } from '../../src/entry/ExecV3'; + +// The executor synthesises two diagnostics itself, rather than getting them from the child: +// "Command not found" (ENOENT) and "Working directory not found" (missing cwd). Both are +// written to the stage's stderr sink, or to its stdout sink when the stage merges with +// "&1". A non-terminal stage has no stdout sink — its stdout is the OS pipe — so a merged +// non-terminal stage has nowhere to put them and they are dropped: the stage reports 127 or +// 126 with no explanation anywhere in the result. Without the merge the same stage reports +// the message correctly, which is what makes this a hole rather than a design. +// +// The assertion is deliberately weak: the message must reach the caller somewhere, in this +// stage's own stderr or downstream in the pipe. Which of the two is the right home is a +// separate question this does not prejudge. + +const messageSomewhereIn = async (input: z.input): Promise => { + const { textContent } = await ExecV3.handler(ExecV3InputSchema.parse(input)); + return textContent.results.map((r) => (r == null ? '' : `${r.stdout}${r.stderr}`)).join(''); +}; + +describe('a merged non-terminal stage reports why it failed', () => { + it('surfaces "Command not found" when the program does not exist', async () => { + const haystack = await messageSomewhereIn({ + intent: 'pipe a missing program, merging its stderr, into cat', + commands: [{ program: 'definitely-not-a-real-command-xyzzy', redirect: { stderr: '&1' }, op: '|' }, { program: 'cat' }], + }); + + const expected = true; + const actual = haystack.includes('Command not found'); + expect(actual).toBe(expected); + }); + + it('surfaces "Working directory not found" when the cwd is missing', async () => { + const haystack = await messageSomewhereIn({ + intent: 'pipe a stage with a missing cwd, merging its stderr, into cat', + commands: [{ program: 'echo', args: ['hi'], cwd: '/nonexistent/path/xyzzy', redirect: { stderr: '&1' }, op: '|' }, { program: 'cat' }], + }); + + const expected = true; + const actual = haystack.includes('Working directory not found'); + expect(actual).toBe(expected); + }); +}); diff --git a/packages/claude-sdk-tools/test/integration/pipeline-teardown.spec.ts b/packages/claude-sdk-tools/test/integration/pipeline-teardown.spec.ts index 040e76e1..2ff795db 100644 --- a/packages/claude-sdk-tools/test/integration/pipeline-teardown.spec.ts +++ b/packages/claude-sdk-tools/test/integration/pipeline-teardown.spec.ts @@ -6,13 +6,12 @@ import type { Command } from '../../src/ExecV3/types'; import { ExecV3, ExecV3InputSchema, passthroughEnvProvider } from '../../src/entry/ExecV3'; import { nodeFs } from '../../src/fs/nodeFs'; -// Pipe-teardown tests — the hang and the SIGPIPE death of a torn-down producer. +// Pipe-teardown tests — the hang, and the death of a producer whose consumer has gone. // -// When a `|` consumer exits early (`find | head -1`), the producer is never told its -// reader has gone and blocks on backpressure forever. These tests hold the fixed -// behaviour from PR #380: the run returns promptly, teardown cascades all the way up a -// multi-stage pipe, and a torn-down producer dies from SIGPIPE (the real broken-pipe -// signal). They go red if the pipe lifecycle regresses. +// When a `|` consumer exits early (`find | head -1`), a producer that is never told its reader +// has gone blocks on backpressure forever. These tests hold the fixed behaviour: the run returns +// promptly, teardown reaches all the way up a multi-stage pipe, and every producer above the +// consumer is stopped. They go red if the pipe lifecycle regresses. // // The bound is the safety net: a hang must not stall the suite, so each run races a // 2s timeout that aborts it. A timed-out run surfaces as `{ timedOut: true }`, which @@ -24,6 +23,15 @@ type Bounded = { timedOut: true } | { timedOut: false; output: ExecOutput }; const BOUND_MS = 2000; +// What a torn-down producer reports is not fixed, so no test here asserts a particular signal. +// Node's stdio is a socketpair rather than a pipe, and a consumer that exits leaving unread bytes +// makes the kernel reset the connection: the producer's write then fails with ECONNRESET and it +// exits non-zero, where a clean close would have raised SIGPIPE. Both happen, and which one is a +// matter of timing. What holds either way is that the producer stopped and did not succeed. +function wasStopped(result: ExecOutput['results'][number]): boolean { + return result != null && result.exitCode !== 0; +} + // Run ExecV3 with a hang guard: if it does not settle within BOUND_MS, abort it and // report the timeout rather than letting the promise (and the suite) hang. async function runBounded(input: z.input): Promise { @@ -45,7 +53,7 @@ async function runBounded(input: z.input): Promise { expect(actual).toBe(expected); }); - it('tears down the first producer (dies from SIGPIPE)', async () => { + it('tears down the first producer', async () => { const outcome = await runBounded(input); - const expected = 'SIGPIPE'; - const actual = outcome.timedOut ? null : outcome.output.results[0]?.signal; + const expected = true; + const actual = !outcome.timedOut && wasStopped(outcome.output.results[0]); expect(actual).toBe(expected); }); - it('tears down the middle stage (dies from SIGPIPE)', async () => { + it('tears down the middle stage', async () => { const outcome = await runBounded(input); - const expected = 'SIGPIPE'; - const actual = outcome.timedOut ? null : outcome.output.results[1]?.signal; + const expected = true; + const actual = !outcome.timedOut && wasStopped(outcome.output.results[1]); expect(actual).toBe(expected); }); @@ -119,13 +127,13 @@ describe('multi-hop teardown — yes | cat | head -n 1', () => { }); // --------------------------------------------------------------------------- -// SIGPIPE death — bash: yes | head -n 1 +// broken-pipe death — bash: yes | head -n 1 // --------------------------------------------------------------------------- // -// A torn-down producer dies from SIGPIPE, the real broken-pipe signal; and overall -// success follows the operator structure — the terminal stage's exit, not the producer's. +// The producer is stopped by the pipe breaking under it, and overall success follows the +// operator structure: the terminal stage's exit, not the producer's. -describe('SIGPIPE death — yes | head -n 1', () => { +describe('broken-pipe death — yes | head -n 1', () => { const input = { intent: 'feed an endless producer into head', commands: [ @@ -134,10 +142,10 @@ describe('SIGPIPE death — yes | head -n 1', () => { ], }; - it('the torn-down producer dies from SIGPIPE', async () => { + it('the torn-down producer is stopped', async () => { const outcome = await runBounded(input); - const expected = 'SIGPIPE'; - const actual = outcome.timedOut ? null : outcome.output.results[0]?.signal; + const expected = true; + const actual = !outcome.timedOut && wasStopped(outcome.output.results[0]); expect(actual).toBe(expected); }); @@ -166,10 +174,10 @@ describe('consumer that never starts — yes | a stage with a missing cwd', () = ], }; - it('the producer dies from SIGPIPE rather than running on', async () => { + it('the producer is stopped rather than running on', async () => { const outcome = await runBounded(input); - const expected = 'SIGPIPE'; - const actual = outcome.timedOut ? null : outcome.output.results[0]?.signal; + const expected = true; + const actual = !outcome.timedOut && wasStopped(outcome.output.results[0]); expect(actual).toBe(expected); }); @@ -233,10 +241,10 @@ describe('external cancel — sleep 5 | cat cancelled mid-flight', () => { // return here — an external abort to release it makes the handler throw ToolCancelledError. // So this drives `evaluate` directly and reads find's result. // -// find dying from SIGPIPE is the proof: SIGPIPE comes only from the consumer-exit teardown -// path. Had find instead survived until the release-abort below, the external-cancel guard -// would leave it a raw SIGTERM, not SIGPIPE — so signal 'SIGPIPE' means head's exit tore it -// down, well before the pipeline (blocked on sleep) ended. +// The proof that it happened early is the absence of SIGTERM. Had yes instead survived until +// the release-abort below, that abort would have killed it with SIGTERM. Any other outcome +// means the broken pipe stopped it when head exited, well before the pipeline (blocked on +// sleep) ended. The specific broken-pipe outcome is not asserted: see wasStopped above. describe('middle-consumer exit — yes | head -n 1 | sleep 500', () => { const commands = [ @@ -245,7 +253,7 @@ describe('middle-consumer exit — yes | head -n 1 | sleep 500', () => { { program: 'sleep', args: ['500'] }, ] satisfies Command[]; - it('tears down the first producer when the middle consumer exits (dies from SIGPIPE)', async () => { + it('tears down the first producer when the middle consumer exits', async () => { // The terminal sleep never exits, so release the pipeline with a short-delay abort; // yes was already torn down the instant head exited, long before this fires. const controller = new AbortController(); @@ -253,8 +261,8 @@ describe('middle-consumer exit — yes | head -n 1 | sleep 500', () => { const executor = new Executor(); try { const output = await evaluate(commands, { cwd: process.cwd(), signal: controller.signal, executor, envProvider: passthroughEnvProvider, now: () => performance.now(), fs: nodeFs }); - const expected = 'SIGPIPE'; - const actual = output.results[0]?.signal; + const expected = true; + const actual = output.results[0]?.signal !== 'SIGTERM'; expect(actual).toBe(expected); } finally { clearTimeout(timer); diff --git a/packages/claude-sdk-tools/test/integration/single-stage-merge.spec.ts b/packages/claude-sdk-tools/test/integration/single-stage-merge.spec.ts new file mode 100644 index 00000000..7b5b9489 --- /dev/null +++ b/packages/claude-sdk-tools/test/integration/single-stage-merge.spec.ts @@ -0,0 +1,52 @@ +import { describe, expect, it } from 'vitest'; +import type { z } from 'zod'; +import { ExecV3, ExecV3InputSchema } from '../../src/entry/ExecV3'; + +// A one-command pipeline that merges stderr never comes back. +// +// resolveStageSinks now builds a stderr capture for every stage, including a merged one, so +// the executor has somewhere to put the diagnostics it writes itself. Executor.runPipeline's +// single-stage path forwards `mergeStderr ? stage.stdout : stage.stderr` to run(), so on a +// merged stage that capture is never handed over, never ended, and the fromStream() awaiting +// it never resolves. The stage's own status resolves; the Promise.all around it does not. +// +// A pipe of two or more is unaffected: its stages settle through finish(), which closes both +// sinks. The single-stage path is the only one that drops one. +// +// This is the shape the tool's own description advertises for capturing a build log, and the +// shape of any `2>&1` command in a && chain, since each link is its own one-stage pipeline. + +const BOUND_MS = 2000; +const HUNG = 'hung'; + +async function settles(input: z.input): Promise { + const guard = new Promise((resolve) => { + setTimeout(() => resolve(HUNG), BOUND_MS).unref(); + }); + // Either settlement proves it came back; only a hang can fail this. + const call = ExecV3.handler(ExecV3InputSchema.parse(input)).then( + () => 'settled', + () => 'settled', + ); + return Promise.race([call, guard]); +} + +describe('a single command that merges stderr into stdout', () => { + it('comes back when its output is captured', async () => { + const expected = 'settled'; + const actual = await settles({ + intent: 'run one command with its stderr merged into stdout', + commands: [{ program: 'echo', args: ['hi'], redirect: { stderr: '&1' } }], + }); + expect(actual).toBe(expected); + }); + + it('comes back when its stdout is redirected to a file', async () => { + const expected = 'settled'; + const actual = await settles({ + intent: 'run one command with stdout to a file and stderr merged into it', + commands: [{ program: 'echo', args: ['hi'], redirect: { stdout: '/tmp/single-stage-merge.log', stderr: '&1' } }], + }); + expect(actual).toBe(expected); + }); +}); diff --git a/packages/exec-core/CHANGELOG.md b/packages/exec-core/CHANGELOG.md index 282c4173..dd5428c7 100644 --- a/packages/exec-core/CHANGELOG.md +++ b/packages/exec-core/CHANGELOG.md @@ -12,7 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Add a README describing the package and pointing to the main documentation - Allow killing a process with a chosen signal - Merge a child's stderr into its stdout by routing both to the same stream -- Run a pipeline as one unit, joining stages with real OS pipes so the kernel provides backpressure and delivers SIGPIPE when a consumer exits +- Run a pipeline as one unit, joining stages at the file descriptor so the kernel provides backpressure and stops a producer whose consumer has exited - Spawned children are detached from the operator's tty and the process group is killed on abort (SIGTERM, then SIGKILL after a grace period) - Stream-based interface for spawning a single process, with stdin, stdout, and stderr wired as streams diff --git a/packages/exec-core/changes.jsonl b/packages/exec-core/changes.jsonl index 5ea86a6e..c3124cea 100644 --- a/packages/exec-core/changes.jsonl +++ b/packages/exec-core/changes.jsonl @@ -4,7 +4,7 @@ {"description":"Add a README describing the package and pointing to the main documentation","category":"added"} {"description":"Allow killing a process with a chosen signal","category":"added"} {"description":"Fix version metadata","category":"fixed"} -{"description":"Run a pipeline as one unit, joining stages with real OS pipes so the kernel provides backpressure and delivers SIGPIPE when a consumer exits","category":"added"} +{"description":"Run a pipeline as one unit, joining stages at the file descriptor so the kernel provides backpressure and stops a producer whose consumer has exited","category":"added"} {"description":"A cancelled multi-stage pipeline now returns instead of hanging after its processes are killed","category":"fixed"} {"description":"Removed the PipeConsumerGone abort reason, which existed only to simulate a broken pipe in userland","category":"removed"} {"description":"IExecutor now requires a runPipeline method, so an existing implementation of the interface must add one","category":"changed"} diff --git a/packages/exec-core/test/integration/Executor.spec.ts b/packages/exec-core/test/integration/Executor.spec.ts index ee7d896a..eb17624a 100644 --- a/packages/exec-core/test/integration/Executor.spec.ts +++ b/packages/exec-core/test/integration/Executor.spec.ts @@ -201,16 +201,21 @@ describe('Executor.runPipeline', () => { expect(actual).toBe(expected); }); - // The kernel, not this package, is what stops the producer: closing the read end is the - // whole mechanism, so this goes red if the fd handoff is ever routed back through the parent. - it('kills a producer with SIGPIPE when its consumer exits early', async () => { + // The kernel, not this package, is what stops the producer: closing the read end is the whole + // mechanism, so an endless producer settling at all is the proof, and this goes red if the fd + // handoff is ever routed back through the parent. What it does not assert is how the death + // reports. Node's stdio is a socketpair, so a consumer exiting with unread bytes left makes the + // kernel reset the connection and the producer's write fails with ECONNRESET, where a clean + // close raises SIGPIPE. Measured over 40 rounds: every time SIGPIPE on macOS, roughly one in + // three on Linux. Only 'stopped, and did not succeed' holds on both. + it('stops a producer when its consumer exits early', async () => { using executor = new Executor(); const terminal = collector(); const [producer] = await Promise.all(executor.runPipeline([{ cmd: spec('yes') }, { cmd: spec('head', ['-n', '1']), stdout: terminal.sink }])); - const expected = 'SIGPIPE'; - const actual = producer.signal; + const expected = true; + const actual = producer.exitCode !== 0; expect(actual).toBe(expected); }); }); From 648acd9f269fef41049e6ab0096273b4d3736299 Mon Sep 17 00:00:00 2001 From: Stephen Hellicar Date: Mon, 3 Aug 2026 22:27:39 +1000 Subject: [PATCH 05/11] Fail a command whose output has nowhere to go A redirect was opened lazily, so a path that could not be written failed on the stream long after the command had run, and that failure was swallowed: the output vanished and the call reported success. Opening before the program starts is the last moment the caller can still be told. Separately, a cwd that was a file passed the existence check and threw at spawn instead, which both escaped as a raw error and left stages already spawned for that pipeline running unwatched. --- apps/claude-sdk-cli/test/MemoryFileSystem.ts | 4 ++ packages/claude-core/src/fs/interfaces.ts | 7 ++++ packages/claude-core/test/MemoryFileSystem.ts | 4 ++ .../claude-core/test/SymlinkFileSystem.ts | 4 ++ .../src/ExecV3/runPipeline.ts | 37 ++++++++++++++++--- .../claude-sdk-tools/src/fs/NodeFileSystem.ts | 7 +++- .../claude-sdk-tools/test/MemoryFileSystem.ts | 16 ++++++++ .../test/find-symlinks.spec.ts | 4 ++ .../integration/redirect-unopenable.spec.ts | 28 ++++++++++++++ packages/exec-core/src/Executor.ts | 30 +++++++++++++-- .../test/integration/Executor.spec.ts | 15 ++++++++ 11 files changed, 147 insertions(+), 9 deletions(-) create mode 100644 packages/claude-sdk-tools/test/integration/redirect-unopenable.spec.ts diff --git a/apps/claude-sdk-cli/test/MemoryFileSystem.ts b/apps/claude-sdk-cli/test/MemoryFileSystem.ts index 78ff27c6..27a0b19f 100644 --- a/apps/claude-sdk-cli/test/MemoryFileSystem.ts +++ b/apps/claude-sdk-cli/test/MemoryFileSystem.ts @@ -274,6 +274,10 @@ export class MemoryFileSystem extends IFileSystem { return this.#arch; } + public openWriteStream(path: string, options: { flags: 'a' | 'w' }): Writable { + return this.createWriteStream(path, options); + } + public createWriteStream(path: string, options: { flags: 'a' | 'w' }): Writable { const initial = options.flags === 'a' ? (this.files.get(path) ?? '') : ''; const chunks: string[] = [initial]; diff --git a/packages/claude-core/src/fs/interfaces.ts b/packages/claude-core/src/fs/interfaces.ts index a9ec0c44..6da32798 100644 --- a/packages/claude-core/src/fs/interfaces.ts +++ b/packages/claude-core/src/fs/interfaces.ts @@ -49,4 +49,11 @@ export abstract class IFileSystem { public abstract arch(): NodeJS.Architecture; /** Open a writable stream to a file, for a redirect target rather than a one-shot write. */ public abstract createWriteStream(path: string, options: { flags: 'a' | 'w' }): Writable; + /** + * The same, except the file is opened before this returns, so a path that cannot be written + * throws here rather than failing later on the stream. That is the difference between a + * caller being able to refuse to run a command whose output would have nowhere to go, and + * reporting success for one whose output it silently discarded. + */ + public abstract openWriteStream(path: string, options: { flags: 'a' | 'w' }): Writable; } diff --git a/packages/claude-core/test/MemoryFileSystem.ts b/packages/claude-core/test/MemoryFileSystem.ts index 77af0345..ea62d20c 100644 --- a/packages/claude-core/test/MemoryFileSystem.ts +++ b/packages/claude-core/test/MemoryFileSystem.ts @@ -122,6 +122,10 @@ export class MemoryFileSystem extends IFileSystem { throw new Error('MemoryFileSystem: createWriteStream() not supported'); } + public openWriteStream(): Writable { + throw new Error('MemoryFileSystem: openWriteStream() not supported'); + } + public readlink(): Promise { throw new Error('MemoryFileSystem: readlink() not supported'); } diff --git a/packages/claude-core/test/SymlinkFileSystem.ts b/packages/claude-core/test/SymlinkFileSystem.ts index 4c6f34fb..92d372a9 100644 --- a/packages/claude-core/test/SymlinkFileSystem.ts +++ b/packages/claude-core/test/SymlinkFileSystem.ts @@ -190,4 +190,8 @@ export class SymlinkFileSystem extends IFileSystem { public createWriteStream(): Writable { throw new Error('SymlinkFileSystem: createWriteStream() not supported'); } + + public openWriteStream(): Writable { + throw new Error('SymlinkFileSystem: openWriteStream() not supported'); + } } diff --git a/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts b/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts index ee4d79f3..81946894 100644 --- a/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts +++ b/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts @@ -26,9 +26,9 @@ function resolveStageSinks(cmd: Command, isLast: boolean, cwd: string, fs: IFile if (redirect?.stdout != null) { // A non-terminal stage with a stdout redirect is rejected at validation (R4), so this // is only reached on a terminal stage. - const file = fs.createWriteStream(resolve(cwd, redirect.stdout), { flags: 'w' }); + const file = fs.openWriteStream(resolve(cwd, redirect.stdout), { flags: 'w' }); file.on('error', () => { - // Redirect write errors should not crash the run. + // A write that fails after the file opened should not crash the run. }); stdout = file; } else if (isLast) { @@ -42,9 +42,9 @@ function resolveStageSinks(cmd: Command, isLast: boolean, cwd: string, fs: IFile let stderr: Writable | undefined; let stderrCapture: PassThrough | undefined; if (!mergeStderr && redirect?.stderr != null) { - const file = fs.createWriteStream(resolve(cwd, redirect.stderr), { flags: 'w' }); + const file = fs.openWriteStream(resolve(cwd, redirect.stderr), { flags: 'w' }); file.on('error', () => { - // Redirect write errors should not crash the run. + // A write that fails after the file opened should not crash the run. }); stderr = file; } else { @@ -55,10 +55,37 @@ function resolveStageSinks(cmd: Command, isLast: boolean, cwd: string, fs: IFile return { stdout, stderr, stdoutCapture, stderrCapture }; } +/** + * A redirect that cannot be opened stops the pipeline before anything is spawned. The stage + * owning the redirect reports why; the others report that they never started. Running them + * anyway is what let a command whose output went nowhere still be reported as a success. + */ +function neverStarted(count: number, failed: number, reason: string): CommandResult[] { + return Array.from({ length: count }, (_, i) => ({ + stdout: '', + stderr: i === failed ? reason : 'not started: another stage in this pipeline could not open its redirect', + exitCode: 1, + signal: null, + durationMs: 0, + })); +} + /** Execute a pipeline (length ≥ 1), one CommandResult per stage. */ export async function runPipeline(commands: Command[], ctx: EngineContext): Promise { const n = commands.length; - const sinks = commands.map((cmd, i) => resolveStageSinks(cmd, i === n - 1, cmd.cwd ?? ctx.cwd, ctx.fs)); + + const sinks: StageSinks[] = []; + for (const [i, cmd] of commands.entries()) { + try { + sinks.push(resolveStageSinks(cmd, i === n - 1, cmd.cwd ?? ctx.cwd, ctx.fs)); + } catch (error) { + for (const opened of sinks) { + opened.stdout?.end(); + opened.stderr?.end(); + } + return neverStarted(n, i, error instanceof Error ? error.message : String(error)); + } + } const stages: PipelineStage[] = commands.map((cmd, i) => ({ cmd: { program: cmd.program, args: cmd.args, cwd: cmd.cwd ?? ctx.cwd, env: ctx.envProvider.buildEnv(cmd.env) }, diff --git a/packages/claude-sdk-tools/src/fs/NodeFileSystem.ts b/packages/claude-sdk-tools/src/fs/NodeFileSystem.ts index ce853b1d..515993ea 100644 --- a/packages/claude-sdk-tools/src/fs/NodeFileSystem.ts +++ b/packages/claude-sdk-tools/src/fs/NodeFileSystem.ts @@ -1,5 +1,5 @@ import type { Stats } from 'node:fs'; -import { createWriteStream, existsSync, lstatSync as fsLstatSync, readlinkSync as fsReadlinkSync, realpathSync as fsRealpathSync } from 'node:fs'; +import { createWriteStream, existsSync, lstatSync as fsLstatSync, openSync, readlinkSync as fsReadlinkSync, realpathSync as fsRealpathSync } from 'node:fs'; import { appendFile, lstat as fsLstat, readdir as fsReaddir, readlink as fsReadlink, realpath as fsRealpath, rename as fsRename, stat as fsStat, mkdir, readFile, rm, rmdir, writeFile } from 'node:fs/promises'; import { homedir as osHomedir, tmpdir as osTmpdir } from 'node:os'; import { dirname } from 'node:path'; @@ -142,6 +142,11 @@ export class NodeFileSystem extends IFileSystem { return createWriteStream(path, options); } + public openWriteStream(path: string, options: { flags: 'a' | 'w' }): Writable { + // openSync is what makes the failure land on the caller rather than on the stream later. + return createWriteStream(path, { fd: openSync(path, options.flags) }); + } + public async readlink(path: string): Promise { return fsReadlink(path); } diff --git a/packages/claude-sdk-tools/test/MemoryFileSystem.ts b/packages/claude-sdk-tools/test/MemoryFileSystem.ts index 68deb6d1..f7401eb4 100644 --- a/packages/claude-sdk-tools/test/MemoryFileSystem.ts +++ b/packages/claude-sdk-tools/test/MemoryFileSystem.ts @@ -219,6 +219,22 @@ export class MemoryFileSystem extends IFileSystem { return this.#arch; } + /** Paths openWriteStream refuses, so a test can exercise a redirect target that cannot be opened. */ + private readonly unopenable = new Set(); + + public refuseOpen(path: string): void { + this.unopenable.add(path); + } + + public openWriteStream(path: string, options: { flags: 'a' | 'w' }): Writable { + if (this.unopenable.has(path)) { + const error = new Error(`ENOENT: no such file or directory, open '${path}'`) as NodeJS.ErrnoException; + error.code = 'ENOENT'; + throw error; + } + return this.createWriteStream(path, options); + } + public createWriteStream(path: string, options: { flags: 'a' | 'w' }): Writable { const initial = options.flags === 'a' ? (this.files.get(path) ?? Buffer.alloc(0)) : Buffer.alloc(0); const chunks: Buffer[] = [initial]; diff --git a/packages/claude-sdk-tools/test/find-symlinks.spec.ts b/packages/claude-sdk-tools/test/find-symlinks.spec.ts index c2393abe..325fcbe8 100644 --- a/packages/claude-sdk-tools/test/find-symlinks.spec.ts +++ b/packages/claude-sdk-tools/test/find-symlinks.spec.ts @@ -64,6 +64,10 @@ class SymlinkMockFileSystem extends IFileSystem { throw new Error('not implemented'); } + public openWriteStream(): Writable { + throw new Error('not implemented'); + } + public cwd(): string { return ROOT; } diff --git a/packages/claude-sdk-tools/test/integration/redirect-unopenable.spec.ts b/packages/claude-sdk-tools/test/integration/redirect-unopenable.spec.ts new file mode 100644 index 00000000..7b797a4b --- /dev/null +++ b/packages/claude-sdk-tools/test/integration/redirect-unopenable.spec.ts @@ -0,0 +1,28 @@ +import { describe, expect, it } from 'vitest'; +import { ExecV3 } from '../../src/entry/ExecV3'; +import { call } from '../helpers'; + +// A redirect that cannot be opened must not be reported as success. The output has nowhere to +// go, so a caller told the command succeeded will believe a file exists that never did, and +// act on it. Bash refuses the command outright; the requirement here is only that the failure +// reaches the caller instead of being swallowed. + +const unopenable = '/nonexistent-dir-xyzzy-redirect/out.log'; + +describe('a stdout redirect that cannot be opened', () => { + it('does not report success', async () => { + const result = await call(ExecV3, { intent: 'redirect stdout to a path that cannot be opened', commands: [{ program: 'echo', args: ['this-should-not-vanish'], redirect: { stdout: unopenable } }] }); + + const expected = false; + const actual = result.success; + expect(actual).toBe(expected); + }); + + it('says why it failed', async () => { + const result = await call(ExecV3, { intent: 'redirect stdout to a path that cannot be opened', commands: [{ program: 'echo', args: ['this-should-not-vanish'], redirect: { stdout: unopenable } }] }); + + const expected = true; + const actual = (result.results[0]?.stderr.length ?? 0) > 0; + expect(actual).toBe(expected); + }); +}); diff --git a/packages/exec-core/src/Executor.ts b/packages/exec-core/src/Executor.ts index e3a1c6d6..97054a28 100644 --- a/packages/exec-core/src/Executor.ts +++ b/packages/exec-core/src/Executor.ts @@ -1,9 +1,19 @@ import { type ChildProcess, spawn } from 'node:child_process'; -import { existsSync } from 'node:fs'; +import { statSync } from 'node:fs'; import { Duplex, type Writable } from 'node:stream'; import { finished } from 'node:stream/promises'; import type { CommandSpec, ExitStatus, IExecutor, PipelineOpts, PipelineStage, SpawnOpts } from './types.js'; +// A cwd must exist *and* be a directory. Merely existing is not enough: spawn throws ENOTDIR +// synchronously for a file, which escapes as a raw error instead of the stage reporting 126. +function isDirectory(path: string): boolean { + try { + return statSync(path).isDirectory(); + } catch { + return false; + } +} + // End each distinct output sink, and wait for the ones whose flush is ours to complete. // // A write-only sink is a destination this process owns the whole of: a file, where 'finish' @@ -59,7 +69,7 @@ export class Executor implements IExecutor { return { exitCode: null, signal: 'SIGTERM' }; } - if (!existsSync(cmd.cwd)) { + if (!isDirectory(cmd.cwd)) { opts.stderr?.write(`Working directory not found: ${cmd.cwd}`); await closeSinks([opts.stdout, opts.stderr]); return { exitCode: 126, signal: null }; @@ -177,11 +187,17 @@ export class Executor implements IExecutor { const children = new Array(n); + // Spawning is synchronous, so a throw part-way through would leave every stage spawned + // before it running with nobody watching: no close handler, no result, alive until the + // process exits. Each child is owned by this stack until they are all up and the close + // handlers below take over, at which point the stack is released. + using spawned = new DisposableStack(); + // Spawn from the tail, because a stage's stdout IS its consumer's stdin fd and that fd // only exists once the consumer has been spawned. for (let i = n - 1; i >= 0; i--) { const stage = stages[i]; - if (!existsSync(stage.cmd.cwd)) { + if (!isDirectory(stage.cmd.cwd)) { stage.stderr?.write(`Working directory not found: ${stage.cmd.cwd}`); void finish(i, { exitCode: 126, signal: null }); continue; @@ -201,6 +217,12 @@ export class Executor implements IExecutor { if (child.pid != null) { this.#pids.add(child.pid); } + spawned.defer(() => { + if (child.pid != null) { + this.#groupKill(child.pid); + this.#pids.delete(child.pid); + } + }); } // Drop this process's copy of each write end now that the producer holds its own. While @@ -281,6 +303,8 @@ export class Executor implements IExecutor { opts.signal?.addEventListener('abort', onAbort, { once: true }); void Promise.all(results).then(() => opts.signal?.removeEventListener('abort', onAbort)); + // Every stage is up and watched, so the stack has nothing left to rescue. + spawned.move(); return results; } diff --git a/packages/exec-core/test/integration/Executor.spec.ts b/packages/exec-core/test/integration/Executor.spec.ts index eb17624a..78094b7b 100644 --- a/packages/exec-core/test/integration/Executor.spec.ts +++ b/packages/exec-core/test/integration/Executor.spec.ts @@ -1,4 +1,5 @@ import { PassThrough, Writable } from 'node:stream'; +import { fileURLToPath } from 'node:url'; import { describe, expect, it } from 'vitest'; import { Executor } from '../../src/Executor.js'; import { fromStream } from '../../src/fromStream.js'; @@ -201,6 +202,20 @@ describe('Executor.runPipeline', () => { expect(actual).toBe(expected); }); + // A cwd that exists but is a file used to pass the existence check and then make spawn throw + // ENOTDIR synchronously, which escaped as a raw error rather than the stage reporting 126. + it('reports 126 for a cwd that exists but is a file', async () => { + using executor = new Executor(); + const terminal = collector(); + const diagnostics = collector(); + + const [status] = await Promise.all(executor.runPipeline([{ cmd: { program: 'echo', args: ['x'], cwd: fileURLToPath(import.meta.url), env: process.env }, stdout: terminal.sink, stderr: diagnostics.sink }])); + + const expected = 126; + const actual = status.exitCode; + expect(actual).toBe(expected); + }); + // The kernel, not this package, is what stops the producer: closing the read end is the whole // mechanism, so an endless producer settling at all is the proof, and this goes red if the fd // handoff is ever routed back through the parent. What it does not assert is how the death From db8ec9b9854387f08c123c55892fdf86b0fc3d11 Mon Sep 17 00:00:00 2001 From: Stephen Hellicar Date: Mon, 3 Aug 2026 22:53:56 +1000 Subject: [PATCH 06/11] Bound a command's captured output so it cannot take down the call A capture was collected into one string with nothing limiting it, and `yes` produces about 1.5GB a second, so a single word reached V8's maximum string length in well under a second and the whole call died with a message about string lengths. Reading has to continue past the limit and discard, because a capture that stops being read stalls the process filling it. Separately, stdout and stderr aimed at the same file each opened their own stream at offset zero, so one overwrote the other and half the output went missing while the command reported success. --- packages/claude-core/CHANGELOG.md | 1 + packages/claude-core/changes.jsonl | 1 + packages/claude-sdk-tools/CHANGELOG.md | 3 + packages/claude-sdk-tools/changes.jsonl | 3 + .../src/ExecV3/runPipeline.ts | 22 ++++- .../claude-sdk-tools/src/ExecV3/schema.ts | 11 +++ .../test/ExecV3/redirect.spec.ts | 85 +++++++++++++++++++ .../test/ExecV3/scenarios.spec.ts | 20 +++++ packages/exec-core/CHANGELOG.md | 3 + packages/exec-core/changes.jsonl | 3 + packages/exec-core/package.json | 1 + packages/exec-core/src/entry/index.ts | 6 +- packages/exec-core/src/fromStream.ts | 34 ++++++++ packages/exec-core/test/drainToString.spec.ts | 53 ++++++++++++ packages/exec-core/vitest.config.ts | 9 ++ 15 files changed, 248 insertions(+), 7 deletions(-) create mode 100644 packages/claude-sdk-tools/test/ExecV3/redirect.spec.ts create mode 100644 packages/exec-core/test/drainToString.spec.ts create mode 100644 packages/exec-core/vitest.config.ts diff --git a/packages/claude-core/CHANGELOG.md b/packages/claude-core/CHANGELOG.md index 4f2b4fc7..bf241405 100644 --- a/packages/claude-core/CHANGELOG.md +++ b/packages/claude-core/CHANGELOG.md @@ -32,6 +32,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Adopt core-di-lite property injection: config loading splits into a pure read, a holder, and a watch handle with no load or start step, and the shared provider and contract abstractions live here for every package to resolve against - Depend on @shellicar/core-di instead of @shellicar/core-di-lite - File discovery returns records carrying type, size, and symlink target instead of bare path strings +- IFileSystem gained openWriteStream, which opens the file before returning so an unwritable path fails at the caller rather than later on the stream; an existing implementation must add it - setupKeypressHandler accepts an optional escFastPathEnabled callback, checked live before taking the lone-ESC fast path, so a consumer can disable it (e.g. over a fragmented remote connection) without a restart - Update runtime and build dependencies - Updated patch and minor dependencies diff --git a/packages/claude-core/changes.jsonl b/packages/claude-core/changes.jsonl index 802ac550..73054d61 100644 --- a/packages/claude-core/changes.jsonl +++ b/packages/claude-core/changes.jsonl @@ -28,3 +28,4 @@ {"description":"IFileSystem gains tmpdir, uid, mkdir with an explicit mode, lstat, and a synchronous readlinkSync, so code that needs a temporary directory, or has to create one and check who owns it, can reach all of it through the filesystem seam instead of node:os and node:fs directly","category":"added"} {"description":"StatResult now carries uid and mode, so a caller can tell who owns a path and who else can reach it","category":"added"} {"description":"canonicalisePath resolves a path to where it actually lands, following symlinks even when the target does not exist yet, for callers that must decide on the destination rather than the string they were handed","category":"added"} +{"description":"IFileSystem gained openWriteStream, which opens the file before returning so an unwritable path fails at the caller rather than later on the stream; an existing implementation must add it","category":"changed"} diff --git a/packages/claude-sdk-tools/CHANGELOG.md b/packages/claude-sdk-tools/CHANGELOG.md index 155c69e9..fc54d800 100644 --- a/packages/claude-sdk-tools/CHANGELOG.md +++ b/packages/claude-sdk-tools/CHANGELOG.md @@ -90,6 +90,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- A command producing very large output no longer fails the whole ExecV3 call; each captured stream keeps up to 8 MiB and says so when there was more +- A command whose redirect target cannot be opened now fails and says why, instead of running and reporting success while its output went nowhere - A failed tsserver request now throws instead of returning an empty result that was indistinguishable from a clean file - An interactive az identity no longer gets a silent, unattended background relogin; the browser/MFA prompt only ever appears attached to a real caller's call - AzCli, EscalatedAzCli, and every AzureDevOps.PullRequest.* tool now honor cancellation — an in-progress az login or command can be aborted instead of blocking until the process crashes or restarts @@ -97,6 +99,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Binary files are blocked from text reads when the format is recognised; unrecognised formats are still treated as text - ExecV3 and Memory import defineTool, ToolCancelledError, ToolRefusedError, and pathSchema from their own claude-sdk subpaths instead of the barrel, so a consumer bundling this package no longer pulls in the whole SDK module graph - ExecV3 pipelines now run over real OS pipes, so a cancelled or timed-out pipe returns instead of hanging the caller +- ExecV3 rejects stdout and stderr redirected to the same file, which silently discarded one of them; use stderr "&1" to merge - Find tool follows symlinks with cycle detection - Fix version metadata - GitHub_PullRequest_AutoMerge takes a required strategy (merge, squash, rebase) when enabling, so it can queue a specific merge method instead of only accepting the repo default diff --git a/packages/claude-sdk-tools/changes.jsonl b/packages/claude-sdk-tools/changes.jsonl index 4dd58627..51ee898f 100644 --- a/packages/claude-sdk-tools/changes.jsonl +++ b/packages/claude-sdk-tools/changes.jsonl @@ -89,3 +89,6 @@ {"description":"The az session's own login and command env now strips the same ambient Azure credential vars ExecV3 strips, so the CLI's own environment can no longer steer a login it believes it fully controls","category":"security"} {"description":"NodeFileSystem implements the new IFileSystem members: the real OS temp directory, the process user id, a recursive create that honours an explicit mode, a symlink-preserving lstat, and a readlinkSync that answers null rather than throwing when there is nothing to follow","category":"added"} {"description":"ExecV3 pipelines now run over real OS pipes, so a cancelled or timed-out pipe returns instead of hanging the caller","category":"fixed"} +{"description":"A command whose redirect target cannot be opened now fails and says why, instead of running and reporting success while its output went nowhere","category":"fixed"} +{"description":"ExecV3 rejects stdout and stderr redirected to the same file, which silently discarded one of them; use stderr \"&1\" to merge","category":"fixed"} +{"description":"A command producing very large output no longer fails the whole ExecV3 call; each captured stream keeps up to 8 MiB and says so when there was more","category":"fixed"} diff --git a/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts b/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts index 81946894..48975968 100644 --- a/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts +++ b/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts @@ -1,10 +1,23 @@ import { resolve } from 'node:path'; import { PassThrough, Readable, type Writable } from 'node:stream'; import type { IFileSystem } from '@shellicar/claude-core/fs/interfaces'; -import { fromStream, type PipelineStage } from '@shellicar/exec-core'; +import { type DrainedStream, drainToString, type PipelineStage } from '@shellicar/exec-core'; import type { EngineContext } from './engine'; import type { Command, CommandResult } from './types'; +/** + * How much of one stream a result will carry. Not a policy about how much output a command may + * produce: it is a backstop, set far above anything a caller reads and far below the point where + * building the string fails outright. `yes` clears half a gigabyte in under a second, and + * unbounded that ends the whole call with a message about string lengths. + */ +const CAPTURE_LIMIT_BYTES = 8 * 1024 * 1024; + +/** Truncation is never silent: the text says so, and says how much there was. */ +function captured(drained: DrainedStream): string { + return drained.truncated ? `${drained.text}\n[truncated: kept ${CAPTURE_LIMIT_BYTES} bytes of ${drained.bytes}]` : drained.text; +} + interface StageSinks { stdout?: Writable; stderr?: Writable; @@ -105,12 +118,13 @@ export async function runPipeline(commands: Command[], ctx: EngineContext): Prom return Promise.all( runs.map((run, i) => { const { stdoutCapture, stderrCapture } = sinks[i]; - return Promise.all([run, stdoutCapture ? fromStream(stdoutCapture) : Promise.resolve(''), stderrCapture ? fromStream(stderrCapture) : Promise.resolve('')]).then(([status, out, err]): CommandResult => { + const empty: DrainedStream = { text: '', bytes: 0, truncated: false }; + return Promise.all([run, stdoutCapture ? drainToString(stdoutCapture, CAPTURE_LIMIT_BYTES) : Promise.resolve(empty), stderrCapture ? drainToString(stderrCapture, CAPTURE_LIMIT_BYTES) : Promise.resolve(empty)]).then(([status, out, err]): CommandResult => { // A producer whose consumer exited dies from a kernel SIGPIPE, so its real exit is // already the honest broken-pipe death. Report it as-is. return { - stdout: out, - stderr: err, + stdout: captured(out), + stderr: captured(err), exitCode: status.exitCode, signal: status.signal, durationMs: Math.round(ctx.now() - startedAt[i]), diff --git a/packages/claude-sdk-tools/src/ExecV3/schema.ts b/packages/claude-sdk-tools/src/ExecV3/schema.ts index 3d337834..531834fd 100644 --- a/packages/claude-sdk-tools/src/ExecV3/schema.ts +++ b/packages/claude-sdk-tools/src/ExecV3/schema.ts @@ -122,6 +122,17 @@ export const ExecV3InputSchema = z }); } + // R5: stdout and stderr aimed at the same file. Each becomes its own stream opened at + // offset zero, so one overwrites the other and half the output is lost while the command + // still reports success. "&1" is the way to say what this was trying to say. + if (cmd.redirect?.stdout != null && cmd.redirect.stderr != null && cmd.redirect.stderr !== '&1' && cmd.redirect.stdout === cmd.redirect.stderr) { + ctx.addIssue({ + code: 'custom', + path: ['commands', i, 'redirect', 'stderr'], + message: 'stdout and stderr cannot both be redirected to the same file; use stderr: "&1" to merge them', + }); + } + // NE2: stdin literal on the TARGET of a pipe (previous command had op "|") — // the pipe occupies stdin. if (prev?.op === '|' && cmd.stdin != null) { diff --git a/packages/claude-sdk-tools/test/ExecV3/redirect.spec.ts b/packages/claude-sdk-tools/test/ExecV3/redirect.spec.ts new file mode 100644 index 00000000..6bc6fdb7 --- /dev/null +++ b/packages/claude-sdk-tools/test/ExecV3/redirect.spec.ts @@ -0,0 +1,85 @@ +import { describe, expect, it } from 'vitest'; +import { StaticRulesConfigProvider } from '../../src/Exec/IRulesConfigProvider'; +import { createExecV3 } from '../../src/ExecV3/ExecV3'; +import { passthroughEnvProvider } from '../../src/entry/ExecV3'; +import { FakeExecutor, shellLikeResponder } from '../FakeExecutor'; +import { MemoryFileSystem } from '../MemoryFileSystem'; + +// The claim worth pinning is not that the call fails, but that the command never ran. A command +// whose output has nowhere to go must not have had its side effects before anyone was told: that +// is the difference between a caller who can retry and one who has already changed something. +// No real filesystem and no real process — the fake refuses the open, the fake executor records +// whether it was ever asked to run anything. + +const refused = '/refused/out.log'; + +function toolRefusing(target: string) { + const fs = new MemoryFileSystem(); + fs.refuseOpen(target); + const executor = new FakeExecutor(shellLikeResponder()); + return { tool: createExecV3(fs, executor, passthroughEnvProvider, new StaticRulesConfigProvider()), executor }; +} + +const input = { intent: 'redirect stdout to a target that cannot be opened', commands: [{ program: 'echo', args: ['gone'], redirect: { stdout: refused } }] }; + +describe('a redirect target that cannot be opened', () => { + it('never runs the command', async () => { + const { tool, executor } = toolRefusing(refused); + + await tool.handler(tool.input_schema.parse(input)); + + const expected = 0; + const actual = executor.calls.length; + expect(actual).toBe(expected); + }); + + it('does not report success', async () => { + const { tool } = toolRefusing(refused); + + const { textContent } = await tool.handler(tool.input_schema.parse(input)); + + const expected = false; + const actual = textContent.success; + expect(actual).toBe(expected); + }); + + it('names the target that could not be opened', async () => { + const { tool } = toolRefusing(refused); + + const { textContent } = await tool.handler(tool.input_schema.parse(input)); + + const expected = true; + const actual = textContent.results[0]?.stderr.includes(refused) ?? false; + expect(actual).toBe(expected); + }); +}); + +describe('a pipeline where one stage cannot open its redirect', () => { + const piped = { + intent: 'pipe into a stage whose stderr redirect cannot be opened', + commands: [ + { program: 'echo', args: ['a'], op: '|' as const }, + { program: 'cat', redirect: { stderr: refused } }, + ], + }; + + it('runs no stage of that pipeline', async () => { + const { tool, executor } = toolRefusing(refused); + + await tool.handler(tool.input_schema.parse(piped)); + + const expected = 0; + const actual = executor.calls.length; + expect(actual).toBe(expected); + }); + + it('tells the other stage it never started', async () => { + const { tool } = toolRefusing(refused); + + const { textContent } = await tool.handler(tool.input_schema.parse(piped)); + + const expected = true; + const actual = textContent.results[0]?.stderr.includes('not started') ?? false; + expect(actual).toBe(expected); + }); +}); diff --git a/packages/claude-sdk-tools/test/ExecV3/scenarios.spec.ts b/packages/claude-sdk-tools/test/ExecV3/scenarios.spec.ts index 0e78d16a..a10ad117 100644 --- a/packages/claude-sdk-tools/test/ExecV3/scenarios.spec.ts +++ b/packages/claude-sdk-tools/test/ExecV3/scenarios.spec.ts @@ -650,6 +650,26 @@ describe('validation — pipe with stdout redirect (R4)', () => { }); }); +describe('validation — stdout and stderr to the same file (R5)', () => { + it('rejects two redirects aimed at one file', () => { + const expected = false; + const actual = ExecV3InputSchema.safeParse({ intent: 'x', commands: [{ program: 'echo', redirect: { stdout: '/tmp/a.log', stderr: '/tmp/a.log' } }] }).success; + expect(actual).toBe(expected); + }); + + it('accepts "&1", which is how the same file is actually asked for', () => { + const expected = true; + const actual = ExecV3InputSchema.safeParse({ intent: 'x', commands: [{ program: 'echo', redirect: { stdout: '/tmp/a.log', stderr: '&1' } }] }).success; + expect(actual).toBe(expected); + }); + + it('accepts two redirects aimed at different files', () => { + const expected = true; + const actual = ExecV3InputSchema.safeParse({ intent: 'x', commands: [{ program: 'echo', redirect: { stdout: '/tmp/a.log', stderr: '/tmp/b.log' } }] }).success; + expect(actual).toBe(expected); + }); +}); + describe('validation — stdin on a pipe target (NE2)', () => { it('rejects stdin on the target of a pipe', () => { const expected = false; diff --git a/packages/exec-core/CHANGELOG.md b/packages/exec-core/CHANGELOG.md index dd5428c7..8b4660d8 100644 --- a/packages/exec-core/CHANGELOG.md +++ b/packages/exec-core/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added - Add a README describing the package and pointing to the main documentation +- Added drainToString, which collects a stream to a string while keeping at most a given number of bytes and reporting how many there were - Allow killing a process with a chosen signal - Merge a child's stderr into its stdout by routing both to the same stream - Run a pipeline as one unit, joining stages at the file descriptor so the kernel provides backpressure and stops a producer whose consumer has exited @@ -28,4 +29,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - A cancelled multi-stage pipeline now returns instead of hanging after its processes are killed - A command no longer hangs when an output sink it was given is drained slowly or not at all +- A pipeline that cannot start one of its stages no longer leaves the stages already started running unwatched +- A working directory that exists but is not a directory now reports 126 like a missing one, instead of failing the whole call with a raw spawn error - Fix version metadata diff --git a/packages/exec-core/changes.jsonl b/packages/exec-core/changes.jsonl index c3124cea..38e145af 100644 --- a/packages/exec-core/changes.jsonl +++ b/packages/exec-core/changes.jsonl @@ -9,3 +9,6 @@ {"description":"Removed the PipeConsumerGone abort reason, which existed only to simulate a broken pipe in userland","category":"removed"} {"description":"IExecutor now requires a runPipeline method, so an existing implementation of the interface must add one","category":"changed"} {"description":"A command no longer hangs when an output sink it was given is drained slowly or not at all","category":"fixed"} +{"description":"A working directory that exists but is not a directory now reports 126 like a missing one, instead of failing the whole call with a raw spawn error","category":"fixed"} +{"description":"A pipeline that cannot start one of its stages no longer leaves the stages already started running unwatched","category":"fixed"} +{"description":"Added drainToString, which collects a stream to a string while keeping at most a given number of bytes and reporting how many there were","category":"added"} diff --git a/packages/exec-core/package.json b/packages/exec-core/package.json index ec14dc6a..f8c341d8 100644 --- a/packages/exec-core/package.json +++ b/packages/exec-core/package.json @@ -43,6 +43,7 @@ "dev": "tsup --watch", "watch": "tsup --watch", "type-check": "tsc -p tsconfig.check.json", + "test": "vitest run", "test:integration": "vitest run --config vitest.integration.config.ts" }, "devDependencies": { diff --git a/packages/exec-core/src/entry/index.ts b/packages/exec-core/src/entry/index.ts index 84e3ce6b..f8840703 100644 --- a/packages/exec-core/src/entry/index.ts +++ b/packages/exec-core/src/entry/index.ts @@ -1,6 +1,6 @@ import { Executor } from '../Executor.js'; -import { fromStream } from '../fromStream.js'; +import { type DrainedStream, drainToString, fromStream } from '../fromStream.js'; import type { CommandSpec, ExitStatus, IExecutor, PipelineOpts, PipelineStage, SpawnOpts } from '../types.js'; -export type { CommandSpec, ExitStatus, IExecutor, PipelineOpts, PipelineStage, SpawnOpts }; -export { Executor, fromStream }; +export type { CommandSpec, DrainedStream, ExitStatus, IExecutor, PipelineOpts, PipelineStage, SpawnOpts }; +export { drainToString, Executor, fromStream }; diff --git a/packages/exec-core/src/fromStream.ts b/packages/exec-core/src/fromStream.ts index c5eb5be7..2bd4c878 100644 --- a/packages/exec-core/src/fromStream.ts +++ b/packages/exec-core/src/fromStream.ts @@ -8,3 +8,37 @@ export async function fromStream(stream: Readable): Promise { } return Buffer.concat(chunks).toString('utf-8'); } + +/** What a bounded drain saw: the text it kept, and how much went past it. */ +export interface DrainedStream { + text: string; + /** Every byte the stream produced, including the ones dropped. */ + bytes: number; + truncated: boolean; +} + +/** + * Collect a readable stream to a UTF-8 string, keeping at most `limit` bytes. + * + * Reading continues past the limit and the excess is dropped, which is the whole point: a + * capture that stops being read stalls the process filling it, so discarding is the only way to + * bound the memory without stalling the writer. Unbounded, a command that produces output faster + * than it is consumed reaches V8's maximum string length and takes the whole call down with it. + */ +export async function drainToString(stream: Readable, limit: number): Promise { + const chunks: Buffer[] = []; + let kept = 0; + let bytes = 0; + + for await (const chunk of stream) { + const buffer = chunk as Buffer; + bytes += buffer.length; + if (kept < limit) { + const slice = buffer.length <= limit - kept ? buffer : buffer.subarray(0, limit - kept); + chunks.push(slice); + kept += slice.length; + } + } + + return { text: Buffer.concat(chunks).toString('utf-8'), bytes, truncated: bytes > kept }; +} diff --git a/packages/exec-core/test/drainToString.spec.ts b/packages/exec-core/test/drainToString.spec.ts new file mode 100644 index 00000000..73265ad7 --- /dev/null +++ b/packages/exec-core/test/drainToString.spec.ts @@ -0,0 +1,53 @@ +import { Readable } from 'node:stream'; +import { describe, expect, it } from 'vitest'; +import { drainToString } from '../src/fromStream.js'; + +// No process and no disk: a bounded drain is about what it keeps and what it consumes, and both +// are observable from a plain readable. + +const streamOf = (...parts: string[]): Readable => Readable.from(parts.map((part) => Buffer.from(part))); + +describe('drainToString within the limit', () => { + it('keeps everything', async () => { + const expected = 'abcdef'; + const { text } = await drainToString(streamOf('abc', 'def'), 100); + expect(text).toBe(expected); + }); + + it('is not marked truncated', async () => { + const expected = false; + const { truncated } = await drainToString(streamOf('abc', 'def'), 100); + expect(truncated).toBe(expected); + }); +}); + +describe('drainToString past the limit', () => { + it('keeps exactly the limit', async () => { + const expected = 'abcd'; + const { text } = await drainToString(streamOf('abc', 'def'), 4); + expect(text).toBe(expected); + }); + + it('is marked truncated', async () => { + const expected = true; + const { truncated } = await drainToString(streamOf('abc', 'def'), 4); + expect(truncated).toBe(expected); + }); + + // The count is what makes the truncation honest rather than a silently short result, and it + // only exists because reading continued past the limit instead of stopping. + it('reports every byte the stream produced, not just the kept ones', async () => { + const expected = 6; + const { bytes } = await drainToString(streamOf('abc', 'def'), 4); + expect(bytes).toBe(expected); + }); + + it('drains the stream to the end', async () => { + const stream = streamOf('abc', 'def'); + await drainToString(stream, 4); + + const expected = true; + const actual = stream.readableEnded; + expect(actual).toBe(expected); + }); +}); diff --git a/packages/exec-core/vitest.config.ts b/packages/exec-core/vitest.config.ts new file mode 100644 index 00000000..65bff0ae --- /dev/null +++ b/packages/exec-core/vitest.config.ts @@ -0,0 +1,9 @@ +import { defineConfig } from 'vitest/config'; + +// The default tier: specs directly under test/, which touch no real process or disk. The ones +// under test/integration/ spawn for real and are excluded here, run only by `pnpm test:integration`. +export default defineConfig({ + test: { + include: ['test/*.spec.ts'], + }, +}); From 8e84bfb2b58a394f71168a8bbed4884df210a0b8 Mon Sep 17 00:00:00 2001 From: Stephen Hellicar Date: Mon, 3 Aug 2026 23:13:22 +1000 Subject: [PATCH 07/11] Refuse two redirects that resolve to one file, whatever the spelling Validation already refuses the same path written twice, but it runs before any working directory is known, so all it can compare is the two strings. Resolved against the command's own cwd, two spellings turn out to name one file, and two streams on one file each open at offset zero: one overwrites the other and half the output goes missing while the command reports success. --- .../src/ExecV3/runPipeline.ts | 18 +++++++--- .../test/ExecV3/redirect.spec.ts | 35 +++++++++++++++++++ 2 files changed, 49 insertions(+), 4 deletions(-) diff --git a/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts b/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts index 48975968..34298e2b 100644 --- a/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts +++ b/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts @@ -33,13 +33,23 @@ interface StageSinks { function resolveStageSinks(cmd: Command, isLast: boolean, cwd: string, fs: IFileSystem): StageSinks { const redirect = cmd.redirect; const mergeStderr = redirect?.stderr === '&1'; + const stdoutTarget = redirect?.stdout != null ? resolve(cwd, redirect.stdout) : undefined; + const stderrTarget = !mergeStderr && redirect?.stderr != null ? resolve(cwd, redirect.stderr) : undefined; + + // Validation (R5) already refuses the same path written twice, but it runs before any cwd is + // known, so all it can compare is the two strings. Resolved against this command's own cwd, + // two spellings of one file become the same path — and two streams on one file each open at + // offset zero, so one silently overwrites the other. + if (stdoutTarget != null && stdoutTarget === stderrTarget) { + throw new Error(`stdout and stderr both resolve to ${stdoutTarget}; use stderr: "&1" to merge them`); + } let stdout: Writable | undefined; let stdoutCapture: PassThrough | undefined; - if (redirect?.stdout != null) { + if (stdoutTarget != null) { // A non-terminal stage with a stdout redirect is rejected at validation (R4), so this // is only reached on a terminal stage. - const file = fs.openWriteStream(resolve(cwd, redirect.stdout), { flags: 'w' }); + const file = fs.openWriteStream(stdoutTarget, { flags: 'w' }); file.on('error', () => { // A write that fails after the file opened should not crash the run. }); @@ -54,8 +64,8 @@ function resolveStageSinks(cmd: Command, isLast: boolean, cwd: string, fs: IFile // has to reach the caller on the stage that failed. let stderr: Writable | undefined; let stderrCapture: PassThrough | undefined; - if (!mergeStderr && redirect?.stderr != null) { - const file = fs.openWriteStream(resolve(cwd, redirect.stderr), { flags: 'w' }); + if (stderrTarget != null) { + const file = fs.openWriteStream(stderrTarget, { flags: 'w' }); file.on('error', () => { // A write that fails after the file opened should not crash the run. }); diff --git a/packages/claude-sdk-tools/test/ExecV3/redirect.spec.ts b/packages/claude-sdk-tools/test/ExecV3/redirect.spec.ts index 6bc6fdb7..98b43140 100644 --- a/packages/claude-sdk-tools/test/ExecV3/redirect.spec.ts +++ b/packages/claude-sdk-tools/test/ExecV3/redirect.spec.ts @@ -20,6 +20,11 @@ function toolRefusing(target: string) { return { tool: createExecV3(fs, executor, passthroughEnvProvider, new StaticRulesConfigProvider()), executor }; } +function plainTool() { + const executor = new FakeExecutor(shellLikeResponder()); + return { tool: createExecV3(new MemoryFileSystem(), executor, passthroughEnvProvider, new StaticRulesConfigProvider()), executor }; +} + const input = { intent: 'redirect stdout to a target that cannot be opened', commands: [{ program: 'echo', args: ['gone'], redirect: { stdout: refused } }] }; describe('a redirect target that cannot be opened', () => { @@ -54,6 +59,36 @@ describe('a redirect target that cannot be opened', () => { }); }); +// Validation rejects the same path written twice, but it runs before any cwd is known, so it +// compares strings. These two spellings only become one file once resolved against the +// command's cwd, which is where the second guard lives. +describe('stdout and stderr that resolve to the same file', () => { + const aliased = { + intent: 'aim both streams at one file, spelled two ways', + commands: [{ program: 'echo', args: ['x'], cwd: '/work', redirect: { stdout: 'out.log', stderr: '/work/out.log' } }], + }; + + it('never runs the command', async () => { + const { tool, executor } = plainTool(); + + await tool.handler(tool.input_schema.parse(aliased)); + + const expected = 0; + const actual = executor.calls.length; + expect(actual).toBe(expected); + }); + + it('names the file both streams resolved to', async () => { + const { tool } = plainTool(); + + const { textContent } = await tool.handler(tool.input_schema.parse(aliased)); + + const expected = true; + const actual = textContent.results[0]?.stderr.includes('/work/out.log') ?? false; + expect(actual).toBe(expected); + }); +}); + describe('a pipeline where one stage cannot open its redirect', () => { const piped = { intent: 'pipe into a stage whose stderr redirect cannot be opened', From 662ea84afd9f376f77b45104feb4226c8a87a404 Mon Sep 17 00:00:00 2001 From: Stephen Hellicar Date: Mon, 3 Aug 2026 23:19:25 +1000 Subject: [PATCH 08/11] Record that a broken-pipe stop reports two ways --- packages/claude-sdk-tools/CHANGELOG.md | 1 + packages/claude-sdk-tools/changes.jsonl | 1 + packages/exec-core/CHANGELOG.md | 1 + packages/exec-core/changes.jsonl | 1 + 4 files changed, 4 insertions(+) diff --git a/packages/claude-sdk-tools/CHANGELOG.md b/packages/claude-sdk-tools/CHANGELOG.md index fc54d800..6b5e1d0b 100644 --- a/packages/claude-sdk-tools/CHANGELOG.md +++ b/packages/claude-sdk-tools/CHANGELOG.md @@ -57,6 +57,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- A pipe stage stopped because its consumer exited may report a SIGPIPE signal or a non-zero exit carrying the program's own broken-pipe message; both mean the pipe broke, and which one appears is timing - Adopt core-di-lite property injection: TsServerService resolves its options through injection and disposes its tsserver process on scope exit - Az account changes take effect immediately across AzCli, EscalatedAzCli, and AzureDevOps.PullRequest.*, with no restart - AzureDevOps.PullRequest.* tools reuse AzCli/EscalatedAzCli's session cache instead of logging in fresh each call diff --git a/packages/claude-sdk-tools/changes.jsonl b/packages/claude-sdk-tools/changes.jsonl index 51ee898f..2c81f572 100644 --- a/packages/claude-sdk-tools/changes.jsonl +++ b/packages/claude-sdk-tools/changes.jsonl @@ -92,3 +92,4 @@ {"description":"A command whose redirect target cannot be opened now fails and says why, instead of running and reporting success while its output went nowhere","category":"fixed"} {"description":"ExecV3 rejects stdout and stderr redirected to the same file, which silently discarded one of them; use stderr \"&1\" to merge","category":"fixed"} {"description":"A command producing very large output no longer fails the whole ExecV3 call; each captured stream keeps up to 8 MiB and says so when there was more","category":"fixed"} +{"description":"A pipe stage stopped because its consumer exited may report a SIGPIPE signal or a non-zero exit carrying the program's own broken-pipe message; both mean the pipe broke, and which one appears is timing","category":"changed"} diff --git a/packages/exec-core/CHANGELOG.md b/packages/exec-core/CHANGELOG.md index 8b4660d8..ea419813 100644 --- a/packages/exec-core/CHANGELOG.md +++ b/packages/exec-core/CHANGELOG.md @@ -19,6 +19,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- A producer stopped by a broken pipe reports either a SIGPIPE signal or a non-zero exit, decided by timing rather than by anything a caller controls; treat both as the same outcome - IExecutor now requires a runPipeline method, so an existing implementation of the interface must add one ### Removed diff --git a/packages/exec-core/changes.jsonl b/packages/exec-core/changes.jsonl index 38e145af..c0f08093 100644 --- a/packages/exec-core/changes.jsonl +++ b/packages/exec-core/changes.jsonl @@ -12,3 +12,4 @@ {"description":"A working directory that exists but is not a directory now reports 126 like a missing one, instead of failing the whole call with a raw spawn error","category":"fixed"} {"description":"A pipeline that cannot start one of its stages no longer leaves the stages already started running unwatched","category":"fixed"} {"description":"Added drainToString, which collects a stream to a string while keeping at most a given number of bytes and reporting how many there were","category":"added"} +{"description":"A producer stopped by a broken pipe reports either a SIGPIPE signal or a non-zero exit, decided by timing rather than by anything a caller controls; treat both as the same outcome","category":"changed"} From 9a69b60949a91920ac5f6bf00f5b90e0c0162ed2 Mon Sep 17 00:00:00 2001 From: Stephen Hellicar Date: Mon, 3 Aug 2026 23:32:52 +1000 Subject: [PATCH 09/11] Sort the import that the rebase conflict resolution reintroduced --- packages/claude-sdk-tools/src/fs/NodeFileSystem.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/packages/claude-sdk-tools/src/fs/NodeFileSystem.ts b/packages/claude-sdk-tools/src/fs/NodeFileSystem.ts index 515993ea..77547b6f 100644 --- a/packages/claude-sdk-tools/src/fs/NodeFileSystem.ts +++ b/packages/claude-sdk-tools/src/fs/NodeFileSystem.ts @@ -1,5 +1,5 @@ import type { Stats } from 'node:fs'; -import { createWriteStream, existsSync, lstatSync as fsLstatSync, openSync, readlinkSync as fsReadlinkSync, realpathSync as fsRealpathSync } from 'node:fs'; +import { createWriteStream, existsSync, lstatSync as fsLstatSync, readlinkSync as fsReadlinkSync, realpathSync as fsRealpathSync, openSync } from 'node:fs'; import { appendFile, lstat as fsLstat, readdir as fsReaddir, readlink as fsReadlink, realpath as fsRealpath, rename as fsRename, stat as fsStat, mkdir, readFile, rm, rmdir, writeFile } from 'node:fs/promises'; import { homedir as osHomedir, tmpdir as osTmpdir } from 'node:os'; import { dirname } from 'node:path'; From 320c80ad484cda5ca72e4ab6bed6b581fed4ba34 Mon Sep 17 00:00:00 2001 From: Stephen Hellicar Date: Mon, 3 Aug 2026 23:42:22 +1000 Subject: [PATCH 10/11] Compare where two redirect paths land, not how they were written Two spellings of one path, or two links to one file, both end up as a single file with two streams opened on it at offset zero, so one overwrites the other. Canonicalising answers where a path lands, and it answers for a target that does not exist yet, which a redirect target usually does not. It takes the working directory rather than being handed an already-resolved path because expansion has to come first: resolve by hand and a leading ~ becomes a directory name that no later expansion can undo. --- packages/claude-core/CHANGELOG.md | 1 + packages/claude-core/changes.jsonl | 1 + .../claude-core/src/fs/canonicalisePath.ts | 9 +++- .../claude-core/test/canonicalisePath.spec.ts | 30 +++++++++++ .../src/ExecV3/runPipeline.ts | 22 +++++--- .../test/integration/redirect-symlink.spec.ts | 52 +++++++++++++++++++ 6 files changed, 106 insertions(+), 9 deletions(-) create mode 100644 packages/claude-sdk-tools/test/integration/redirect-symlink.spec.ts diff --git a/packages/claude-core/CHANGELOG.md b/packages/claude-core/CHANGELOG.md index bf241405..1e5cd4bf 100644 --- a/packages/claude-core/CHANGELOG.md +++ b/packages/claude-core/CHANGELOG.md @@ -30,6 +30,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed - Adopt core-di-lite property injection: config loading splits into a pure read, a holder, and a watch handle with no load or start step, and the shared provider and contract abstractions live here for every package to resolve against +- canonicalisePath takes an optional working directory, for a caller whose relative paths belong to somewhere other than the filesystem's own; resolving by hand first would strip the chance to expand ~ and $VAR - Depend on @shellicar/core-di instead of @shellicar/core-di-lite - File discovery returns records carrying type, size, and symlink target instead of bare path strings - IFileSystem gained openWriteStream, which opens the file before returning so an unwritable path fails at the caller rather than later on the stream; an existing implementation must add it diff --git a/packages/claude-core/changes.jsonl b/packages/claude-core/changes.jsonl index 73054d61..e4df007d 100644 --- a/packages/claude-core/changes.jsonl +++ b/packages/claude-core/changes.jsonl @@ -29,3 +29,4 @@ {"description":"StatResult now carries uid and mode, so a caller can tell who owns a path and who else can reach it","category":"added"} {"description":"canonicalisePath resolves a path to where it actually lands, following symlinks even when the target does not exist yet, for callers that must decide on the destination rather than the string they were handed","category":"added"} {"description":"IFileSystem gained openWriteStream, which opens the file before returning so an unwritable path fails at the caller rather than later on the stream; an existing implementation must add it","category":"changed"} +{"description":"canonicalisePath takes an optional working directory, for a caller whose relative paths belong to somewhere other than the filesystem's own; resolving by hand first would strip the chance to expand ~ and $VAR","category":"changed"} diff --git a/packages/claude-core/src/fs/canonicalisePath.ts b/packages/claude-core/src/fs/canonicalisePath.ts index dae0b214..d4d0be6c 100644 --- a/packages/claude-core/src/fs/canonicalisePath.ts +++ b/packages/claude-core/src/fs/canonicalisePath.ts @@ -14,9 +14,14 @@ const MAX_DANGLING_HOPS = 32; * it was written. Throws when the path cannot be canonicalised at all, carrying the OS's own reason: * a caller that needs a verdict rather than a path decides what to make of that, and a caller that * needs a path is better told than handed something that only looks like one. + * + * `cwd` is the directory a relative path is relative *to*, and defaults to the filesystem's own. + * A caller whose paths belong to somewhere else passes it rather than resolving first: expansion + * has to happen before resolution, so `~/x` resolved by hand becomes a literal `~` directory that + * no later expansion can undo. */ -export function canonicalisePath(value: string, fs: IFileSystem): string { - const absolute = path.resolve(fs.cwd(), expandPath(value, fs)); +export function canonicalisePath(value: string, fs: IFileSystem, cwd: string = fs.cwd()): string { + const absolute = path.resolve(cwd, expandPath(value, fs)); return resolve(absolute, fs, MAX_DANGLING_HOPS); } diff --git a/packages/claude-core/test/canonicalisePath.spec.ts b/packages/claude-core/test/canonicalisePath.spec.ts index 5fc46bac..17c1371b 100644 --- a/packages/claude-core/test/canonicalisePath.spec.ts +++ b/packages/claude-core/test/canonicalisePath.spec.ts @@ -98,3 +98,33 @@ describe('canonicalisePath', () => { expect(actual).toBe(expected); }); }); + +// A caller whose paths are relative to somewhere other than the filesystem's own working directory +// passes that directory rather than resolving first, because expansion has to happen before +// resolution: `~/x` resolved by hand becomes a literal `~` component nothing can undo afterwards. +describe('canonicalisePath with a caller-supplied working directory', () => { + it('resolves a relative path against the directory it was given', () => { + const expected = '/private/var/folders/xk/T/claude-501/conversation/scratchpad/existing.txt'; + const actual = canonicalisePath('existing.txt', fsWith(), WORKSPACE); + expect(actual).toBe(expected); + }); + + it('leaves an absolute path alone', () => { + const expected = '/project/src/file.ts'; + const actual = canonicalisePath('/project/src/file.ts', fsWith(), WORKSPACE); + expect(actual).toBe(expected); + }); + + it('expands the home directory before resolving, so it is never treated as a directory name', () => { + const home = fsWith().homedir(); + const expected = canonicalisePath(`${home}/notes.txt`, fsWith()); + const actual = canonicalisePath('~/notes.txt', fsWith(), WORKSPACE); + expect(actual).toBe(expected); + }); + + it('still defaults to the filesystem working directory when none is given', () => { + const expected = '/project/src/file.ts'; + const actual = canonicalisePath('src/file.ts', fsWith()); + expect(actual).toBe(expected); + }); +}); diff --git a/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts b/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts index 34298e2b..aad95250 100644 --- a/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts +++ b/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts @@ -1,5 +1,6 @@ import { resolve } from 'node:path'; import { PassThrough, Readable, type Writable } from 'node:stream'; +import { canonicalisePath } from '@shellicar/claude-core/fs/canonicalisePath'; import type { IFileSystem } from '@shellicar/claude-core/fs/interfaces'; import { type DrainedStream, drainToString, type PipelineStage } from '@shellicar/exec-core'; import type { EngineContext } from './engine'; @@ -33,15 +34,22 @@ interface StageSinks { function resolveStageSinks(cmd: Command, isLast: boolean, cwd: string, fs: IFileSystem): StageSinks { const redirect = cmd.redirect; const mergeStderr = redirect?.stderr === '&1'; - const stdoutTarget = redirect?.stdout != null ? resolve(cwd, redirect.stdout) : undefined; - const stderrTarget = !mergeStderr && redirect?.stderr != null ? resolve(cwd, redirect.stderr) : undefined; + const stdoutPath = redirect?.stdout; + const stderrPath = mergeStderr ? undefined : redirect?.stderr; + const stdoutTarget = stdoutPath != null ? resolve(cwd, stdoutPath) : undefined; + const stderrTarget = stderrPath != null ? resolve(cwd, stderrPath) : undefined; // Validation (R5) already refuses the same path written twice, but it runs before any cwd is - // known, so all it can compare is the two strings. Resolved against this command's own cwd, - // two spellings of one file become the same path — and two streams on one file each open at - // offset zero, so one silently overwrites the other. - if (stdoutTarget != null && stdoutTarget === stderrTarget) { - throw new Error(`stdout and stderr both resolve to ${stdoutTarget}; use stderr: "&1" to merge them`); + // known, so all it can compare is the two strings. What matters is where each path lands: two + // spellings, or two symlinks, can name one file, and two streams on one file each open at + // offset zero, so one silently overwrites the other. Canonicalising answers that, and it works + // on a target that does not exist yet, which a redirect target usually does not. + if (stdoutPath != null && stderrPath != null) { + const stdoutFile = canonicalisePath(stdoutPath, fs, cwd); + const stderrFile = canonicalisePath(stderrPath, fs, cwd); + if (stdoutFile === stderrFile) { + throw new Error(`stdout and stderr both resolve to ${stdoutFile}; use stderr: "&1" to merge them`); + } } let stdout: Writable | undefined; diff --git a/packages/claude-sdk-tools/test/integration/redirect-symlink.spec.ts b/packages/claude-sdk-tools/test/integration/redirect-symlink.spec.ts new file mode 100644 index 00000000..7fc03a48 --- /dev/null +++ b/packages/claude-sdk-tools/test/integration/redirect-symlink.spec.ts @@ -0,0 +1,52 @@ +import { mkdtempSync, rmSync, symlinkSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { ExecV3 } from '../../src/entry/ExecV3'; +import { call } from '../helpers'; + +// Two paths that name one file only once a link is followed. Nothing about how they were written +// gives it away, so this needs a real symlink on a real filesystem: the memory fake resolves every +// path to itself, which would make the test pass without proving anything. + +describe('stdout and stderr pointed at one file through a symlink', () => { + let dir: string; + let real: string; + let link: string; + + beforeEach(() => { + dir = mkdtempSync(join(tmpdir(), 'execv3-symlink-')); + real = join(dir, 'real.log'); + link = join(dir, 'link.log'); + writeFileSync(real, ''); + symlinkSync(real, link); + }); + + afterEach(() => { + rmSync(dir, { recursive: true, force: true }); + }); + + it('is refused rather than losing one of the streams', async () => { + const result = await call(ExecV3, { intent: 'redirect stdout at a file and stderr at a link to it', commands: [{ program: 'echo', args: ['x'], redirect: { stdout: real, stderr: link } }] }); + + const expected = false; + const actual = result.success; + expect(actual).toBe(expected); + }); + + it('says both resolve to the same file', async () => { + const result = await call(ExecV3, { intent: 'redirect stdout at a file and stderr at a link to it', commands: [{ program: 'echo', args: ['x'], redirect: { stdout: real, stderr: link } }] }); + + const expected = true; + const actual = result.results[0]?.stderr.includes('both resolve to') ?? false; + expect(actual).toBe(expected); + }); + + it('still allows two genuinely different files in the same directory', async () => { + const result = await call(ExecV3, { intent: 'redirect the two streams at two different files', commands: [{ program: 'echo', args: ['x'], redirect: { stdout: join(dir, 'out.log'), stderr: join(dir, 'err.log') } }] }); + + const expected = true; + const actual = result.success; + expect(actual).toBe(expected); + }); +}); From 292f9fd9f48a4b6af9088aba6afff1369726c71c Mon Sep 17 00:00:00 2001 From: Stephen Hellicar Date: Mon, 3 Aug 2026 23:55:31 +1000 Subject: [PATCH 11/11] Say what the redirect rule is instead of numbering it --- packages/claude-sdk-tools/src/ExecV3/runPipeline.ts | 4 ++-- packages/claude-sdk-tools/src/ExecV3/schema.ts | 2 +- packages/claude-sdk-tools/test/ExecV3/scenarios.spec.ts | 2 +- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts b/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts index aad95250..1501ee24 100644 --- a/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts +++ b/packages/claude-sdk-tools/src/ExecV3/runPipeline.ts @@ -39,8 +39,8 @@ function resolveStageSinks(cmd: Command, isLast: boolean, cwd: string, fs: IFile const stdoutTarget = stdoutPath != null ? resolve(cwd, stdoutPath) : undefined; const stderrTarget = stderrPath != null ? resolve(cwd, stderrPath) : undefined; - // Validation (R5) already refuses the same path written twice, but it runs before any cwd is - // known, so all it can compare is the two strings. What matters is where each path lands: two + // Validation already refuses the same path written twice, but it runs before any cwd is known, + // so all it can compare is the two strings. What matters is where each path lands: two // spellings, or two symlinks, can name one file, and two streams on one file each open at // offset zero, so one silently overwrites the other. Canonicalising answers that, and it works // on a target that does not exist yet, which a redirect target usually does not. diff --git a/packages/claude-sdk-tools/src/ExecV3/schema.ts b/packages/claude-sdk-tools/src/ExecV3/schema.ts index 531834fd..6330b0ec 100644 --- a/packages/claude-sdk-tools/src/ExecV3/schema.ts +++ b/packages/claude-sdk-tools/src/ExecV3/schema.ts @@ -122,7 +122,7 @@ export const ExecV3InputSchema = z }); } - // R5: stdout and stderr aimed at the same file. Each becomes its own stream opened at + // stdout and stderr aimed at the same file. Each becomes its own stream opened at // offset zero, so one overwrites the other and half the output is lost while the command // still reports success. "&1" is the way to say what this was trying to say. if (cmd.redirect?.stdout != null && cmd.redirect.stderr != null && cmd.redirect.stderr !== '&1' && cmd.redirect.stdout === cmd.redirect.stderr) { diff --git a/packages/claude-sdk-tools/test/ExecV3/scenarios.spec.ts b/packages/claude-sdk-tools/test/ExecV3/scenarios.spec.ts index a10ad117..9d3d042a 100644 --- a/packages/claude-sdk-tools/test/ExecV3/scenarios.spec.ts +++ b/packages/claude-sdk-tools/test/ExecV3/scenarios.spec.ts @@ -650,7 +650,7 @@ describe('validation — pipe with stdout redirect (R4)', () => { }); }); -describe('validation — stdout and stderr to the same file (R5)', () => { +describe('validation — stdout and stderr to the same file', () => { it('rejects two redirects aimed at one file', () => { const expected = false; const actual = ExecV3InputSchema.safeParse({ intent: 'x', commands: [{ program: 'echo', redirect: { stdout: '/tmp/a.log', stderr: '/tmp/a.log' } }] }).success;