diff --git a/packages/core/src/process.ts b/packages/core/src/process.ts index 8d13d79433..a9bfb85fc3 100644 --- a/packages/core/src/process.ts +++ b/packages/core/src/process.ts @@ -9,6 +9,14 @@ export class AppProcessError extends Schema.TaggedErrorClass()( command: Schema.String, exitCode: Schema.optional(Schema.Number), stderr: Schema.optional(Schema.String), + /** + * Whatever the child had written to stdout when this failed. + * + * Present so a TIMEOUT says something. Without it the only report was "Timed out", which cannot tell a + * child that hung before producing a byte from one that did most of its work and then stalled -- and on + * a flake that only reproduces on one platform in CI, that distinction is the whole investigation. + */ + stdout: Schema.optional(Schema.String), cause: Schema.optional(Schema.Defect()), }) { override get message() { @@ -118,10 +126,30 @@ const normalizeStdin = ( ? Stream.make(input) : input -export const collectStream = (stream: Stream.Stream, maxOutputBytes: number | undefined) => +/** + * What `collectStream` gathers. Exposed so a CALLER can own it. + * + * The fold mutates one object and hands the same one back on every chunk, which means a caller holding a + * reference sees whatever arrived even if the fold never completes. That is the whole point: a command + * killed by its timeout used to report nothing at all about the child, because the buffers lived inside + * the interrupted fiber and died with it. A subprocess that fails by producing NOTHING and one that fails + * after printing half its work are very different diagnoses, and both looked identical. + */ +export type StreamAccumulator = { chunks: Uint8Array[]; bytes: number; truncated: boolean } + +export const newAccumulator = (): StreamAccumulator => ({ chunks: [], bytes: 0, truncated: false }) + +/** Whatever the accumulator holds right now, as a buffer. Safe to call mid-stream. */ +export const accumulated = (acc: StreamAccumulator) => Buffer.concat(acc.chunks) + +export const collectStream = ( + stream: Stream.Stream, + maxOutputBytes: number | undefined, + into: StreamAccumulator = newAccumulator(), +) => Stream.runFold( stream, - () => ({ chunks: [] as Uint8Array[], bytes: 0, truncated: false }), + () => into, (acc, chunk) => { if (maxOutputBytes === undefined) { acc.chunks.push(chunk) @@ -143,12 +171,19 @@ const layer = Layer.effect( const runCommand = (command: ChildProcess.Command, options?: RunOptions) => { const description = describeCommand(command) + /* + Owned HERE, outside `collect`, so a timeout can still read them. `Effect.timeoutOrElse` interrupts + `collect`, and anything scoped to that fiber goes with it -- which is why a timed-out command used + to report neither stream. + */ + const outAcc = newAccumulator() + const errAcc = newAccumulator() const collect = Effect.scoped( Effect.gen(function* () { const handle = yield* spawner.spawn(command) if (options?.combineOutput) { const [output, exitCode] = yield* Effect.all( - [collectStream(handle.all, options.maxOutputBytes), handle.exitCode], + [collectStream(handle.all, options.maxOutputBytes, outAcc), handle.exitCode], { concurrency: "unbounded" }, ) return { @@ -164,8 +199,8 @@ const layer = Layer.effect( } const [stdout, stderr, exitCode] = yield* Effect.all( [ - collectStream(handle.stdout, options?.maxOutputBytes), - collectStream(handle.stderr, options?.maxErrorBytes), + collectStream(handle.stdout, options?.maxOutputBytes, outAcc), + collectStream(handle.stderr, options?.maxErrorBytes, errAcc), handle.exitCode, ], { concurrency: "unbounded" }, @@ -183,7 +218,19 @@ const layer = Layer.effect( const timed = options?.timeout ? Effect.timeoutOrElse(collect, { duration: options.timeout, - orElse: () => Effect.fail(new AppProcessError({ command: description, cause: new Error("Timed out") })), + orElse: () => + Effect.fail( + new AppProcessError({ + command: description, + /* + Partial output, not nothing. An empty stdout here is now a real observation about the + child rather than an artifact of how the failure was built. + */ + stdout: accumulated(outAcc).toString(), + stderr: accumulated(errAcc).toString(), + cause: new Error("Timed out"), + }), + ), }) : collect const aborted = options?.signal diff --git a/packages/core/test/process/process.test.ts b/packages/core/test/process/process.test.ts index b213ab1559..afa278f49a 100644 --- a/packages/core/test/process/process.test.ts +++ b/packages/core/test/process/process.test.ts @@ -3,7 +3,7 @@ import fs from "fs/promises" import { realpathSync } from "node:fs" import { tmpdir } from "node:os" import path from "node:path" -import { Effect, Exit, Fiber, Stream } from "effect" +import { Cause, Effect, Exit, Fiber, Stream } from "effect" import { ChildProcess } from "effect/unstable/process" import { LayerNode } from "@redrob-code/core/effect/layer-node" import { AppProcess } from "@redrob-code/core/process" @@ -152,6 +152,31 @@ describe("AppProcess", () => { ) if (process.platform !== "win32") { + /* + The claim this pins: a command killed by its timeout reports what the child had ALREADY printed. + + It used to report nothing on either stream, because the accumulators lived inside the fiber + `Effect.timeoutOrElse` interrupts. That turned every subprocess timeout into the same message + regardless of cause -- a child that hung before writing a byte and one that did most of its work + and then stalled were indistinguishable, which is exactly the distinction an intermittent + platform-specific hang turns on. + */ + it.live( + "a timeout reports the output the child produced before it was killed", + Effect.gen(function* () { + const svc = yield* AppProcess.Service + /* Prints, flushes, then hangs -- so there IS partial output to lose. */ + const script = `process.stdout.write('half-done\\n');process.stderr.write('warned\\n');setInterval(()=>{},60000)` + const exit = yield* Effect.exit(svc.run(cmd("-e", script), { timeout: "700 millis" })) + expect(Exit.isFailure(exit)).toBe(true) + if (!Exit.isFailure(exit)) return + const error = Cause.squash(exit.cause) as { stdout?: string; stderr?: string } + expect(error.stdout).toContain("half-done") + expect(error.stderr).toContain("warned") + }), + 8_000, + ) + it.live( "timeout cleans up the scoped child process", Effect.acquireUseRelease( diff --git a/packages/redrob/test/lib/cli-process.ts b/packages/redrob/test/lib/cli-process.ts index eee5542c3c..e9d141e2d7 100644 --- a/packages/redrob/test/lib/cli-process.ts +++ b/packages/redrob/test/lib/cli-process.ts @@ -42,6 +42,27 @@ const cliEntry = path.join(redrobRoot, "src/index.ts") // whole test and would hold permits while a nested `run` waits for one. const spawnGate = Semaphore.makeUnsafe(Math.max(2, availableParallelism() - 1)) +/** + * The harness's wait bounds, derived from one number instead of written three times. + * + * They are not independent: `cliIt.concurrent` carried a comment saying Bun's test timeout must stay + * ABOVE the child timeout, while the two numbers sat in different functions as unrelated literals with + * nothing keeping them in step. Raising one without the other silently breaks the invariant the comment + * promises, and the failure mode is a test that expires holding no spawn permit -- which reads as the + * command being slow rather than as the harness mis-tuned. + * + * Windows gets a multiple of the base. A `redrob.spawn` is `bun run` over the TypeScript entry, so every + * spawn pays a cold transpile plus that platform's process-creation cost, and the runners are slower + * again. This is a MITIGATION and not a diagnosis: the flake it responds to timed out at exactly the + * bound with no output preserved, which is now fixed separately -- the next occurrence will say where the + * child actually got to, and this number should be revisited against that evidence rather than raised + * again by feel. + */ +const SLOW_PLATFORM_FACTOR = process.platform === "win32" ? 3 : 1 +const CHILD_TIMEOUT_MS = 30_000 * SLOW_PLATFORM_FACTOR +const READY_TIMEOUT_MS = 15_000 * SLOW_PLATFORM_FACTOR +const TEST_TIMEOUT_MS = CHILD_TIMEOUT_MS * 2 + export const testModelID = "test/test-model" // Wrap a Bun subprocess pipe (or any ReadableStream) as a Stream. @@ -219,7 +240,7 @@ export function withCliFixture( return yield* spawnGate.withPermit( Effect.gen(function* () { const start = Date.now() - const timeoutMs = opts?.timeoutMs ?? 30_000 + const timeoutMs = opts?.timeoutMs ?? CHILD_TIMEOUT_MS // stdin: "ignore" so the child doesn't see a piped stdin and block // on `Bun.stdin.text()` (see src/cli/cmd/run.ts — non-TTY stdin is // consumed as the prompt). The old Process.run wrapper defaulted to @@ -246,7 +267,13 @@ export function withCliFixture( Effect.succeed({ command: err.command, exitCode: err.exitCode ?? -1, - stdout: Buffer.alloc(0), + /* + The child's own output, not an empty buffer. This used to be `Buffer.alloc(0)` + unconditionally, so `expectExit`'s dump printed an empty stdout on every timeout and a + reader could not tell whether the child had produced nothing or the harness had thrown it + away. It was the latter, which cost a real investigation. + */ + stdout: Buffer.from(err.stdout ?? ""), stderr: Buffer.from((err.stderr ?? String(err.cause ?? err.message)) + "\n"), stdoutTruncated: false, stderrTruncated: false, @@ -375,7 +402,7 @@ export function withCliFixture( ), ) - const readyTimeoutMs = opts?.readyTimeoutMs ?? 15_000 + const readyTimeoutMs = opts?.readyTimeoutMs ?? READY_TIMEOUT_MS const match = yield* Deferred.await(readyDeferred).pipe( Effect.timeoutOrElse({ duration: Duration.millis(readyTimeoutMs), @@ -545,8 +572,9 @@ export const cliIt = { (process.platform === "win32" ? test : test.concurrent)( name, () => Effect.runPromise(Effect.scoped(withCliFixture(body))), - // Bun's timeout includes spawn-gate wait. Keep it above the child timeout so - // queued concurrent CLI tests do not expire while holding no permit. - opts ?? 60_000, + // Bun's timeout includes spawn-gate wait, so it must stay above the child timeout or a queued + // concurrent CLI test expires while holding no permit. Derived from it rather than restated, which + // is what let the two drift apart. + opts ?? TEST_TIMEOUT_MS, ), }