From bf328d1d0fe25d9edb8a89f7b4b44c03fde05919 Mon Sep 17 00:00:00 2001 From: willbot Date: Mon, 2 Nov 2026 17:02:48 +0100 Subject: [PATCH 1/6] feat(engine): ctx.spawn can hand the command its child's output, so Composer can redact secrets before they reach the terminal `ctx.spawn({ command, args, onOutput: (chunk, stream) => ... })` delivers the child's stdout and stderr to the handler as UTF-8 text instead of the terminal or the diagnostic stream. The engine still owns the child: stdin stays inherited, signals are recorded and forwarded as before, an abort runs the SIGTERM-then-SIGKILL ladder, and the exit status settles the run. ctx.spawn resolves after the last chunk. A throwing onOutput ends the child and ctx.spawn rejects with that error. The spawn adapter seam gains `output: "pipe"` on SpawnRequest and `stdout`/`stderr` on SpawnedChild. The shipped cross-spawn adapter and the test harness implement it. The engine ends a child whose adapter did not return both streams. @prisma/cli-engine goes from 0.9.0 to 0.10.0. Conformance allows composer-cli and orm-toolchain to peer 0.9.0 until they release against 0.10.0. Signed-off-by: willbot Signed-off-by: Will Madden --- packages/cli-engine/README.md | 21 ++ packages/cli-engine/package.json | 2 +- packages/cli-engine/src/context.ts | 4 +- packages/cli-engine/src/execution/spawn.ts | 87 +++++- packages/cli-engine/src/exports/index.ts | 1 + packages/cli-engine/src/spawn.ts | 29 +- packages/cli-engine/src/testing.ts | 98 ++++++- packages/cli-engine/tests/fixtures/child.mjs | 16 ++ .../cli-engine/tests/spawn-real-child.test.ts | 112 +++++++- packages/cli-engine/tests/spawn.test.ts | 266 ++++++++++++++++++ packages/cli/package.json | 2 +- packages/cli/scripts/conformance.ts | 19 +- packages/cli/src/spawn.ts | 35 ++- packages/cli/tests/spawn-adapter.test.ts | 68 +++++ packages/prisma/package.json | 2 +- pnpm-lock.yaml | 4 +- 16 files changed, 733 insertions(+), 33 deletions(-) diff --git a/packages/cli-engine/README.md b/packages/cli-engine/README.md index 419a0d4a..2af35c8f 100644 --- a/packages/cli-engine/README.md +++ b/packages/cli-engine/README.md @@ -75,4 +75,25 @@ A statement is the command's own declared input, unlike a `--confirm` value, whi Each flag answers one question. A run that succeeds with a flag nothing consumed fails with `CLI.CONSENT_UNUSED`, so a mistyped subject or a second answer to one question is never ignored. The same holds for a `--confirm` token no consent asked for, on a command that declares statements (other commands still ignore a leftover `--confirm`; see [#339](https://github.com/prisma/prisma-cli/issues/339)). A refusal also names any given statement flag that matches none of its questions, and any `--confirm` token, so the user hears of it before answering a prompt. That check runs when the handler returns, after the command has acted. A command that asks everything in one batch passes `statements(questions, { last: true })` to get the same failure right after the questions are answered, before it does anything with them. +## Reading a child's output + +A command that declares `maySpawn` runs a child process with `ctx.spawn`. By default the child writes straight to the terminal, or, with `--json`, to stderr. To read the output instead, pass `onOutput`: + +```ts +const result = await ctx.spawn({ + command: "alchemy", + args: ["deploy"], + onOutput: (chunk, stream) => { + redactor.write(stream, chunk); + }, +}); +redactor.flush(); +``` + +`chunk` is text decoded as UTF-8, and `stream` is `"stdout"` or `"stderr"`. Chunks from one stream arrive in the order the child wrote them; there is no guaranteed order between the two streams. `ctx.spawn` resolves after the last chunk, so a command that holds text back between chunks writes the rest after the `await`. In json mode stdout carries the engine's frames, so write both streams to stderr. + +The engine still owns the child. Stdin stays inherited, Ctrl-C reaches the child as before, and a SIGTERM or an abort of `ctx.signal` ends it with SIGTERM, then SIGKILL after a grace period. If `onOutput` throws, the engine stops delivering, ends the child the same way, and `ctx.spawn` rejects with that error once the child has exited. + +A spawn adapter supports this by starting the child with stdout and stderr piped when the request's `output` is `"pipe"`, and returning both as `stdout` and `stderr` on the child. The engine ends a child whose adapter did not return both. + Part of [prisma/prisma-cli](https://github.com/prisma/prisma-cli). diff --git a/packages/cli-engine/package.json b/packages/cli-engine/package.json index 02033abf..428fad4b 100644 --- a/packages/cli-engine/package.json +++ b/packages/cli-engine/package.json @@ -1,6 +1,6 @@ { "name": "@prisma/cli-engine", - "version": "0.9.0", + "version": "0.10.0", "description": "The execution engine of the unified Prisma CLI.", "type": "module", "exports": { diff --git a/packages/cli-engine/src/context.ts b/packages/cli-engine/src/context.ts index 357ef066..d5516fc4 100644 --- a/packages/cli-engine/src/context.ts +++ b/packages/cli-engine/src/context.ts @@ -65,7 +65,9 @@ export interface CommandContext< /** * Hands the terminal to a child process and resolves when it ends. * The child inherits stdio and runs in this process's own group - * (POSIX) or console (Windows), so Ctrl-C reaches it natively. + * (POSIX) or console (Windows), so Ctrl-C reaches it natively. With + * `onOutput`, stdin is still inherited but stdout and stderr are + * delivered to the handler instead; everything below is unchanged. * * While a child is live the engine neither aborts nor exits on a * delivered signal: it records signals and replays them into its diff --git a/packages/cli-engine/src/execution/spawn.ts b/packages/cli-engine/src/execution/spawn.ts index 723d9461..ce1dca40 100644 --- a/packages/cli-engine/src/execution/spawn.ts +++ b/packages/cli-engine/src/execution/spawn.ts @@ -7,6 +7,7 @@ import { import type { EngineEvent } from "../events"; import { CliStructuredError } from "../protocol"; import type { + ChildOutputStream, ChildResult, SpawnChild, SpawnedChild, @@ -138,12 +139,18 @@ async function runDelegated( args: options.args ?? [], cwd: options.cwd ?? invocation.runtime.cwd, env: await composeChildEnv(invocation, def, options.env), - output: state.format === "json" ? "diagnostic" : "inherit", + output: childOutput(options, state.format === "json"), }; debug( `spawn: ${request.command} ${request.args.join(" ")} (cwd ${request.cwd})`, ); - const result = await runChild(invocation, terminal, spawnChild, request); + const result = await runChild( + invocation, + terminal, + spawnChild, + request, + options.onOutput, + ); debug( `spawn ended: exitCode=${String(result.exitCode)} signal=${String(result.signal)} recorded=[${terminal.recorded.join(",")}]`, ); @@ -161,6 +168,16 @@ async function runDelegated( } } +function childOutput( + options: SpawnOptions, + structured: boolean, +): SpawnRequest["output"] { + if (options.onOutput !== undefined) { + return "pipe"; + } + return structured ? "diagnostic" : "inherit"; +} + /** A handler that resolved or threw with its child still live: the * engine ends the child on the abort ladder and waits for it, because * the engine outliving the child is not negotiable. */ @@ -176,6 +193,7 @@ async function runChild( terminal: DelegatedTerminal, spawnChild: SpawnChild, request: SpawnRequest, + onOutput: SpawnOptions["onOutput"], ): Promise { let child: SpawnedChild; try { @@ -190,14 +208,73 @@ async function runChild( const ended = new AbortController(); armTerminationLadder(invocation, terminal, child, ended.signal); try { - return recordCompletedChild(invocation, await child.ended); - } catch (cause) { - throw spawnFailedError(request.command, cause); + const delivery = + onOutput === undefined + ? undefined + : deliverOutput(terminal, child, onOutput); + let result: ChildResult; + try { + result = recordCompletedChild(invocation, await child.ended); + } catch (cause) { + throw spawnFailedError(request.command, cause); + } + const failure = await delivery; + if (failure !== undefined) { + throw failure.cause; + } + return result; } finally { ended.abort(); } } +/** Reads both piped streams into the handler's callback. A throwing + * callback ends delivery and terminates the child on the abort ladder; + * the failure is returned so ctx.spawn rejects with it once the child + * has ended. A stream that fails on its own (the adapter destroying a + * pipe a grandchild held open) just ends. */ +async function deliverOutput( + terminal: DelegatedTerminal, + child: SpawnedChild, + onOutput: NonNullable, +): Promise<{ readonly cause: unknown } | undefined> { + if (child.stdout === undefined || child.stderr === undefined) { + terminal.termination.abort(); + return { + cause: new Error( + "@prisma/cli-engine: ctx.spawn was given onOutput but the Runtime's spawn adapter returned no piped stdout and stderr", + ), + }; + } + let failure: { readonly cause: unknown } | undefined; + const read = async ( + source: AsyncIterable, + stream: ChildOutputStream, + ): Promise => { + try { + for await (const chunk of source) { + if (failure !== undefined) { + return; + } + try { + onOutput(chunk, stream); + } catch (cause) { + failure = { cause }; + terminal.termination.abort(); + return; + } + } + } catch { + return; + } + }; + await Promise.all([ + read(child.stdout, "stdout"), + read(child.stderr, "stderr"), + ]); + return failure; +} + /** The engine mints the result the handler sees and keeps it on the * run: the settlement then reads the child's status from the engine's * own record rather than from anything the handler hands back, and a diff --git a/packages/cli-engine/src/exports/index.ts b/packages/cli-engine/src/exports/index.ts index 8dc5b13c..79886b78 100644 --- a/packages/cli-engine/src/exports/index.ts +++ b/packages/cli-engine/src/exports/index.ts @@ -149,6 +149,7 @@ export { type Runtime, } from "../runtime"; export { + type ChildOutputStream, type ChildResult, type ChildStatusSettlement, type ExitWithChildStatusOptions, diff --git a/packages/cli-engine/src/spawn.ts b/packages/cli-engine/src/spawn.ts index bd36f396..9178a875 100644 --- a/packages/cli-engine/src/spawn.ts +++ b/packages/cli-engine/src/spawn.ts @@ -19,10 +19,16 @@ export interface SpawnRequest { * NDJSON on stdout in that mode and cannot detect an adapter that * ignores this field — inherited child stdout would silently corrupt * the stream. Adapters written before this field existed must be - * updated. */ - readonly output: "inherit" | "diagnostic"; + * updated. "pipe" keeps stdin inherited and pipes stdout and stderr + * to this process: the adapter MUST return them as + * SpawnedChild.stdout and SpawnedChild.stderr, and the engine kills a + * child returned without them. */ + readonly output: "inherit" | "diagnostic" | "pipe"; } +/** Which of the child's output streams a chunk came from. */ +export type ChildOutputStream = "stdout" | "stderr"; + /** How a child ended. A signal-killed child carries `signal` and a null * `exitCode`, and is an abort rather than a failure: exitWithChildStatus * settles it as one, and a handler that reads the result itself branches @@ -38,6 +44,13 @@ export interface SpawnedChild { * launched at all; the engine phrases that as CLI.SPAWN_FAILED. */ readonly ended: Promise; readonly kill: (signal: "SIGTERM" | "SIGKILL") => void; + /** The child's piped output, decoded as UTF-8. Required when the + * request's `output` is "pipe", absent otherwise. Each ends when + * the child closes it; an adapter ends both no later than a grace + * period after the child exits, so a grandchild holding the pipe + * open cannot hold the run. */ + readonly stdout?: AsyncIterable; + readonly stderr?: AsyncIterable; } /** @@ -58,6 +71,18 @@ export interface SpawnOptions { /** Added to, and overriding, the invocation environment. The engine's * credential variables are applied last and cannot be overridden. */ readonly env?: Readonly>; + /** + * Receives the child's stdout and stderr instead of the terminal or + * the diagnostic stream, in the order each stream produced them, and + * the handler writes them on itself. The engine still owns the child: + * stdin, signals, the abort ladder and the exit status are as without + * it, and ctx.spawn resolves only after the last chunk was delivered. + * In json mode stdout carries the engine's frames, so write both + * streams to stderr. If it throws, the engine stops delivering, + * terminates the child as on abort, and ctx.spawn rejects with that + * error once the child has ended. + */ + readonly onOutput?: (chunk: string, stream: ChildOutputStream) => void; } export const CHILD_STATUS: unique symbol = Symbol.for( diff --git a/packages/cli-engine/src/testing.ts b/packages/cli-engine/src/testing.ts index a90b0cf9..9f4c4f9d 100644 --- a/packages/cli-engine/src/testing.ts +++ b/packages/cli-engine/src/testing.ts @@ -18,20 +18,72 @@ import type { PackageManagerId, PackageManagerRunner } from "./package-manager"; import type { PresentedResult } from "./presentation"; import type { RunSummary } from "./run-summary"; import type { Host, Runtime } from "./runtime"; -import type { ChildResult, SpawnChild, SpawnRequest } from "./spawn"; +import type { + ChildOutputStream, + ChildResult, + SpawnChild, + SpawnRequest, +} from "./spawn"; import type { TelemetryPayload } from "./telemetry/payload"; import type { TelemetryDeclaration } from "./telemetry/report"; /** * What a scripted fake child does. `nextKill` resolves with each signal * the engine delivers, so a script can model a child that ignores - * SIGTERM and only dies on SIGKILL. + * SIGTERM and only dies on SIGKILL. `write` emits output on a piped + * request (`output: "pipe"`) and is ignored otherwise; both streams end + * when the program returns. */ export type ScriptedChildProgram = ( request: SpawnRequest, - child: { readonly nextKill: () => Promise<"SIGTERM" | "SIGKILL"> }, + child: { + readonly nextKill: () => Promise<"SIGTERM" | "SIGKILL">; + readonly write: (stream: ChildOutputStream, chunk: string) => void; + }, ) => ChildResult | Promise; +interface ScriptedStream extends AsyncIterable { + push(chunk: string): void; + end(): void; +} + +function scriptedStream(): ScriptedStream { + const chunks: string[] = []; + let ended = false; + let wake: (() => void) | undefined; + const notify = () => { + const waiting = wake; + wake = undefined; + waiting?.(); + }; + return { + push: (chunk) => { + chunks.push(chunk); + notify(); + }, + end: () => { + ended = true; + notify(); + }, + [Symbol.asyncIterator]: () => { + const next = async (): Promise> => { + const chunk = chunks.shift(); + if (chunk !== undefined) { + return { value: chunk, done: false }; + } + if (ended) { + return { value: undefined, done: true }; + } + await new Promise((resolve) => { + wake = resolve; + }); + return next(); + }; + return { next }; + }, + }; +} + /** One ctx.spawn call, as the harness saw it. */ export interface SpawnRecord { readonly command: string; @@ -53,19 +105,33 @@ function scriptedSpawn(program: ScriptedChildProgram): SpawnChild { return (request) => { const delivered: ("SIGTERM" | "SIGKILL")[] = []; const waiting: Array<(signal: "SIGTERM" | "SIGKILL") => void> = []; + const piped = + request.output === "pipe" + ? { stdout: scriptedStream(), stderr: scriptedStream() } + : undefined; return { - ended: (async () => - program(request, { - nextKill: async () => { - const queued = delivered.shift(); - return ( - queued ?? - (await new Promise<"SIGTERM" | "SIGKILL">((resolve) => { - waiting.push(resolve); - })) - ); - }, - }))(), + ended: (async () => { + try { + return await program(request, { + nextKill: async () => { + const queued = delivered.shift(); + return ( + queued ?? + (await new Promise<"SIGTERM" | "SIGKILL">((resolve) => { + waiting.push(resolve); + })) + ); + }, + write: (stream, chunk) => { + piped?.[stream].push(chunk); + }, + }); + } finally { + piped?.stdout.end(); + piped?.stderr.end(); + } + })(), + ...piped, kill: (signal) => { const waiter = waiting.shift(); if (waiter === undefined) { @@ -99,6 +165,8 @@ function recordingSpawn( record.kills.push(signal); child.kill(signal); }, + stdout: child.stdout, + stderr: child.stderr, }; }; } diff --git a/packages/cli-engine/tests/fixtures/child.mjs b/packages/cli-engine/tests/fixtures/child.mjs index e4680b93..90276971 100644 --- a/packages/cli-engine/tests/fixtures/child.mjs +++ b/packages/cli-engine/tests/fixtures/child.mjs @@ -40,6 +40,22 @@ switch (mode) { idle(); writeFileSync(value, "ready"); break; + case "chatter": { + const pause = () => new Promise((resolve) => setTimeout(resolve, 50)); + process.stdout.write("out-1\n"); + await pause(); + process.stderr.write("err-1\n"); + await pause(); + process.stdout.write("out-2\n"); + await pause(); + process.exitCode = Number(value); + break; + } + case "chatter-ignore-term": + process.on("SIGTERM", () => {}); + setInterval(() => process.stdout.write("tick\n"), 20); + writeFileSync(value, String(process.pid)); + break; default: process.exit(9); } diff --git a/packages/cli-engine/tests/spawn-real-child.test.ts b/packages/cli-engine/tests/spawn-real-child.test.ts index eb0e1301..c3c37766 100644 --- a/packages/cli-engine/tests/spawn-real-child.test.ts +++ b/packages/cli-engine/tests/spawn-real-child.test.ts @@ -21,6 +21,7 @@ import { createTestCli } from "@prisma/cli-engine/testing"; import { describe, expect, test } from "vitest"; const NODE = process.execPath; +const TICKS = /^(tick\n)+$/; const CHILD = fileURLToPath(new URL("./fixtures/child.mjs", import.meta.url)); const HOST = fileURLToPath( new URL("./fixtures/spawn-host.mjs", import.meta.url), @@ -46,6 +47,48 @@ const realSpawn: SpawnChild = (request) => { }; }; +/** The bin adapter's pipe mode: stdin inherited, both output streams + * handed to the engine. */ +const pipingSpawn: SpawnChild = (request) => { + const child = spawn(request.command, [...request.args], { + cwd: request.cwd, + env: request.env, + stdio: ["ignore", "pipe", "pipe"], + }); + child.stdout.setEncoding("utf8"); + child.stderr.setEncoding("utf8"); + return { + ended: new Promise((resolve, reject) => { + child.on("error", reject); + child.on("close", (exitCode, signal) => resolve({ exitCode, signal })); + }), + kill: (signal) => { + child.kill(signal); + }, + stdout: child.stdout, + stderr: child.stderr, + }; +}; + +type Chunk = readonly [stream: "stdout" | "stderr", chunk: string]; + +function pipingCommand(chunks: Chunk[], ...childArgs: string[]) { + return defineCommand({ + help: { summary: "Reads the fixture child's output" }, + maySpawn: true, + handler: async (_args, ctx) => { + await ctx.spawn({ + command: NODE, + args: [CHILD, ...childArgs], + onOutput: (chunk, stream) => { + chunks.push([stream, chunk]); + }, + }); + return ok(exitWithChildStatus()); + }, + }); +} + function childCommand(...childArgs: string[]) { return defineCommand({ help: { summary: "Runs the fixture child" }, @@ -79,7 +122,7 @@ function termAfterReady(adapter: SpawnChild, ready: string): SpawnChild { return (request) => { const child = adapter(request); return { - ended: child.ended, + ...child, kill: (signal) => { if (signal === "SIGTERM") { waitForFile(ready, 2_000) @@ -257,6 +300,73 @@ describe.skipIf(process.platform === "win32")( expect(result.spawns[0].kills).toEqual(["SIGTERM", "SIGKILL"]); expect(result.exitCode).toBe(137); }); + + test("piped output reaches onOutput in order, and the exit code passes through", async () => { + const chunks: Chunk[] = []; + const cli = createTestCli({ + commands: { converge: pipingCommand(chunks, "chatter", "4") }, + spawn: pipingSpawn, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.exitCode).toBe(4); + expect(chunks).toEqual([ + ["stdout", "out-1\n"], + ["stderr", "err-1\n"], + ["stdout", "out-2\n"], + ]); + expect(result.stdout).toBe(""); + }); + + test("a piped child that ignores SIGTERM is killed on abort, leaving no orphan", async () => { + const dir = scratch(); + const pidFile = join(dir, "pid"); + const chunks: Chunk[] = []; + const controller = new AbortController(); + const watcher = defineCommand({ + help: { summary: "Aborts while a piped child runs" }, + maySpawn: true, + handler: async (_args, ctx) => { + ctx.report({ kind: "status", subject: "run", status: "ready" }); + await new Promise((resolve) => setTimeout(resolve, 0)); + await ctx.spawn({ + command: NODE, + args: [CHILD, "chatter-ignore-term", pidFile], + onOutput: (chunk, stream) => { + chunks.push([stream, chunk]); + }, + }); + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: watcher }, + spawn: termAfterReady(pipingSpawn, pidFile), + delay: async () => { + await new Promise((resolve) => setTimeout(resolve, 300)); + }, + }); + + const result = await cli.run(["converge"], { + abort: controller.signal, + onEvent: (event) => { + if (event.kind === "status") { + controller.abort(); + } + }, + }); + + expect(result.spawns[0].kills).toEqual(["SIGTERM", "SIGKILL"]); + expect(result.exitCode).toBe(137); + expect(chunks.length).toBeGreaterThan(0); + expect( + chunks.every( + ([stream, chunk]) => stream === "stdout" && TICKS.test(chunk), + ), + ).toBe(true); + expect(alive(Number(readFileSync(pidFile, "utf8")))).toBe(false); + }); }, 20_000, ); diff --git a/packages/cli-engine/tests/spawn.test.ts b/packages/cli-engine/tests/spawn.test.ts index d11d97de..f0e24f9d 100644 --- a/packages/cli-engine/tests/spawn.test.ts +++ b/packages/cli-engine/tests/spawn.test.ts @@ -1808,3 +1808,269 @@ function controllableRuntime() { }, }; } + +type Chunk = readonly [stream: "stdout" | "stderr", chunk: string]; + +function pipingCommand(chunks: Chunk[]) { + return defineCommand({ + help: { summary: "Reads a child's output" }, + maySpawn: true, + handler: async (_args, ctx) => { + await ctx.spawn({ + command: "alchemy", + args: ["deploy"], + onOutput: (chunk, stream) => { + chunks.push([stream, chunk]); + }, + }); + return ok(exitWithChildStatus()); + }, + }); +} + +describe("piped child output", () => { + test("onOutput receives both streams in the order the child wrote them", async () => { + const chunks: Chunk[] = []; + const cli = createTestCli({ + commands: { converge: pipingCommand(chunks) }, + now: CLOCK, + spawnScript: async (_request, child) => { + child.write("stdout", "one "); + child.write("stdout", "two\n"); + await new Promise((resolve) => setTimeout(resolve, 0)); + child.write("stderr", "warn\n"); + await new Promise((resolve) => setTimeout(resolve, 0)); + child.write("stdout", "three\n"); + return { exitCode: 0, signal: null }; + }, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.exitCode).toBe(0); + expect(result.spawns[0]?.output).toBe("pipe"); + expect(chunks).toEqual([ + ["stdout", "one "], + ["stdout", "two\n"], + ["stderr", "warn\n"], + ["stdout", "three\n"], + ]); + expect(result.stdout).toBe(""); + expect(result.stderr).toBe(""); + }); + + test("json mode pipes too, and keeps stdout framed", async () => { + const chunks: Chunk[] = []; + const cli = createTestCli({ + commands: { converge: pipingCommand(chunks) }, + now: CLOCK, + spawnScript: (_request, child) => { + child.write("stdout", "hello"); + return { exitCode: 0, signal: null }; + }, + }); + + const result = await cli.run(["converge", "--json"]); + + expect(result.spawns[0]?.output).toBe("pipe"); + expect(chunks).toEqual([["stdout", "hello"]]); + expect(result.json).toHaveLength(1); + expect(result.json[0]).toMatchObject({ kind: "result" }); + }); + + test("ctx.spawn resolves only after the last chunk was delivered", async () => { + const seen: string[] = []; + const command = defineCommand({ + help: { summary: "Reads a child's output" }, + maySpawn: true, + handler: async (_args, ctx) => { + await ctx.spawn({ + command: "alchemy", + onOutput: (chunk) => { + seen.push(chunk); + }, + }); + seen.push("resolved"); + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: command }, + now: CLOCK, + spawnScript: (_request, child) => { + for (const chunk of ["a", "b", "c"]) { + child.write("stderr", chunk); + } + return { exitCode: 0, signal: null }; + }, + }); + + await cli.run(["converge"]); + + expect(seen).toEqual(["a", "b", "c", "resolved"]); + }); + + test("the child's exit code still settles the run", async () => { + const cli = createTestCli({ + commands: { converge: pipingCommand([]) }, + now: CLOCK, + spawnScript: (_request, child) => { + child.write("stderr", "failed\n"); + return { exitCode: 3, signal: null }; + }, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.exitCode).toBe(3); + }); + + test("a SIGTERM delivered while piping is forwarded and the child's end settles the run", async () => { + const chunks: Chunk[] = []; + const controller = new AbortController(); + const cli = createTestCli({ + commands: { converge: pipingCommand(chunks) }, + now: CLOCK, + spawnScript: async (_request, child) => { + child.write("stdout", "working\n"); + controller.abort("SIGTERM"); + await child.nextKill(); + return { exitCode: null, signal: "SIGTERM" }; + }, + }); + + const result = await cli.run(["converge"], { abort: controller.signal }); + + expect(result.spawns[0]?.kills).toEqual(["SIGTERM"]); + expect(chunks).toEqual([["stdout", "working\n"]]); + expect(result.exitCode).toBe(143); + }); + + test("an abort runs the termination ladder on a piped child", async () => { + const controller = new AbortController(); + const late = defineCommand({ + help: { summary: "Spawns after the signal already fired" }, + maySpawn: true, + handler: async (_args, ctx) => { + ctx.report({ kind: "status", subject: "run", status: "ready" }); + await new Promise((resolve) => setTimeout(resolve, 0)); + await ctx.spawn({ command: "alchemy", onOutput: () => {} }); + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { late }, + now: CLOCK, + spawnScript: async (_request, child) => { + child.write("stdout", "still deploying\n"); + await child.nextKill(); + await child.nextKill(); + return { exitCode: null, signal: "SIGKILL" }; + }, + }); + + const result = await cli.run(["late"], { + abort: controller.signal, + onEvent: (event) => { + if (event.kind === "status") { + controller.abort(); + } + }, + }); + + expect(result.spawns[0]?.kills).toEqual(["SIGTERM", "SIGKILL"]); + expect(result.exitCode).toBe(137); + }); + + test("a throwing onOutput terminates the child and ctx.spawn rejects with its error", async () => { + let caught: unknown; + const delivered: string[] = []; + const command = defineCommand({ + help: { summary: "Fails while reading a child's output" }, + maySpawn: true, + handler: async (_args, ctx) => { + try { + await ctx.spawn({ + command: "alchemy", + onOutput: (chunk) => { + delivered.push(chunk); + throw new Error("redaction failed"); + }, + }); + } catch (cause) { + caught = cause; + } + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: command }, + now: CLOCK, + delay: (_ms, signal) => signalDone(signal), + spawnScript: async (_request, child) => { + child.write("stdout", "first"); + child.write("stdout", "second"); + await child.nextKill(); + return { exitCode: null, signal: "SIGTERM" }; + }, + }); + + const result = await cli.run(["converge"]); + + expect(result.spawns[0]?.kills).toEqual(["SIGTERM"]); + expect(delivered).toEqual(["first"]); + expect(caught).toBeInstanceOf(Error); + expect((caught as Error).message).toBe("redaction failed"); + }); + + test("an adapter that returns no piped streams has its child terminated", async () => { + let caught: unknown; + const kills: string[] = []; + const command = defineCommand({ + help: { summary: "Asks for piped output" }, + maySpawn: true, + handler: async (_args, ctx) => { + try { + await ctx.spawn({ command: "alchemy", onOutput: () => {} }); + } catch (cause) { + caught = cause; + } + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: command }, + now: CLOCK, + delay: (_ms, signal) => signalDone(signal), + spawn: () => { + let end: (signal: "SIGTERM" | "SIGKILL") => void = () => {}; + return { + ended: new Promise((resolve) => { + end = (signal) => resolve({ exitCode: null, signal }); + }), + kill: (signal) => { + kills.push(signal); + end(signal); + }, + }; + }, + }); + + await cli.run(["converge"]); + + expect(kills).toEqual(["SIGTERM"]); + expect((caught as Error).message).toContain( + "returned no piped stdout and stderr", + ); + }); + + test("without onOutput the default handoffs are unchanged", async () => { + const cli = createTestCli({ commands: { converge }, now: CLOCK }); + + const human = await cli.run(["converge"]); + const json = await cli.run(["converge", "--json"]); + + expect(human.spawns[0]?.output).toBe("inherit"); + expect(json.spawns[0]?.output).toBe("diagnostic"); + }); +}); diff --git a/packages/cli/package.json b/packages/cli/package.json index d8dfbe90..5b6ffab3 100644 --- a/packages/cli/package.json +++ b/packages/cli/package.json @@ -49,7 +49,7 @@ }, "dependencies": { "@manypkg/tools": "^2.1.2", - "@prisma/cli-engine": "workspace:0.9.0", + "@prisma/cli-engine": "workspace:0.10.0", "@prisma/composer-cli": "0.36.0", "@prisma/compute-sdk": "0.43.0", "@prisma/management-api-sdk": "1.80.0", diff --git a/packages/cli/scripts/conformance.ts b/packages/cli/scripts/conformance.ts index 3403120d..370eb7ce 100644 --- a/packages/cli/scripts/conformance.ts +++ b/packages/cli/scripts/conformance.ts @@ -109,7 +109,24 @@ async function tarball(): Promise { shellPackage: "@prisma/cli", enginePackage: "@prisma/cli-engine", familyPackages: ["@prisma/composer-cli", "@prisma/orm-toolchain"], - exceptions: [], + exceptions: [ + { + familyPackage: "@prisma/composer-cli", + familyPin: "0.9.0", + shellPin: "0.10.0", + reason: "engine 0.10.0 must publish before composer-cli can peer it", + removeWhen: + "composer-cli releases peering 0.10.0 and the follow-up bump PR pins that release", + }, + { + familyPackage: "@prisma/orm-toolchain", + familyPin: "0.9.0", + shellPin: "0.10.0", + reason: "engine 0.10.0 must publish before orm-toolchain can peer it", + removeWhen: + "orm-toolchain releases peering 0.10.0 and the follow-up bump PR pins that release", + }, + ], channel: CHANNEL, sandboxDir: join(WORK_DIR, "sandbox"), }, diff --git a/packages/cli/src/spawn.ts b/packages/cli/src/spawn.ts index 15b8a132..87f1ba02 100644 --- a/packages/cli/src/spawn.ts +++ b/packages/cli/src/spawn.ts @@ -27,8 +27,9 @@ export interface SpawnChildOptions { /** * The engine's spawn seam, adapted through cross-spawn. Human mode * inherits stdio; structured mode pipes both child output streams to - * diagnostics. Neither mode detaches or opens a new console, so the child - * stays in this process's group (POSIX) or console (Windows). + * diagnostics; pipe mode hands both to the engine. No mode detaches or + * opens a new console, so the child stays in this process's group + * (POSIX) or console (Windows). * * The child's own status settles the run: `ended` resolves from the * process `exit` event, waits for the diagnostic relay only up to the @@ -45,7 +46,8 @@ export function makeSpawnChild( const child = spawn(request.command, [...request.args], { cwd: request.cwd, env: request.env, - stdio: structured ? ["inherit", "pipe", "pipe"] : "inherit", + stdio: + request.output === "inherit" ? "inherit" : ["inherit", "pipe", "pipe"], }); const processEnded = new Promise((resolve, reject) => { child.on("error", reject); @@ -53,6 +55,33 @@ export function makeSpawnChild( resolve({ exitCode, signal }); }); }); + const kill = (signal: "SIGTERM" | "SIGKILL") => { + child.kill(signal); + }; + if (request.output === "pipe") { + const sources = [child.stdout, child.stderr].filter( + (source): source is Readable => source !== null, + ); + const closed = Promise.all( + sources.map((source) => { + source.setEncoding("utf8"); + return new Promise((resolve) => source.once("close", resolve)); + }), + ); + return { + ended: processEnded.then(async (result) => { + const drainDeadline = setTimeout(() => { + for (const source of sources) source.destroy(); + }, drainGraceMs); + await closed; + clearTimeout(drainDeadline); + return result; + }), + kill, + ...(child.stdout === null ? {} : { stdout: child.stdout }), + ...(child.stderr === null ? {} : { stderr: child.stderr }), + }; + } if (!structured) { return { ended: processEnded, diff --git a/packages/cli/tests/spawn-adapter.test.ts b/packages/cli/tests/spawn-adapter.test.ts index 05a09a1c..b3abb1cf 100644 --- a/packages/cli/tests/spawn-adapter.test.ts +++ b/packages/cli/tests/spawn-adapter.test.ts @@ -46,6 +46,19 @@ beforeEach(() => { diagnosticText = ""; }); +async function readAll( + source: AsyncIterable | undefined, +): Promise { + if (source === undefined) { + throw new Error("the adapter returned no piped stream"); + } + let text = ""; + for await (const chunk of source) { + text += chunk; + } + return text; +} + async function waitForFile(path: string, timeoutMs = 10_000): Promise { const deadline = Date.now() + timeoutMs; while (!existsSync(path)) { @@ -335,4 +348,59 @@ describe("the shipped spawn adapter", () => { ).toBe(true); } }); + + test("pipe mode hands both decoded streams to the engine and writes nothing itself", async () => { + const child = spawnChild({ + command: NODE, + args: [ + "-e", + "process.stdout.write(Buffer.from([0xf0, 0x9f])); setTimeout(() => { process.stdout.write(Buffer.from([0x98, 0x80])); process.stderr.write('child-err'); process.exit(5); }, 50)", + ], + cwd: process.cwd(), + env: process.env, + output: "pipe", + }); + + const [stdout, stderr, result] = await Promise.all([ + readAll(child.stdout), + readAll(child.stderr), + child.ended, + ]); + + expect(spawnOptionsSeen.at(-1)?.stdio).toEqual(["inherit", "pipe", "pipe"]); + expect(result).toEqual({ exitCode: 5, signal: null }); + expect(stdout).toBe("😀"); + expect(stderr).toBe("child-err"); + expect(diagnosticText).toBe(""); + }); + + test.skipIf(process.platform === "win32")( + "pipe mode: a grandchild holding the pipes does not block settlement past the drain grace", + async () => { + const gracedSpawn = makeSpawnChild( + { write: () => {} }, + { drainGraceMs: 200 }, + ); + const startedAt = Date.now(); + const child = gracedSpawn({ + command: NODE, + args: [ + "-e", + `require("node:child_process").spawn(process.execPath, ["-e", "setTimeout(() => {}, 10000)"], { detached: true, stdio: ["ignore", "inherit", "ignore"] }).unref();`, + ], + cwd: process.cwd(), + env: process.env, + output: "pipe", + }); + const reading = Promise.all([ + readAll(child.stdout).catch(() => ""), + readAll(child.stderr).catch(() => ""), + ]); + + await expect(child.ended).resolves.toEqual({ exitCode: 0, signal: null }); + await reading; + expect(Date.now() - startedAt).toBeLessThan(3_000); + }, + 20_000, + ); }); diff --git a/packages/prisma/package.json b/packages/prisma/package.json index 3af55153..eeafc3a0 100644 --- a/packages/prisma/package.json +++ b/packages/prisma/package.json @@ -50,7 +50,7 @@ }, "dependencies": { "@manypkg/tools": "^2.1.2", - "@prisma/cli-engine": "workspace:0.9.0", + "@prisma/cli-engine": "workspace:0.10.0", "@prisma/composer-cli": "0.36.0", "@prisma/compute-sdk": "0.43.0", "@prisma/management-api-sdk": "1.80.0", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 9688e7e4..15d790ba 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -27,7 +27,7 @@ importers: specifier: ^2.1.2 version: 2.1.2 '@prisma/cli-engine': - specifier: workspace:0.9.0 + specifier: workspace:0.10.0 version: link:../cli-engine '@prisma/composer-cli': specifier: 0.36.0 @@ -210,7 +210,7 @@ importers: specifier: ^2.1.2 version: 2.1.2 '@prisma/cli-engine': - specifier: workspace:0.9.0 + specifier: workspace:0.10.0 version: link:../cli-engine '@prisma/composer-cli': specifier: 0.36.0 From 415c668ae33fc8bb336c58394a6251077b610719 Mon Sep 17 00:00:00 2001 From: willbot Date: Mon, 2 Nov 2026 17:19:31 +0100 Subject: [PATCH 2/6] feat(engine): ctx.spawn transforms a child's output and writes it where the output format sends it Replaces onOutput with transformOutput: the command returns one TransformStream per stream, and the engine pipes the child through it to the stream the default handoff would have used, with backpressure and a flush before ctx.spawn settles. The engine redacts the credential it injects before the command's transform runs. ctx.spawnOptions lets a command check for support, and ctx.spawn refuses option names it does not know. The adapter output setting "pipe" is renamed "engine"; an adapter that returns no streams has its child SIGKILLed at once. Signed-off-by: willbot Signed-off-by: Will Madden --- docs/product/output-conventions.md | 1 + packages/cli-engine/README.md | 31 +- packages/cli-engine/src/context.ts | 18 +- .../src/execution/command-context.ts | 2 + packages/cli-engine/src/execution/spawn.ts | 225 +++++++-- packages/cli-engine/src/exports/index.ts | 3 +- packages/cli-engine/src/runtime.ts | 21 +- packages/cli-engine/src/spawn.ts | 50 +- packages/cli-engine/src/testing.ts | 12 +- .../cli-engine/tests/spawn-real-child.test.ts | 33 +- packages/cli-engine/tests/spawn.test.ts | 436 ++++++++++++++---- packages/cli/src/runtime.ts | 23 + packages/cli/src/spawn.ts | 78 +++- packages/cli/tests/spawn-adapter.test.ts | 8 +- .../cli/tests/spawn-engine-output.test.ts | 196 ++++++++ 15 files changed, 915 insertions(+), 222 deletions(-) create mode 100644 packages/cli/tests/spawn-engine-output.test.ts diff --git a/docs/product/output-conventions.md b/docs/product/output-conventions.md index 5ead36eb..a338cade 100644 --- a/docs/product/output-conventions.md +++ b/docs/product/output-conventions.md @@ -37,6 +37,7 @@ command: - in human mode, the child inherits stdin, stdout, and stderr - in JSON mode, stdin remains inherited while the child's stdout and stderr are routed to the CLI's diagnostic stderr stream +- when the command transforms the child's output (`transformOutput`), stdin remains inherited and the engine reads the child's stdout and stderr, passes each through the command's transform, and writes the result to the same place the two cases above would have: the child's stdout to stdout and its stderr to stderr in human and markdown format, and both to stderr in JSON mode. The engine also replaces the credential it injected into the child's environment with `[redacted]` - JSON stdout remains an NDJSON event stream with exactly one terminal result frame - a non-zero child status is preserved as the process exit code and is diff --git a/packages/cli-engine/README.md b/packages/cli-engine/README.md index 2af35c8f..cce367fe 100644 --- a/packages/cli-engine/README.md +++ b/packages/cli-engine/README.md @@ -75,25 +75,38 @@ A statement is the command's own declared input, unlike a `--confirm` value, whi Each flag answers one question. A run that succeeds with a flag nothing consumed fails with `CLI.CONSENT_UNUSED`, so a mistyped subject or a second answer to one question is never ignored. The same holds for a `--confirm` token no consent asked for, on a command that declares statements (other commands still ignore a leftover `--confirm`; see [#339](https://github.com/prisma/prisma-cli/issues/339)). A refusal also names any given statement flag that matches none of its questions, and any `--confirm` token, so the user hears of it before answering a prompt. That check runs when the handler returns, after the command has acted. A command that asks everything in one batch passes `statements(questions, { last: true })` to get the same failure right after the questions are answered, before it does anything with them. -## Reading a child's output +## Transforming a child's output -A command that declares `maySpawn` runs a child process with `ctx.spawn`. By default the child writes straight to the terminal, or, with `--json`, to stderr. To read the output instead, pass `onOutput`: +A command that declares `maySpawn` runs a child process with `ctx.spawn`. By default the child writes straight to the terminal, or, with `--json`, to stderr. To change the text before it is written, for example to redact secrets, pass `transformOutput`: ```ts +if (!ctx.spawnOptions?.includes("transformOutput")) { + throw new Error("This command needs @prisma/cli-engine 0.10.0 or later."); +} const result = await ctx.spawn({ command: "alchemy", args: ["deploy"], - onOutput: (chunk, stream) => { - redactor.write(stream, chunk); - }, + transformOutput: (stream) => redactor(stream), }); -redactor.flush(); ``` -`chunk` is text decoded as UTF-8, and `stream` is `"stdout"` or `"stderr"`. Chunks from one stream arrive in the order the child wrote them; there is no guaranteed order between the two streams. `ctx.spawn` resolves after the last chunk, so a command that holds text back between chunks writes the rest after the `await`. In json mode stdout carries the engine's frames, so write both streams to stderr. +The engine calls `transformOutput` once for `"stdout"` and once for `"stderr"` before the child starts. It pipes each stream, decoded as UTF-8, through the `TransformStream` you return, and writes what comes out where the child's output would have gone without it: + +- human and markdown format: the child's stdout to stdout, its stderr to stderr; +- json format: both to stderr, so stdout carries only the engine's NDJSON output. + +Each stream keeps its own order; there is no order between the two. The engine reads the child only as fast as the terminal takes the text, so a slow reader slows the child down instead of filling memory. When a stream ends, the transform's `flush` runs and its output is written before `ctx.spawn` settles, including when the child was aborted or failed. A transform can therefore hold text back between chunks, as a redactor that matches across chunk boundaries must. + +When the command declares `needs: { credentials: "child" }`, the engine puts the access token in the child's `PRISMA_SERVICE_TOKEN`. The command never sees that value, so the engine replaces it with `[redacted]` itself, before your transform runs. + +The engine still owns the child. Stdin stays inherited, Ctrl-C reaches the child as before, and a SIGTERM or an abort of `ctx.signal` ends it with SIGTERM, then SIGKILL after a grace period. If a transform throws or errors, the engine stops reading that stream, ends the child the same way, and `ctx.spawn` rejects with that error once the child has exited. With `PRISMA_DEBUG=1`, a stream that fails, or that a grandchild keeps open after the child exits, leaves a line saying its later output was dropped. + +### Checking for support + +Hosts older than 0.10.0 ignore spawn options they do not know, so on those hosts a child would run with its output untransformed. `ctx.spawnOptions` lists the option names `ctx.spawn` accepts; it is absent on those hosts. Check it before spawning, as above. From 0.10.0 on, `ctx.spawn` refuses any option name it does not know, so a future option cannot be ignored silently. -The engine still owns the child. Stdin stays inherited, Ctrl-C reaches the child as before, and a SIGTERM or an abort of `ctx.signal` ends it with SIGTERM, then SIGKILL after a grace period. If `onOutput` throws, the engine stops delivering, ends the child the same way, and `ctx.spawn` rejects with that error once the child has exited. +### Spawn adapters -A spawn adapter supports this by starting the child with stdout and stderr piped when the request's `output` is `"pipe"`, and returning both as `stdout` and `stderr` on the child. The engine ends a child whose adapter did not return both. +A spawn adapter supports `transformOutput` by starting the child with stdout and stderr piped when the request's `output` is `"engine"`, and returning both as `stdout` and `stderr` on the child. If the adapter returns no streams, the engine kills the child with SIGKILL at once and `ctx.spawn` rejects. An adapter written before `"engine"` existed has usually started that child with inherited output by then, so some of it can reach the terminal before the kill. This check catches a wrongly wired adapter; it does not make redaction safe with one. Part of [prisma/prisma-cli](https://github.com/prisma/prisma-cli). diff --git a/packages/cli-engine/src/context.ts b/packages/cli-engine/src/context.ts index d5516fc4..37df427a 100644 --- a/packages/cli-engine/src/context.ts +++ b/packages/cli-engine/src/context.ts @@ -63,11 +63,11 @@ export interface CommandContext< readonly api: ManagementApiClient; /** - * Hands the terminal to a child process and resolves when it ends. - * The child inherits stdio and runs in this process's own group - * (POSIX) or console (Windows), so Ctrl-C reaches it natively. With - * `onOutput`, stdin is still inherited but stdout and stderr are - * delivered to the handler instead; everything below is unchanged. + * Runs a child process and resolves when it ends. The child inherits + * stdin and runs in this process's own group (POSIX) or console + * (Windows), so Ctrl-C reaches it natively. Its stdout and stderr go + * where the output format sends them (see output-conventions.md), + * passing through `transformOutput` when the handler gives one. * * While a child is live the engine neither aborts nor exits on a * delivered signal: it records signals and replays them into its @@ -87,6 +87,14 @@ export interface CommandContext< */ readonly spawn: (options: SpawnOptions) => Promise; + /** + * The option names ctx.spawn accepts on this host. ctx.spawn refuses + * any other name. Absent on hosts older than 0.10.0, which ignore + * options they do not know, so a command that depends on an option + * checks for it here before spawning. + */ + readonly spawnOptions: readonly (keyof SpawnOptions)[]; + /** * How the run's most recent completed child ended, or undefined when * none has run. The engine records every child ctx.spawn returns, so diff --git a/packages/cli-engine/src/execution/command-context.ts b/packages/cli-engine/src/execution/command-context.ts index e3f27187..49e2d1dc 100644 --- a/packages/cli-engine/src/execution/command-context.ts +++ b/packages/cli-engine/src/execution/command-context.ts @@ -15,6 +15,7 @@ import { } from "../presentation"; import { type Diagnostic, notOk, okVoid } from "../protocol"; import type { LoadedConfigFile, OutputStream } from "../runtime"; +import { SPAWN_OPTION_NAMES } from "../spawn"; import { buildManagementApiClient } from "./api-client"; import { constructionError } from "./command-tree"; import type { Invocation, RunState } from "./engine"; @@ -178,6 +179,7 @@ export function makeContext( return api; }, spawn: makeSpawn(invocation, def), + spawnOptions: SPAWN_OPTION_NAMES, lastChild: () => state.lastChild, report: (event) => reportEvent(invocation, event), prompt: makePromptSurface(invocation), diff --git a/packages/cli-engine/src/execution/spawn.ts b/packages/cli-engine/src/execution/spawn.ts index ce1dca40..7e1f56d3 100644 --- a/packages/cli-engine/src/execution/spawn.ts +++ b/packages/cli-engine/src/execution/spawn.ts @@ -6,14 +6,16 @@ import { } from "../credential-manager"; import type { EngineEvent } from "../events"; import { CliStructuredError } from "../protocol"; +import type { OutputStream } from "../runtime"; import type { - ChildOutputStream, ChildResult, + ChildStreamName, SpawnChild, SpawnedChild, SpawnOptions, SpawnRequest, } from "../spawn"; +import { SPAWN_OPTION_NAMES } from "../spawn"; import { constructionError } from "./command-tree"; import { makeDebugLog } from "./debug"; import type { Invocation } from "./engine"; @@ -74,6 +76,14 @@ export function makeSpawn( `command '${state.commandId}' called ctx.spawn without declaring maySpawn`, ); } + const unknown = Object.keys(options).filter( + (name) => !(SPAWN_OPTION_NAMES as readonly string[]).includes(name), + ); + if (unknown.length > 0) { + throw constructionError( + `command '${state.commandId}' passed ctx.spawn options it does not accept: ${unknown.join(", ")}`, + ); + } if (state.delegatedTerminal !== undefined) { throw constructionError( `command '${state.commandId}' called ctx.spawn while a child was still live (one live child per run)`, @@ -134,12 +144,14 @@ async function runDelegated( const state = invocation.state; const debug = makeDebugLog(invocation.runtime); try { + const transforms = makeTransforms(options); + const env = await composeChildEnv(invocation, def, options.env); const request: SpawnRequest = { command: options.command, args: options.args ?? [], cwd: options.cwd ?? invocation.runtime.cwd, - env: await composeChildEnv(invocation, def, options.env), - output: childOutput(options, state.format === "json"), + env, + output: childOutput(transforms, state.format === "json"), }; debug( `spawn: ${request.command} ${request.args.join(" ")} (cwd ${request.cwd})`, @@ -149,7 +161,8 @@ async function runDelegated( terminal, spawnChild, request, - options.onOutput, + transforms, + injectedSecrets(def, env), ); debug( `spawn ended: exitCode=${String(result.exitCode)} signal=${String(result.signal)} recorded=[${terminal.recorded.join(",")}]`, @@ -168,12 +181,27 @@ async function runDelegated( } } +type ChildTransforms = Readonly< + Record> +>; + +function makeTransforms(options: SpawnOptions): ChildTransforms | undefined { + const transformOutput = options.transformOutput; + if (transformOutput === undefined) { + return undefined; + } + return { + stdout: transformOutput("stdout"), + stderr: transformOutput("stderr"), + }; +} + function childOutput( - options: SpawnOptions, + transforms: ChildTransforms | undefined, structured: boolean, ): SpawnRequest["output"] { - if (options.onOutput !== undefined) { - return "pipe"; + if (transforms !== undefined) { + return "engine"; } return structured ? "diagnostic" : "inherit"; } @@ -193,7 +221,8 @@ async function runChild( terminal: DelegatedTerminal, spawnChild: SpawnChild, request: SpawnRequest, - onOutput: SpawnOptions["onOutput"], + transforms: ChildTransforms | undefined, + secrets: readonly string[], ): Promise { let child: SpawnedChild; try { @@ -205,20 +234,28 @@ async function runChild( if (terminal.recorded.includes("SIGTERM") || terminal.recorded.length > 1) { child.kill("SIGTERM"); } + if ( + transforms !== undefined && + (child.stdout === undefined || child.stderr === undefined) + ) { + return await refuseUnpipedChild(invocation, child); + } const ended = new AbortController(); armTerminationLadder(invocation, terminal, child, ended.signal); try { - const delivery = - onOutput === undefined + const relay = + transforms === undefined ? undefined - : deliverOutput(terminal, child, onOutput); + : relayOutput(invocation, terminal, child, transforms, secrets); let result: ChildResult; try { result = recordCompletedChild(invocation, await child.ended); } catch (cause) { + terminal.termination.abort(); + await relay; throw spawnFailedError(request.command, cause); } - const failure = await delivery; + const failure = await relay; if (failure !== undefined) { throw failure.cause; } @@ -228,53 +265,143 @@ async function runChild( } } -/** Reads both piped streams into the handler's callback. A throwing - * callback ends delivery and terminates the child on the abort ladder; - * the failure is returned so ctx.spawn rejects with it once the child - * has ended. A stream that fails on its own (the adapter destroying a - * pipe a grandchild held open) just ends. */ -async function deliverOutput( +/** The adapter ignored output "engine", so the child is writing + * wherever the adapter sent it. Nothing was promised to a child the + * engine cannot read, so it gets no grace period. */ +async function refuseUnpipedChild( + invocation: Invocation, + child: SpawnedChild, +): Promise { + makeDebugLog(invocation.runtime)( + "spawn: the adapter returned no stdout and stderr for output engine; SIGKILL", + ); + child.kill("SIGKILL"); + const ended = await child.ended.catch(() => undefined); + if (ended !== undefined) { + recordCompletedChild(invocation, ended); + } + throw new Error( + "@prisma/cli-engine: ctx.spawn was given transformOutput but the Runtime's spawn adapter returned no piped stdout and stderr", + ); +} + +/** Pipes each child stream through its transform to the stream the + * default output mode would have used. A failing transform ends the + * child on the abort ladder and is returned, so ctx.spawn rejects with + * it; the other stream keeps flowing until the child ends. */ +async function relayOutput( + invocation: Invocation, terminal: DelegatedTerminal, child: SpawnedChild, - onOutput: NonNullable, + transforms: ChildTransforms, + secrets: readonly string[], ): Promise<{ readonly cause: unknown } | undefined> { - if (child.stdout === undefined || child.stderr === undefined) { - terminal.termination.abort(); - return { - cause: new Error( - "@prisma/cli-engine: ctx.spawn was given onOutput but the Runtime's spawn adapter returned no piped stdout and stderr", - ), - }; - } + const runtime = invocation.runtime; + const debug = makeDebugLog(runtime); + const structured = invocation.state.format === "json"; let failure: { readonly cause: unknown } | undefined; - const read = async ( - source: AsyncIterable, - stream: ChildOutputStream, - ): Promise => { - try { - for await (const chunk of source) { - if (failure !== undefined) { - return; - } - try { - onOutput(chunk, stream); - } catch (cause) { - failure = { cause }; - terminal.termination.abort(); - return; - } - } - } catch { + const relay = async (stream: ChildStreamName): Promise => { + const source = stream === "stdout" ? child.stdout : child.stderr; + if (source === undefined) { return; } + const destination = + stream === "stdout" && !structured ? runtime.stdout : runtime.stderr; + try { + await relayStream( + source, + [...secrets.map(redactingTransform), transforms[stream]], + destination, + (cause) => { + debug( + `spawn: the child's ${stream} failed, so its later output was dropped: ${errorMessage(cause)}`, + ); + }, + ); + } catch (cause) { + failure ??= { cause }; + terminal.termination.abort(); + } }; - await Promise.all([ - read(child.stdout, "stdout"), - read(child.stderr, "stderr"), - ]); + await Promise.all([relay("stdout"), relay("stderr")]); return failure; } +/** Reads `source` only as fast as `destination` drains, so a slow + * destination makes the child block on its own writes. A source that + * fails ends normally, so the transform still flushes. Rejects when + * the transform or the destination fails. */ +export async function relayStream( + source: AsyncIterable, + transforms: readonly TransformStream[], + destination: OutputStream, + onSourceError: (cause: unknown) => void, +): Promise { + await transforms + .reduce( + (readable, transform) => readable.pipeThrough(transform), + ReadableStream.from(endOnError(source, onSourceError)), + ) + .pipeTo( + new WritableStream({ + write: async (text) => { + destination.write(text); + await destination.drained?.(); + }, + }), + ); +} + +/** The credential the engine put in the child's environment. The + * command never sees it, so only the engine can redact it. */ +function injectedSecrets( + def: AnyCommand, + env: Readonly>, +): readonly string[] { + if (def.needs.credentials !== "child") { + return []; + } + const token = env[SERVICE_TOKEN_ENV_VAR]; + return token === undefined || token === "" ? [] : [token]; +} + +/** Replaces `secret` with "[redacted]" across chunk boundaries by + * holding back up to `secret.length - 1` characters until the next + * chunk or the end of the stream. */ +function redactingTransform(secret: string): TransformStream { + let held = ""; + return new TransformStream({ + transform: (chunk, controller) => { + const text = (held + chunk).replaceAll(secret, "[redacted]"); + const keep = Math.min(text.length, secret.length - 1); + held = text.slice(text.length - keep); + if (text.length > keep) { + controller.enqueue(text.slice(0, text.length - keep)); + } + }, + flush: (controller) => { + if (held !== "") { + controller.enqueue(held); + } + }, + }); +} + +async function* endOnError( + source: AsyncIterable, + onError: (cause: unknown) => void, +): AsyncGenerator { + try { + yield* source; + } catch (cause) { + onError(cause); + } +} + +function errorMessage(cause: unknown): string { + return cause instanceof Error ? cause.message : String(cause); +} + /** The engine mints the result the handler sees and keeps it on the * run: the settlement then reads the child's status from the engine's * own record rather than from anything the handler hands back, and a diff --git a/packages/cli-engine/src/exports/index.ts b/packages/cli-engine/src/exports/index.ts index 79886b78..020f6ee7 100644 --- a/packages/cli-engine/src/exports/index.ts +++ b/packages/cli-engine/src/exports/index.ts @@ -140,6 +140,7 @@ export { export type { EngineCommandSnapshot, RunSummary } from "../run-summary"; export { type Host, + type HostOutputStream, type HostProcess, type InputStream, type LoadedConfig, @@ -149,9 +150,9 @@ export { type Runtime, } from "../runtime"; export { - type ChildOutputStream, type ChildResult, type ChildStatusSettlement, + type ChildStreamName, type ExitWithChildStatusOptions, exitWithChildStatus, type SpawnChild, diff --git a/packages/cli-engine/src/runtime.ts b/packages/cli-engine/src/runtime.ts index 7bbb620e..216002dc 100644 --- a/packages/cli-engine/src/runtime.ts +++ b/packages/cli-engine/src/runtime.ts @@ -12,6 +12,9 @@ export interface OutputStream { * The engine reads it at render time rather than caching it, so a * terminal resized mid-run is respected by the next thing drawn. */ readonly columns?: number; + /** Resolves once the stream can take more text without buffering it + * in memory. Absent means the stream never asks the writer to wait. */ + readonly drained?: () => Promise; } /** @@ -155,6 +158,16 @@ export interface Host { readonly arch: string; } +/** The parts of a Node writable the bin reads off process.stdout and + * process.stderr. */ +export interface HostOutputStream { + write(text: string): unknown; + isTTY?: boolean; + writableNeedDrain?: boolean; + once?(event: "drain" | "error" | "close", listener: () => void): unknown; + off?(event: "drain" | "error" | "close", listener: () => void): unknown; +} + /** * The minimal process surface a bin adapts a Runtime from — Node's * `process` satisfies it structurally. The engine never reads it; it @@ -168,12 +181,8 @@ export interface HostProcess { readonly platform: string; readonly arch: string; cwd(): string; - readonly stdout: { write(text: string): unknown; isTTY?: boolean }; - readonly stderr: { - write(text: string): unknown; - isTTY?: boolean; - columns?: number; - }; + readonly stdout: HostOutputStream; + readonly stderr: HostOutputStream & { columns?: number }; readonly stdin: { isTTY?: boolean; setRawMode?(enabled: boolean): unknown; diff --git a/packages/cli-engine/src/spawn.ts b/packages/cli-engine/src/spawn.ts index 9178a875..9a0088f9 100644 --- a/packages/cli-engine/src/spawn.ts +++ b/packages/cli-engine/src/spawn.ts @@ -19,15 +19,17 @@ export interface SpawnRequest { * NDJSON on stdout in that mode and cannot detect an adapter that * ignores this field — inherited child stdout would silently corrupt * the stream. Adapters written before this field existed must be - * updated. "pipe" keeps stdin inherited and pipes stdout and stderr - * to this process: the adapter MUST return them as - * SpawnedChild.stdout and SpawnedChild.stderr, and the engine kills a - * child returned without them. */ - readonly output: "inherit" | "diagnostic" | "pipe"; + * updated. "engine" keeps stdin inherited and pipes stdout and stderr + * back to the engine: the adapter MUST return them as + * SpawnedChild.stdout and SpawnedChild.stderr. The engine SIGKILLs a + * child returned without them, but an adapter that predates "engine" + * has already started that child with inherited output, so some of + * it can reach the terminal before the kill. */ + readonly output: "inherit" | "diagnostic" | "engine"; } -/** Which of the child's output streams a chunk came from. */ -export type ChildOutputStream = "stdout" | "stderr"; +/** Names one of the child's two output streams. */ +export type ChildStreamName = "stdout" | "stderr"; /** How a child ended. A signal-killed child carries `signal` and a null * `exitCode`, and is an abort rather than a failure: exitWithChildStatus @@ -45,7 +47,7 @@ export interface SpawnedChild { readonly ended: Promise; readonly kill: (signal: "SIGTERM" | "SIGKILL") => void; /** The child's piped output, decoded as UTF-8. Required when the - * request's `output` is "pipe", absent otherwise. Each ends when + * request's `output` is "engine", absent otherwise. Each ends when * the child closes it; an adapter ends both no later than a grace * period after the child exits, so a grandchild holding the pipe * open cannot hold the run. */ @@ -58,7 +60,9 @@ export interface SpawnedChild { * inherited stdio, in the caller's own process group (POSIX) / console * (Windows). In structured mode stdin remains inherited while stdout and * stderr are routed to the host's diagnostic stream, preserving framed - * stdout. Neither mode detaches or opens a new console. + * stdout. In engine mode stdin remains inherited and both output + * streams are returned to the engine. No mode detaches or opens a new + * console. */ export type SpawnChild = (request: SpawnRequest) => SpawnedChild; @@ -72,19 +76,27 @@ export interface SpawnOptions { * credential variables are applied last and cannot be overridden. */ readonly env?: Readonly>; /** - * Receives the child's stdout and stderr instead of the terminal or - * the diagnostic stream, in the order each stream produced them, and - * the handler writes them on itself. The engine still owns the child: - * stdin, signals, the abort ladder and the exit status are as without - * it, and ctx.spawn resolves only after the last chunk was delivered. - * In json mode stdout carries the engine's frames, so write both - * streams to stderr. If it throws, the engine stops delivering, - * terminates the child as on abort, and ctx.spawn rejects with that - * error once the child has ended. + * Passes the child's output through a transform before the engine + * writes it. The engine calls this once per stream before the child + * starts, pipes that stream through the returned transform as UTF-8 + * text, and writes the result where the child's output goes without + * it. The transform's `flush` runs when the stream ends, before + * ctx.spawn settles. If a transform throws or errors, the engine ends + * the child as on abort and ctx.spawn rejects with that error. The + * engine replaces the credential it injects into the child's + * environment with `[redacted]` before the transform sees the text. + * Hosts older than 0.10.0 ignore this option: check + * `ctx.spawnOptions?.includes("transformOutput")` first. */ - readonly onOutput?: (chunk: string, stream: ChildOutputStream) => void; + readonly transformOutput?: ( + stream: ChildStreamName, + ) => TransformStream; } +/** Every option ctx.spawn accepts; it refuses any other. */ +export const SPAWN_OPTION_NAMES: readonly (keyof SpawnOptions)[] = + Object.freeze(["command", "args", "cwd", "env", "transformOutput"]); + export const CHILD_STATUS: unique symbol = Symbol.for( "@prisma/cli-engine.childStatus", ); diff --git a/packages/cli-engine/src/testing.ts b/packages/cli-engine/src/testing.ts index 9f4c4f9d..11d12539 100644 --- a/packages/cli-engine/src/testing.ts +++ b/packages/cli-engine/src/testing.ts @@ -19,8 +19,8 @@ import type { PresentedResult } from "./presentation"; import type { RunSummary } from "./run-summary"; import type { Host, Runtime } from "./runtime"; import type { - ChildOutputStream, ChildResult, + ChildStreamName, SpawnChild, SpawnRequest, } from "./spawn"; @@ -30,15 +30,15 @@ import type { TelemetryDeclaration } from "./telemetry/report"; /** * What a scripted fake child does. `nextKill` resolves with each signal * the engine delivers, so a script can model a child that ignores - * SIGTERM and only dies on SIGKILL. `write` emits output on a piped - * request (`output: "pipe"`) and is ignored otherwise; both streams end - * when the program returns. + * SIGTERM and only dies on SIGKILL. `write` emits output when the + * request's `output` is "engine" and is ignored otherwise; both streams + * end when the program returns. */ export type ScriptedChildProgram = ( request: SpawnRequest, child: { readonly nextKill: () => Promise<"SIGTERM" | "SIGKILL">; - readonly write: (stream: ChildOutputStream, chunk: string) => void; + readonly write: (stream: ChildStreamName, chunk: string) => void; }, ) => ChildResult | Promise; @@ -106,7 +106,7 @@ function scriptedSpawn(program: ScriptedChildProgram): SpawnChild { const delivered: ("SIGTERM" | "SIGKILL")[] = []; const waiting: Array<(signal: "SIGTERM" | "SIGKILL") => void> = []; const piped = - request.output === "pipe" + request.output === "engine" ? { stdout: scriptedStream(), stderr: scriptedStream() } : undefined; return { diff --git a/packages/cli-engine/tests/spawn-real-child.test.ts b/packages/cli-engine/tests/spawn-real-child.test.ts index c3c37766..82f033c8 100644 --- a/packages/cli-engine/tests/spawn-real-child.test.ts +++ b/packages/cli-engine/tests/spawn-real-child.test.ts @@ -12,6 +12,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { fileURLToPath } from "node:url"; import { + type ChildStreamName, defineCommand, exitWithChildStatus, type SpawnChild, @@ -47,8 +48,9 @@ const realSpawn: SpawnChild = (request) => { }; }; -/** The bin adapter's pipe mode: stdin inherited, both output streams - * handed to the engine. */ +/** Engine mode with both output streams returned to the engine. The + * shipped adapter's engine mode is tested end to end in + * packages/cli/tests/spawn-engine-output.test.ts. */ const pipingSpawn: SpawnChild = (request) => { const child = spawn(request.command, [...request.args], { cwd: request.cwd, @@ -70,25 +72,33 @@ const pipingSpawn: SpawnChild = (request) => { }; }; -type Chunk = readonly [stream: "stdout" | "stderr", chunk: string]; +type Chunk = readonly [stream: ChildStreamName, chunk: string]; function pipingCommand(chunks: Chunk[], ...childArgs: string[]) { return defineCommand({ - help: { summary: "Reads the fixture child's output" }, + help: { summary: "Transforms the fixture child's output" }, maySpawn: true, handler: async (_args, ctx) => { await ctx.spawn({ command: NODE, args: [CHILD, ...childArgs], - onOutput: (chunk, stream) => { - chunks.push([stream, chunk]); - }, + transformOutput: recording(chunks), }); return ok(exitWithChildStatus()); }, }); } +function recording(chunks: Chunk[]) { + return (stream: ChildStreamName) => + new TransformStream({ + transform: (chunk, controller) => { + chunks.push([stream, chunk]); + controller.enqueue(chunk); + }, + }); +} + function childCommand(...childArgs: string[]) { return defineCommand({ help: { summary: "Runs the fixture child" }, @@ -301,7 +311,7 @@ describe.skipIf(process.platform === "win32")( expect(result.exitCode).toBe(137); }); - test("piped output reaches onOutput in order, and the exit code passes through", async () => { + test("transformed output keeps each stream's order, and the exit code passes through", async () => { const chunks: Chunk[] = []; const cli = createTestCli({ commands: { converge: pipingCommand(chunks, "chatter", "4") }, @@ -316,7 +326,8 @@ describe.skipIf(process.platform === "win32")( ["stderr", "err-1\n"], ["stdout", "out-2\n"], ]); - expect(result.stdout).toBe(""); + expect(result.stdout).toBe("out-1\nout-2\n"); + expect(result.stderr).toBe("err-1\n"); }); test("a piped child that ignores SIGTERM is killed on abort, leaving no orphan", async () => { @@ -333,9 +344,7 @@ describe.skipIf(process.platform === "win32")( await ctx.spawn({ command: NODE, args: [CHILD, "chatter-ignore-term", pidFile], - onOutput: (chunk, stream) => { - chunks.push([stream, chunk]); - }, + transformOutput: recording(chunks), }); return ok(exitWithChildStatus()); }, diff --git a/packages/cli-engine/tests/spawn.test.ts b/packages/cli-engine/tests/spawn.test.ts index f0e24f9d..b62d0e82 100644 --- a/packages/cli-engine/tests/spawn.test.ts +++ b/packages/cli-engine/tests/spawn.test.ts @@ -10,8 +10,9 @@ import { defineSessionCommand, exitWithChildStatus, type Runtime, + type SpawnOptions, } from "@prisma/cli-engine"; -import { ok } from "@prisma/cli-engine/protocol"; +import { type CliStructuredError, ok } from "@prisma/cli-engine/protocol"; import { createTestCli, mintTestJwt, @@ -19,6 +20,7 @@ import { type SpawnRecord, } from "@prisma/cli-engine/testing"; import { describe, expect, test } from "vitest"; +import { relayStream } from "../src/execution/spawn"; const NOW = new Date("2030-01-01T00:00:00.000Z"); const CLOCK = () => NOW; @@ -1809,110 +1811,135 @@ function controllableRuntime() { }; } -type Chunk = readonly [stream: "stdout" | "stderr", chunk: string]; +/** Replaces `secret` across chunk boundaries by holding back up to + * `secret.length - 1` characters, which flush writes at the end. */ +function redacting(secret: string): () => TransformStream { + return () => { + let held = ""; + return new TransformStream({ + transform: (chunk, controller) => { + const text = (held + chunk).replaceAll(secret, "[redacted]"); + const keep = Math.min(text.length, secret.length - 1); + held = text.slice(text.length - keep); + if (text.length > keep) { + controller.enqueue(text.slice(0, text.length - keep)); + } + }, + flush: (controller) => { + if (held !== "") { + controller.enqueue(held); + } + }, + }); + }; +} -function pipingCommand(chunks: Chunk[]) { +const upperCase = () => + new TransformStream({ + transform: (chunk, controller) => { + controller.enqueue(chunk.toUpperCase()); + }, + }); + +function transformingCommand( + transformOutput: SpawnOptions["transformOutput"], + after?: () => void, +) { return defineCommand({ - help: { summary: "Reads a child's output" }, + help: { summary: "Transforms a child's output" }, maySpawn: true, handler: async (_args, ctx) => { await ctx.spawn({ command: "alchemy", args: ["deploy"], - onOutput: (chunk, stream) => { - chunks.push([stream, chunk]); - }, + transformOutput, }); + after?.(); return ok(exitWithChildStatus()); }, }); } -describe("piped child output", () => { - test("onOutput receives both streams in the order the child wrote them", async () => { - const chunks: Chunk[] = []; +const chatter: ScriptedChildProgram = async (_request, child) => { + child.write("stdout", "one "); + child.write("stdout", "two\n"); + await new Promise((resolve) => setTimeout(resolve, 0)); + child.write("stderr", "warn\n"); + await new Promise((resolve) => setTimeout(resolve, 0)); + child.write("stdout", "three\n"); + return { exitCode: 0, signal: null }; +}; + +describe("transformed child output", () => { + test("human mode writes each transformed stream where the child would have written it", async () => { const cli = createTestCli({ - commands: { converge: pipingCommand(chunks) }, + commands: { converge: transformingCommand(upperCase) }, now: CLOCK, - spawnScript: async (_request, child) => { - child.write("stdout", "one "); - child.write("stdout", "two\n"); - await new Promise((resolve) => setTimeout(resolve, 0)); - child.write("stderr", "warn\n"); - await new Promise((resolve) => setTimeout(resolve, 0)); - child.write("stdout", "three\n"); - return { exitCode: 0, signal: null }; - }, + spawnScript: chatter, }); const result = await cli.run(["converge", "--format", "human"]); expect(result.exitCode).toBe(0); - expect(result.spawns[0]?.output).toBe("pipe"); - expect(chunks).toEqual([ - ["stdout", "one "], - ["stdout", "two\n"], - ["stderr", "warn\n"], - ["stdout", "three\n"], - ]); - expect(result.stdout).toBe(""); - expect(result.stderr).toBe(""); + expect(result.spawns[0]?.output).toBe("engine"); + expect(result.stdout).toBe("ONE TWO\nTHREE\n"); + expect(result.stderr).toBe("WARN\n"); }); - test("json mode pipes too, and keeps stdout framed", async () => { - const chunks: Chunk[] = []; + test("markdown mode follows the inherited handoff too", async () => { const cli = createTestCli({ - commands: { converge: pipingCommand(chunks) }, + commands: { converge: transformingCommand(upperCase) }, now: CLOCK, - spawnScript: (_request, child) => { - child.write("stdout", "hello"); - return { exitCode: 0, signal: null }; - }, + spawnScript: chatter, + }); + + const result = await cli.run(["converge", "--format", "markdown"]); + + expect(result.stdout).toBe("ONE TWO\nTHREE\n"); + expect(result.stderr).toBe("WARN\n"); + }); + + test("json mode writes both streams to stderr and keeps stdout to the result frame", async () => { + const cli = createTestCli({ + commands: { converge: transformingCommand(upperCase) }, + now: CLOCK, + spawnScript: chatter, }); const result = await cli.run(["converge", "--json"]); - expect(result.spawns[0]?.output).toBe("pipe"); - expect(chunks).toEqual([["stdout", "hello"]]); + expect(result.spawns[0]?.output).toBe("engine"); + expect(result.stderr).toBe("ONE TWO\nWARN\nTHREE\n"); expect(result.json).toHaveLength(1); expect(result.json[0]).toMatchObject({ kind: "result" }); }); - test("ctx.spawn resolves only after the last chunk was delivered", async () => { - const seen: string[] = []; - const command = defineCommand({ - help: { summary: "Reads a child's output" }, - maySpawn: true, - handler: async (_args, ctx) => { - await ctx.spawn({ - command: "alchemy", - onOutput: (chunk) => { - seen.push(chunk); - }, - }); - seen.push("resolved"); - return ok(exitWithChildStatus()); - }, - }); + test("a held-back tail is flushed before ctx.spawn resolves", async () => { + const order: string[] = []; const cli = createTestCli({ - commands: { converge: command }, + commands: { + converge: transformingCommand(redacting("hunter2"), () => { + order.push("resolved"); + }), + }, now: CLOCK, - spawnScript: (_request, child) => { - for (const chunk of ["a", "b", "c"]) { - child.write("stderr", chunk); - } + spawnScript: async (_request, child) => { + child.write("stdout", "token=hun"); + await new Promise((resolve) => setTimeout(resolve, 0)); + child.write("stdout", "ter2 done"); return { exitCode: 0, signal: null }; }, }); - await cli.run(["converge"]); + const result = await cli.run(["converge", "--format", "human"]); - expect(seen).toEqual(["a", "b", "c", "resolved"]); + expect(result.stdout).toBe("token=[redacted] done"); + expect(order).toEqual(["resolved"]); }); test("the child's exit code still settles the run", async () => { const cli = createTestCli({ - commands: { converge: pipingCommand([]) }, + commands: { converge: transformingCommand(upperCase) }, now: CLOCK, spawnScript: (_request, child) => { child.write("stderr", "failed\n"); @@ -1923,16 +1950,16 @@ describe("piped child output", () => { const result = await cli.run(["converge", "--format", "human"]); expect(result.exitCode).toBe(3); + expect(result.stderr).toBe("FAILED\n"); }); - test("a SIGTERM delivered while piping is forwarded and the child's end settles the run", async () => { - const chunks: Chunk[] = []; + test("a SIGTERM delivered while transforming is forwarded, and the held-back tail is still written", async () => { const controller = new AbortController(); const cli = createTestCli({ - commands: { converge: pipingCommand(chunks) }, + commands: { converge: transformingCommand(redacting("hunter2")) }, now: CLOCK, spawnScript: async (_request, child) => { - child.write("stdout", "working\n"); + child.write("stdout", "working on hunt"); controller.abort("SIGTERM"); await child.nextKill(); return { exitCode: null, signal: "SIGTERM" }; @@ -1942,19 +1969,19 @@ describe("piped child output", () => { const result = await cli.run(["converge"], { abort: controller.signal }); expect(result.spawns[0]?.kills).toEqual(["SIGTERM"]); - expect(chunks).toEqual([["stdout", "working\n"]]); + expect(result.stdout).toBe("working on hunt"); expect(result.exitCode).toBe(143); }); - test("an abort runs the termination ladder on a piped child", async () => { + test("an abort runs the termination ladder on a child whose output is transformed", async () => { const controller = new AbortController(); const late = defineCommand({ - help: { summary: "Spawns after the signal already fired" }, + help: { summary: "Aborts while a child runs" }, maySpawn: true, handler: async (_args, ctx) => { ctx.report({ kind: "status", subject: "run", status: "ready" }); await new Promise((resolve) => setTimeout(resolve, 0)); - await ctx.spawn({ command: "alchemy", onOutput: () => {} }); + await ctx.spawn({ command: "alchemy", transformOutput: upperCase }); return ok(exitWithChildStatus()); }, }); @@ -1979,23 +2006,28 @@ describe("piped child output", () => { }); expect(result.spawns[0]?.kills).toEqual(["SIGTERM", "SIGKILL"]); + expect(result.stdout).toBe("STILL DEPLOYING\n"); expect(result.exitCode).toBe(137); }); - test("a throwing onOutput terminates the child and ctx.spawn rejects with its error", async () => { + test("a throwing transform ends the child, and ctx.spawn rejects with its error", async () => { let caught: unknown; - const delivered: string[] = []; const command = defineCommand({ - help: { summary: "Fails while reading a child's output" }, + help: { summary: "Fails while transforming a child's output" }, maySpawn: true, handler: async (_args, ctx) => { try { await ctx.spawn({ command: "alchemy", - onOutput: (chunk) => { - delivered.push(chunk); - throw new Error("redaction failed"); - }, + transformOutput: (stream) => + new TransformStream({ + transform: (chunk, controller) => { + if (stream === "stdout" && chunk === "second") { + throw new Error("redaction failed"); + } + controller.enqueue(chunk); + }, + }), }); } catch (cause) { caught = cause; @@ -2010,28 +2042,59 @@ describe("piped child output", () => { spawnScript: async (_request, child) => { child.write("stdout", "first"); child.write("stdout", "second"); + child.write("stdout", "third"); await child.nextKill(); + child.write("stderr", "stopping"); return { exitCode: null, signal: "SIGTERM" }; }, }); - const result = await cli.run(["converge"]); + const result = await cli.run(["converge", "--format", "human"]); expect(result.spawns[0]?.kills).toEqual(["SIGTERM"]); - expect(delivered).toEqual(["first"]); - expect(caught).toBeInstanceOf(Error); + expect(result.stdout).toBe("first"); + expect(result.stderr).toBe("stopping"); expect((caught as Error).message).toBe("redaction failed"); }); - test("an adapter that returns no piped streams has its child terminated", async () => { + test("a throwing transformOutput rejects ctx.spawn before any child starts", async () => { + let caught: unknown; + const command = defineCommand({ + help: { summary: "Fails to build its transform" }, + maySpawn: true, + handler: async (_args, ctx) => { + try { + await ctx.spawn({ + command: "alchemy", + transformOutput: () => { + throw new Error("no redactor"); + }, + }); + } catch (cause) { + caught = cause; + } + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ commands: { converge: command }, now: CLOCK }); + + const result = await cli.run(["converge"]); + + expect(result.spawns).toEqual([]); + expect((caught as Error).message).toBe("no redactor"); + }); + + test("a child the adapter could not keep running still has its held-back tail flushed before ctx.spawn rejects", async () => { let caught: unknown; - const kills: string[] = []; const command = defineCommand({ - help: { summary: "Asks for piped output" }, + help: { summary: "Spawns a child whose adapter fails" }, maySpawn: true, handler: async (_args, ctx) => { try { - await ctx.spawn({ command: "alchemy", onOutput: () => {} }); + await ctx.spawn({ + command: "alchemy", + transformOutput: redacting("hunter2"), + }); } catch (cause) { caught = cause; } @@ -2042,6 +2105,36 @@ describe("piped child output", () => { commands: { converge: command }, now: CLOCK, delay: (_ms, signal) => signalDone(signal), + spawnScript: (_request, child) => { + child.write("stdout", "partial hun"); + throw new Error("kill EPERM"); + }, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.stdout).toBe("partial hun"); + expect((caught as CliStructuredError).code).toBe("CLI.SPAWN_FAILED"); + }); + + test("an adapter that returns no streams has its child killed at once", async () => { + let caught: unknown; + const kills: string[] = []; + const command = defineCommand({ + help: { summary: "Asks for transformed output" }, + maySpawn: true, + handler: async (_args, ctx) => { + try { + await ctx.spawn({ command: "alchemy", transformOutput: upperCase }); + } catch (cause) { + caught = cause; + } + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: command }, + now: CLOCK, spawn: () => { let end: (signal: "SIGTERM" | "SIGKILL") => void = () => {}; return { @@ -2058,13 +2151,146 @@ describe("piped child output", () => { await cli.run(["converge"]); - expect(kills).toEqual(["SIGTERM"]); + expect(kills).toEqual(["SIGKILL"]); expect((caught as Error).message).toContain( "returned no piped stdout and stderr", ); }); - test("without onOutput the default handoffs are unchanged", async () => { + test("a stream that fails leaves a debug line, and the transform still flushes", async () => { + const failing: AsyncIterable = { + async *[Symbol.asyncIterator]() { + yield "before hun"; + throw new Error("the pipe was closed"); + }, + }; + const cli = createTestCli({ + commands: { converge: transformingCommand(redacting("hunter2")) }, + now: CLOCK, + spawn: () => ({ + ended: Promise.resolve({ exitCode: 0, signal: null }), + kill: () => {}, + stdout: failing, + stderr: (async function* () {})(), + }), + }); + + const result = await cli.run(["converge", "--format", "human"], { + env: { PRISMA_DEBUG: "1" }, + }); + + expect(result.stdout).toBe("before hun"); + expect(result.stderr).toContain( + "spawn: the child's stdout failed, so its later output was dropped: the pipe was closed", + ); + }); + + test("the credential the engine injects is redacted before the command's transform sees it", async () => { + const token = jwtExpiringIn(3600, "ws_1"); + const seenByCommand: string[] = []; + const command = defineCommand({ + help: { + summary: "Hands credentials to a child and transforms its output", + }, + maySpawn: true, + needs: { credentials: "child" }, + handler: async (_args, ctx) => { + await ctx.spawn({ + command: "alchemy", + transformOutput: () => + new TransformStream({ + transform: (chunk, controller) => { + seenByCommand.push(chunk); + controller.enqueue(chunk); + }, + }), + }); + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: command }, + now: CLOCK, + sessions: [ + { + workspaceId: "ws_1", + workspaceName: undefined, + credential: { token, refreshToken: "r", expiresAt: undefined }, + }, + ], + selectedWorkspaceId: "ws_1", + spawnScript: async (request, child) => { + const injected = request.env.PRISMA_SERVICE_TOKEN ?? ""; + const middle = Math.floor(injected.length / 2); + child.write("stdout", `token=${injected.slice(0, middle)}`); + await new Promise((resolve) => setTimeout(resolve, 0)); + child.write("stdout", `${injected.slice(middle)} end`); + child.write("stderr", `again ${injected}`); + return { exitCode: 0, signal: null }; + }, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.stdout).toBe("token=[redacted] end"); + expect(result.stderr).toBe("again [redacted]"); + expect(seenByCommand.join("")).not.toContain(token); + }); + + test("ctx.spawnOptions names transformOutput", async () => { + let names: readonly string[] | undefined; + const command = defineCommand({ + help: { summary: "Checks spawn support" }, + handler: async (_args, ctx) => { + names = ctx.spawnOptions; + return ok( + ctx.present( + { data: null }, + { + human: () => [], + stdout: () => [], + json: () => null, + next: () => [], + }, + ), + ); + }, + }); + const cli = createTestCli({ commands: { check: command }, now: CLOCK }); + + await cli.run(["check"]); + + expect(names).toContain("transformOutput"); + }); + + test("ctx.spawn refuses an option it does not know", async () => { + let caught: unknown; + const command = defineCommand({ + help: { summary: "Passes an unknown spawn option" }, + maySpawn: true, + handler: async (_args, ctx) => { + try { + await ctx.spawn({ + command: "alchemy", + onOutput: () => {}, + } as SpawnOptions); + } catch (cause) { + caught = cause; + } + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ commands: { converge: command }, now: CLOCK }); + + const result = await cli.run(["converge"]); + + expect(result.spawns).toEqual([]); + expect((caught as Error).message).toContain( + "passed ctx.spawn options it does not accept: onOutput", + ); + }); + + test("without transformOutput the default handoffs are unchanged", async () => { const cli = createTestCli({ commands: { converge }, now: CLOCK }); const human = await cli.run(["converge"]); @@ -2074,3 +2300,41 @@ describe("piped child output", () => { expect(json.spawns[0]?.output).toBe("diagnostic"); }); }); + +describe("relayStream", () => { + test("does not read ahead of a destination that has not drained", async () => { + let pulled = 0; + const source: AsyncIterable = { + async *[Symbol.asyncIterator]() { + for (let index = 0; index < 100; index += 1) { + pulled += 1; + yield `chunk ${index}\n`; + } + }, + }; + let release: () => void = () => {}; + const blocked = new Promise((resolve) => { + release = resolve; + }); + const written: string[] = []; + const relaying = relayStream( + source, + [new TransformStream()], + { + write: (text) => { + written.push(text); + }, + drained: () => blocked, + }, + () => {}, + ); + + await new Promise((resolve) => setTimeout(resolve, 20)); + const pulledWhileBlocked = pulled; + release(); + await relaying; + + expect(pulledWhileBlocked).toBeLessThan(10); + expect(written).toHaveLength(100); + }); +}); diff --git a/packages/cli/src/runtime.ts b/packages/cli/src/runtime.ts index 4ba5599f..825f3869 100644 --- a/packages/cli/src/runtime.ts +++ b/packages/cli/src/runtime.ts @@ -1,6 +1,7 @@ import { fstatSync } from "node:fs"; import { fileURLToPath } from "node:url"; import { + type HostOutputStream, type HostProcess, type InputStream, type LoadedConfig, @@ -146,6 +147,26 @@ export function makeStdin(proc: HostProcess): InputStream { }; } +/** A stream that errors or closes instead of draining resolves too, + * so a writer waiting on it is never stuck. */ +function drained(stream: HostOutputStream): Promise { + const { once, off } = stream; + if (stream.writableNeedDrain !== true || once === undefined) { + return Promise.resolve(); + } + return new Promise((resolve) => { + const done = () => { + off?.call(stream, "drain", done); + off?.call(stream, "error", done); + off?.call(stream, "close", done); + resolve(); + }; + once.call(stream, "drain", done); + once.call(stream, "error", done); + once.call(stream, "close", done); + }); +} + export async function assembleRuntime(proc: HostProcess): Promise { const stdin = makeStdin(proc); warnOnDeprecatedStateFileEnvVar(proc); @@ -156,11 +177,13 @@ export async function assembleRuntime(proc: HostProcess): Promise { write: (text) => { proc.stdout.write(text); }, + drained: () => drained(proc.stdout), }, stderr: { write: (text) => { proc.stderr.write(text); }, + drained: () => drained(proc.stderr), get columns() { return proc.stderr.columns; }, diff --git a/packages/cli/src/spawn.ts b/packages/cli/src/spawn.ts index 87f1ba02..1a83dada 100644 --- a/packages/cli/src/spawn.ts +++ b/packages/cli/src/spawn.ts @@ -1,6 +1,7 @@ +import type { ChildProcess } from "node:child_process"; import { type Readable, Writable } from "node:stream"; import { pipeline } from "node:stream/promises"; -import type { ChildResult, SpawnChild } from "@prisma/cli-engine"; +import type { ChildResult, SpawnChild, SpawnedChild } from "@prisma/cli-engine"; import spawn from "cross-spawn"; interface DiagnosticStream { @@ -27,7 +28,7 @@ export interface SpawnChildOptions { /** * The engine's spawn seam, adapted through cross-spawn. Human mode * inherits stdio; structured mode pipes both child output streams to - * diagnostics; pipe mode hands both to the engine. No mode detaches or + * diagnostics; engine mode returns both to the engine. No mode detaches or * opens a new console, so the child stays in this process's group * (POSIX) or console (Windows). * @@ -58,29 +59,8 @@ export function makeSpawnChild( const kill = (signal: "SIGTERM" | "SIGKILL") => { child.kill(signal); }; - if (request.output === "pipe") { - const sources = [child.stdout, child.stderr].filter( - (source): source is Readable => source !== null, - ); - const closed = Promise.all( - sources.map((source) => { - source.setEncoding("utf8"); - return new Promise((resolve) => source.once("close", resolve)); - }), - ); - return { - ended: processEnded.then(async (result) => { - const drainDeadline = setTimeout(() => { - for (const source of sources) source.destroy(); - }, drainGraceMs); - await closed; - clearTimeout(drainDeadline); - return result; - }), - kill, - ...(child.stdout === null ? {} : { stdout: child.stdout }), - ...(child.stderr === null ? {} : { stderr: child.stderr }), - }; + if (request.output === "engine") { + return engineReadChild(child, kill, processEnded, drainGraceMs); } if (!structured) { return { @@ -112,6 +92,54 @@ export function makeSpawnChild( }; } +/** The engine reads both pipes. A grandchild that inherited them can + * hold them open after the child exits, so they are destroyed after + * the drain grace, with an error that says why. */ +function engineReadChild( + child: ChildProcess, + kill: SpawnedChild["kill"], + processEnded: Promise, + drainGraceMs: number, +): SpawnedChild { + const sources = [child.stdout, child.stderr].filter( + (source): source is Readable => source !== null, + ); + const closed = Promise.all( + sources.map((source) => { + source.setEncoding("utf8"); + return new Promise((resolve) => source.once("close", resolve)); + }), + ); + const afterDrain = async (): Promise => { + const drainDeadline = setTimeout(() => { + for (const source of sources) { + source.destroy( + new Error( + `the pipe was still open ${drainGraceMs}ms after the child exited, so it was closed`, + ), + ); + } + }, drainGraceMs); + await closed; + clearTimeout(drainDeadline); + }; + return { + ended: processEnded.then( + async (result) => { + await afterDrain(); + return result; + }, + async (cause: unknown) => { + await afterDrain(); + throw cause; + }, + ), + kill, + ...(child.stdout === null ? {} : { stdout: child.stdout }), + ...(child.stderr === null ? {} : { stderr: child.stderr }), + }; +} + /** Best-effort relay: a forwarding failure never rejects, so the child's * real status still settles the run when the diagnostic sink dies. */ function forwardStructuredOutput( diff --git a/packages/cli/tests/spawn-adapter.test.ts b/packages/cli/tests/spawn-adapter.test.ts index b3abb1cf..a403b7f2 100644 --- a/packages/cli/tests/spawn-adapter.test.ts +++ b/packages/cli/tests/spawn-adapter.test.ts @@ -349,7 +349,7 @@ describe("the shipped spawn adapter", () => { } }); - test("pipe mode hands both decoded streams to the engine and writes nothing itself", async () => { + test("engine mode returns both decoded streams to the engine and writes nothing itself", async () => { const child = spawnChild({ command: NODE, args: [ @@ -358,7 +358,7 @@ describe("the shipped spawn adapter", () => { ], cwd: process.cwd(), env: process.env, - output: "pipe", + output: "engine", }); const [stdout, stderr, result] = await Promise.all([ @@ -375,7 +375,7 @@ describe("the shipped spawn adapter", () => { }); test.skipIf(process.platform === "win32")( - "pipe mode: a grandchild holding the pipes does not block settlement past the drain grace", + "engine mode: a grandchild holding the pipes does not block settlement past the drain grace", async () => { const gracedSpawn = makeSpawnChild( { write: () => {} }, @@ -390,7 +390,7 @@ describe("the shipped spawn adapter", () => { ], cwd: process.cwd(), env: process.env, - output: "pipe", + output: "engine", }); const reading = Promise.all([ readAll(child.stdout).catch(() => ""), diff --git a/packages/cli/tests/spawn-engine-output.test.ts b/packages/cli/tests/spawn-engine-output.test.ts new file mode 100644 index 00000000..af700672 --- /dev/null +++ b/packages/cli/tests/spawn-engine-output.test.ts @@ -0,0 +1,196 @@ +/** + * The engine and the SHIPPED spawn adapter together, with a real child, + * when the command transforms the child's output. + */ +import { + type ChildStreamName, + defineCommand, + exitWithChildStatus, + type SpawnOptions, +} from "@prisma/cli-engine"; +import { ok } from "@prisma/cli-engine/protocol"; +import { createTestCli } from "@prisma/cli-engine/testing"; +import { describe, expect, test } from "vitest"; +import { makeSpawnChild } from "../src/spawn"; + +const NODE = process.execPath; +const TICKS = /^(tick\n)+$/; + +function redacting(secret: string) { + return (_stream: ChildStreamName) => { + let held = ""; + return new TransformStream({ + transform: (chunk, controller) => { + const text = (held + chunk).replaceAll(secret, "[redacted]"); + const keep = Math.min(text.length, secret.length - 1); + held = text.slice(text.length - keep); + if (text.length > keep) { + controller.enqueue(text.slice(0, text.length - keep)); + } + }, + flush: (controller) => { + if (held !== "") { + controller.enqueue(held); + } + }, + }); + }; +} + +function nodeCommand( + script: string, + transformOutput: SpawnOptions["transformOutput"], + onError?: (cause: unknown) => void, +) { + return defineCommand({ + help: { summary: "Runs a node child and transforms its output" }, + maySpawn: true, + handler: async (_args, ctx) => { + try { + await ctx.spawn({ + command: NODE, + args: ["-e", script], + transformOutput, + }); + } catch (cause) { + onError?.(cause); + } + return ok(exitWithChildStatus()); + }, + }); +} + +function cliFor(command: ReturnType) { + return createTestCli({ + commands: { converge: command }, + spawn: makeSpawnChild({ write: () => {} }, { drainGraceMs: 200 }), + }); +} + +const CHATTER = ` +const pause = () => new Promise((resolve) => setTimeout(resolve, 30)); +(async () => { + process.stdout.write("out-1 hunt"); + await pause(); + process.stderr.write("err-1\\n"); + await pause(); + process.stdout.write("er2\\nout-2\\n"); + await pause(); + process.exitCode = 4; +})(); +`; + +describe.skipIf(process.platform === "win32")( + "transformed output through the shipped spawn adapter", + () => { + test("each stream is transformed in order and written where the child would have written it", async () => { + const cli = cliFor(nodeCommand(CHATTER, redacting("hunter2"))); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.spawns[0]?.output).toBe("engine"); + expect(result.exitCode).toBe(4); + expect(result.stdout).toBe("out-1 [redacted]\nout-2\n"); + expect(result.stderr).toBe("err-1\n"); + }); + + test("json mode writes both streams to stderr", async () => { + const cli = cliFor(nodeCommand(CHATTER, redacting("hunter2"))); + + const result = await cli.run(["converge", "--json"]); + + expect(result.json).toHaveLength(1); + expect(result.stderr.replace("err-1\n", "")).toBe( + "out-1 [redacted]\nout-2\n", + ); + expect(result.stderr).toContain("err-1\n"); + }); + + test("an abort ends the child, and its output written so far still arrives", async () => { + const controller = new AbortController(); + const command = defineCommand({ + help: { summary: "Aborts, then runs a child" }, + maySpawn: true, + handler: async (_args, ctx) => { + ctx.report({ kind: "status", subject: "run", status: "ready" }); + await new Promise((resolve) => setTimeout(resolve, 0)); + await ctx.spawn({ + command: NODE, + args: [ + "-e", + "setInterval(() => process.stdout.write('tick\\n'), 10)", + ], + transformOutput: redacting("hunter2"), + }); + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: command }, + spawn: makeSpawnChild({ write: () => {} }, { drainGraceMs: 200 }), + }); + + const result = await cli.run(["converge"], { + abort: controller.signal, + onEvent: (event) => { + if (event.kind === "status") { + controller.abort(); + } + }, + }); + + expect(result.spawns[0]?.kills[0]).toBe("SIGTERM"); + expect(result.exitCode).toBe(143); + expect(result.stdout === "" || TICKS.test(result.stdout)).toBe(true); + }); + + test("a throwing transform ends a child that ignores SIGTERM, and ctx.spawn rejects with its error", async () => { + let caught: unknown; + const script = ` +process.on("SIGTERM", () => {}); +process.stdout.write("ready\\n"); +setInterval(() => process.stdout.write("tick\\n"), 10); +`; + const cli = cliFor( + nodeCommand( + script, + () => + new TransformStream({ + transform: (chunk, controller) => { + if (chunk.includes("tick")) { + throw new Error("redaction failed"); + } + controller.enqueue(chunk); + }, + }), + (cause) => { + caught = cause; + }, + ), + ); + + const result = await cli.run(["converge", "--format", "human"]); + + expect((caught as Error).message).toBe("redaction failed"); + expect(result.spawns[0]?.kills).toEqual(["SIGTERM", "SIGKILL"]); + expect(result.stdout).toBe("ready\n"); + expect(result.exitCode).toBe(137); + }); + + test("output dropped at the drain grace leaves a debug line", async () => { + const script = `require("node:child_process").spawn(process.execPath, ["-e", "setTimeout(() => {}, 10000)"], { detached: true, stdio: ["ignore", "inherit", "ignore"] }).unref(); process.stdout.write("done\\n");`; + const cli = cliFor(nodeCommand(script, redacting("hunter2"))); + + const result = await cli.run(["converge", "--format", "human"], { + env: { PRISMA_DEBUG: "1" }, + }); + + expect(result.exitCode).toBe(0); + expect(result.stdout).toBe("done\n"); + expect(result.stderr).toContain( + "spawn: the child's stdout failed, so its later output was dropped: the pipe was still open 200ms after the child exited, so it was closed", + ); + }); + }, + 20_000, +); From 44526435f962a5ba541d45d5d919e844d81e2022 Mon Sep 17 00:00:00 2001 From: willbot Date: Mon, 2 Nov 2026 17:32:05 +0100 Subject: [PATCH 3/6] fix(engine): transformed child output shows a prompt at once, keeps emoji whole, and survives a slow reader The engine's token redactor held back the last token.length - 1 characters, so a prompt with no newline stayed invisible while the child waited for input. It now holds back only the longest end of the text that is a start of the token. The test helpers had the same shape and are fixed too. The relay now holds a trailing high surrogate until its pair arrives, so no transform can split an emoji across two writes. The adapter closed the pipes 5 seconds after the child exited, even when the engine had stopped reading because the destination was slow, which dropped the child's last output. The deadline now counts only time the engine spends waiting on a pipe with no output, and the debug line says so. SPAWN_OPTION_NAMES is derived from an object checked against SpawnOptions. CommandContext.spawnOptions is renamed spawnOptionNames and is optional, since it exists for hosts where it is absent. The README says the engine redacts its token only when the command passes transformOutput. Signed-off-by: willbot Signed-off-by: Will Madden --- docs/product/output-conventions.md | 2 +- packages/cli-engine/README.md | 10 +- packages/cli-engine/src/context.ts | 2 +- .../src/execution/command-context.ts | 2 +- packages/cli-engine/src/execution/spawn.ts | 56 ++++++- packages/cli-engine/src/spawn.ts | 20 ++- packages/cli-engine/tests/spawn.test.ts | 137 +++++++++++++++++- packages/cli/src/spawn.ts | 83 ++++++++--- .../cli/tests/spawn-engine-output.test.ts | 35 ++++- 9 files changed, 300 insertions(+), 47 deletions(-) diff --git a/docs/product/output-conventions.md b/docs/product/output-conventions.md index a338cade..84e5b9fd 100644 --- a/docs/product/output-conventions.md +++ b/docs/product/output-conventions.md @@ -37,7 +37,7 @@ command: - in human mode, the child inherits stdin, stdout, and stderr - in JSON mode, stdin remains inherited while the child's stdout and stderr are routed to the CLI's diagnostic stderr stream -- when the command transforms the child's output (`transformOutput`), stdin remains inherited and the engine reads the child's stdout and stderr, passes each through the command's transform, and writes the result to the same place the two cases above would have: the child's stdout to stdout and its stderr to stderr in human and markdown format, and both to stderr in JSON mode. The engine also replaces the credential it injected into the child's environment with `[redacted]` +- when the command transforms the child's output (`transformOutput`), stdin remains inherited and the engine reads the child's stdout and stderr, passes each through the command's transform, and writes the result to the same place the two cases above would have: the child's stdout to stdout and its stderr to stderr in human and markdown format, and both to stderr in JSON mode. The engine also replaces the credential it injected into the child's environment with `[redacted]`; without `transformOutput` it does not. - JSON stdout remains an NDJSON event stream with exactly one terminal result frame - a non-zero child status is preserved as the process exit code and is diff --git a/packages/cli-engine/README.md b/packages/cli-engine/README.md index cce367fe..d5bc314c 100644 --- a/packages/cli-engine/README.md +++ b/packages/cli-engine/README.md @@ -80,7 +80,7 @@ Each flag answers one question. A run that succeeds with a flag nothing consumed A command that declares `maySpawn` runs a child process with `ctx.spawn`. By default the child writes straight to the terminal, or, with `--json`, to stderr. To change the text before it is written, for example to redact secrets, pass `transformOutput`: ```ts -if (!ctx.spawnOptions?.includes("transformOutput")) { +if (!ctx.spawnOptionNames?.includes("transformOutput")) { throw new Error("This command needs @prisma/cli-engine 0.10.0 or later."); } const result = await ctx.spawn({ @@ -95,15 +95,15 @@ The engine calls `transformOutput` once for `"stdout"` and once for `"stderr"` b - human and markdown format: the child's stdout to stdout, its stderr to stderr; - json format: both to stderr, so stdout carries only the engine's NDJSON output. -Each stream keeps its own order; there is no order between the two. The engine reads the child only as fast as the terminal takes the text, so a slow reader slows the child down instead of filling memory. When a stream ends, the transform's `flush` runs and its output is written before `ctx.spawn` settles, including when the child was aborted or failed. A transform can therefore hold text back between chunks, as a redactor that matches across chunk boundaries must. +Each stream keeps its own order; there is no order between the two. The engine reads the child only as fast as the terminal takes the text, so a slow reader slows the child down instead of filling memory. When a stream ends, the transform's `flush` runs and its output is written before `ctx.spawn` settles, including when the child was aborted or failed. A transform can therefore hold text back between chunks, as a redactor that matches across chunk boundaries must. Hold back only what could still become a secret, the longest end of the text that is a start of one, and not a fixed number of characters: a child that prints a prompt with no newline and waits for input must show the prompt at once. The engine never splits a UTF-16 surrogate pair across two writes, so a transform that cuts text by length does not corrupt an emoji. -When the command declares `needs: { credentials: "child" }`, the engine puts the access token in the child's `PRISMA_SERVICE_TOKEN`. The command never sees that value, so the engine replaces it with `[redacted]` itself, before your transform runs. +When the command declares `needs: { credentials: "child" }`, the engine puts the access token in the child's `PRISMA_SERVICE_TOKEN`. The command never sees that value, so when the command passes `transformOutput`, the engine replaces the token with `[redacted]` itself, before your transform runs. Without `transformOutput` the engine does not read the child's output and does not redact the token, in any format, json included. -The engine still owns the child. Stdin stays inherited, Ctrl-C reaches the child as before, and a SIGTERM or an abort of `ctx.signal` ends it with SIGTERM, then SIGKILL after a grace period. If a transform throws or errors, the engine stops reading that stream, ends the child the same way, and `ctx.spawn` rejects with that error once the child has exited. With `PRISMA_DEBUG=1`, a stream that fails, or that a grandchild keeps open after the child exits, leaves a line saying its later output was dropped. +The engine still owns the child. Stdin stays inherited, Ctrl-C reaches the child as before, and a SIGTERM or an abort of `ctx.signal` ends it with SIGTERM, then SIGKILL after a grace period. If a transform throws or errors, the engine stops reading that stream, ends the child the same way, and `ctx.spawn` rejects with that error once the child has exited. With `PRISMA_DEBUG=1`, a stream that fails, or that a grandchild keeps open with no output after the child exits, leaves a line saying its later output was dropped. A slow reader does not cause this: the grace period counts only time the engine spends waiting for output that does not come. ### Checking for support -Hosts older than 0.10.0 ignore spawn options they do not know, so on those hosts a child would run with its output untransformed. `ctx.spawnOptions` lists the option names `ctx.spawn` accepts; it is absent on those hosts. Check it before spawning, as above. From 0.10.0 on, `ctx.spawn` refuses any option name it does not know, so a future option cannot be ignored silently. +Hosts older than 0.10.0 ignore spawn options they do not know, so on those hosts a child would run with its output untransformed. `ctx.spawnOptionNames` lists the option names `ctx.spawn` accepts; it is absent on those hosts, and its type is optional so the compiler makes you check it. Check it before spawning, as above. From 0.10.0 on, `ctx.spawn` refuses any option name it does not know, so a future option cannot be ignored silently. ### Spawn adapters diff --git a/packages/cli-engine/src/context.ts b/packages/cli-engine/src/context.ts index 37df427a..570e311b 100644 --- a/packages/cli-engine/src/context.ts +++ b/packages/cli-engine/src/context.ts @@ -93,7 +93,7 @@ export interface CommandContext< * options they do not know, so a command that depends on an option * checks for it here before spawning. */ - readonly spawnOptions: readonly (keyof SpawnOptions)[]; + readonly spawnOptionNames?: readonly (keyof SpawnOptions)[]; /** * How the run's most recent completed child ended, or undefined when diff --git a/packages/cli-engine/src/execution/command-context.ts b/packages/cli-engine/src/execution/command-context.ts index 49e2d1dc..b8ea105e 100644 --- a/packages/cli-engine/src/execution/command-context.ts +++ b/packages/cli-engine/src/execution/command-context.ts @@ -179,7 +179,7 @@ export function makeContext( return api; }, spawn: makeSpawn(invocation, def), - spawnOptions: SPAWN_OPTION_NAMES, + spawnOptionNames: SPAWN_OPTION_NAMES, lastChild: () => state.lastChild, report: (event) => reportEvent(invocation, event), prompt: makePromptSurface(invocation), diff --git a/packages/cli-engine/src/execution/spawn.ts b/packages/cli-engine/src/execution/spawn.ts index 7e1f56d3..19dc4153 100644 --- a/packages/cli-engine/src/execution/spawn.ts +++ b/packages/cli-engine/src/execution/spawn.ts @@ -337,7 +337,7 @@ export async function relayStream( destination: OutputStream, onSourceError: (cause: unknown) => void, ): Promise { - await transforms + await [...transforms, wholeCharacters()] .reduce( (readable, transform) => readable.pipeThrough(transform), ReadableStream.from(endOnError(source, onSourceError)), @@ -365,18 +365,18 @@ function injectedSecrets( return token === undefined || token === "" ? [] : [token]; } -/** Replaces `secret` with "[redacted]" across chunk boundaries by - * holding back up to `secret.length - 1` characters until the next - * chunk or the end of the stream. */ +/** Replaces `secret` with "[redacted]" across chunk boundaries. It + * holds back only an end of the text that could still become the + * secret, so a prompt with no newline is written at once. */ function redactingTransform(secret: string): TransformStream { let held = ""; return new TransformStream({ transform: (chunk, controller) => { const text = (held + chunk).replaceAll(secret, "[redacted]"); - const keep = Math.min(text.length, secret.length - 1); - held = text.slice(text.length - keep); - if (text.length > keep) { - controller.enqueue(text.slice(0, text.length - keep)); + const cut = text.length - partialMatchLength(text, secret); + held = text.slice(cut); + if (cut > 0) { + controller.enqueue(text.slice(0, cut)); } }, flush: (controller) => { @@ -387,6 +387,46 @@ function redactingTransform(secret: string): TransformStream { }); } +/** The length of the longest end of `text` that is a start of `secret`. */ +function partialMatchLength(text: string, secret: string): number { + for ( + let length = Math.min(text.length, secret.length - 1); + length > 0; + length -= 1 + ) { + if (text.endsWith(secret.slice(0, length))) { + return length; + } + } + return 0; +} + +/** Holds a trailing high surrogate until its pair arrives: a lone half + * written on its own is encoded as U+FFFD. */ +function wholeCharacters(): TransformStream { + let held = ""; + return new TransformStream({ + transform: (chunk, controller) => { + const text = held + chunk; + const cut = endsWithHighSurrogate(text) ? text.length - 1 : text.length; + held = text.slice(cut); + if (cut > 0) { + controller.enqueue(text.slice(0, cut)); + } + }, + flush: (controller) => { + if (held !== "") { + controller.enqueue(held); + } + }, + }); +} + +function endsWithHighSurrogate(text: string): boolean { + const last = text.charCodeAt(text.length - 1); + return last >= 0xd800 && last <= 0xdbff; +} + async function* endOnError( source: AsyncIterable, onError: (cause: unknown) => void, diff --git a/packages/cli-engine/src/spawn.ts b/packages/cli-engine/src/spawn.ts index 9a0088f9..e284191e 100644 --- a/packages/cli-engine/src/spawn.ts +++ b/packages/cli-engine/src/spawn.ts @@ -48,9 +48,11 @@ export interface SpawnedChild { readonly kill: (signal: "SIGTERM" | "SIGKILL") => void; /** The child's piped output, decoded as UTF-8. Required when the * request's `output` is "engine", absent otherwise. Each ends when - * the child closes it; an adapter ends both no later than a grace - * period after the child exits, so a grandchild holding the pipe - * open cannot hold the run. */ + * the child closes it. After the child exits, an adapter ends a + * stream that the engine has waited on for a grace period with no + * output, so a grandchild holding the pipe open cannot hold the run. + * Time the engine spends not reading does not count, so a slow + * destination does not lose the child's last output. */ readonly stdout?: AsyncIterable; readonly stderr?: AsyncIterable; } @@ -86,16 +88,24 @@ export interface SpawnOptions { * engine replaces the credential it injects into the child's * environment with `[redacted]` before the transform sees the text. * Hosts older than 0.10.0 ignore this option: check - * `ctx.spawnOptions?.includes("transformOutput")` first. + * `ctx.spawnOptionNames?.includes("transformOutput")` first. */ readonly transformOutput?: ( stream: ChildStreamName, ) => TransformStream; } +const ACCEPTED_SPAWN_OPTIONS = { + command: true, + args: true, + cwd: true, + env: true, + transformOutput: true, +} as const satisfies Record; + /** Every option ctx.spawn accepts; it refuses any other. */ export const SPAWN_OPTION_NAMES: readonly (keyof SpawnOptions)[] = - Object.freeze(["command", "args", "cwd", "env", "transformOutput"]); + Object.freeze(Object.keys(ACCEPTED_SPAWN_OPTIONS) as (keyof SpawnOptions)[]); export const CHILD_STATUS: unique symbol = Symbol.for( "@prisma/cli-engine.childStatus", diff --git a/packages/cli-engine/tests/spawn.test.ts b/packages/cli-engine/tests/spawn.test.ts index b62d0e82..4f9e5214 100644 --- a/packages/cli-engine/tests/spawn.test.ts +++ b/packages/cli-engine/tests/spawn.test.ts @@ -1811,15 +1811,18 @@ function controllableRuntime() { }; } -/** Replaces `secret` across chunk boundaries by holding back up to - * `secret.length - 1` characters, which flush writes at the end. */ +/** Replaces `secret` across chunk boundaries by holding back only an + * end of the text that could still become the secret. */ function redacting(secret: string): () => TransformStream { return () => { let held = ""; return new TransformStream({ transform: (chunk, controller) => { const text = (held + chunk).replaceAll(secret, "[redacted]"); - const keep = Math.min(text.length, secret.length - 1); + let keep = Math.min(text.length, secret.length - 1); + while (keep > 0 && !text.endsWith(secret.slice(0, keep))) { + keep -= 1; + } held = text.slice(text.length - keep); if (text.length > keep) { controller.enqueue(text.slice(0, text.length - keep)); @@ -2237,12 +2240,66 @@ describe("transformed child output", () => { expect(seenByCommand.join("")).not.toContain(token); }); - test("ctx.spawnOptions names transformOutput", async () => { + test("a prompt with no newline reaches the command's transform before the child exits", async () => { + const token = jwtExpiringIn(3600, "ws_1"); + let promptSeen: () => void = () => {}; + const seen = new Promise((resolve) => { + promptSeen = () => resolve(true); + }); + let seenBeforeExit = false; + const command = defineCommand({ + help: { summary: "Hands credentials to a child that asks a question" }, + maySpawn: true, + needs: { credentials: "child" }, + handler: async (_args, ctx) => { + await ctx.spawn({ + command: "alchemy", + transformOutput: () => + new TransformStream({ + transform: (chunk, controller) => { + if (chunk.includes("(y/N) ")) { + promptSeen(); + } + controller.enqueue(chunk); + }, + }), + }); + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: command }, + now: CLOCK, + sessions: [ + { + workspaceId: "ws_1", + workspaceName: undefined, + credential: { token, refreshToken: "r", expiresAt: undefined }, + }, + ], + selectedWorkspaceId: "ws_1", + spawnScript: async (_request, child) => { + child.write("stdout", "Deploy to production? (y/N) "); + seenBeforeExit = await Promise.race([ + seen, + new Promise((resolve) => setTimeout(resolve, 500, false)), + ]); + return { exitCode: 0, signal: null }; + }, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(seenBeforeExit).toBe(true); + expect(result.stdout).toBe("Deploy to production? (y/N) "); + }); + + test("ctx.spawnOptionNames names transformOutput", async () => { let names: readonly string[] | undefined; const command = defineCommand({ help: { summary: "Checks spawn support" }, handler: async (_args, ctx) => { - names = ctx.spawnOptions; + names = ctx.spawnOptionNames; return ok( ctx.present( { data: null }, @@ -2302,6 +2359,76 @@ describe("transformed child output", () => { }); describe("relayStream", () => { + test("writes a prompt with no newline at once through a redacting transform", async () => { + let finish: () => void = () => {}; + const finished = new Promise((resolve) => { + finish = resolve; + }); + const source: AsyncIterable = { + async *[Symbol.asyncIterator]() { + yield "Deploy to production? (y/N) "; + await finished; + yield "token hun"; + yield "ter2\n"; + }, + }; + const written: string[] = []; + const relaying = relayStream( + source, + [redacting("hunter2")()], + { + write: (text) => { + written.push(text); + }, + }, + () => {}, + ); + + await new Promise((resolve) => setTimeout(resolve, 20)); + const writtenBeforeEnd = written.join(""); + finish(); + await relaying; + + expect(writtenBeforeEnd).toBe("Deploy to production? (y/N) "); + expect(written.join("")).toBe( + "Deploy to production? (y/N) token [redacted]\n", + ); + }); + + test("never splits a surrogate pair across two writes", async () => { + const source: AsyncIterable = { + async *[Symbol.asyncIterator]() { + yield "hello \u{1F600}"; + yield " bye"; + }, + }; + const splitAtSeven = new TransformStream({ + transform: (chunk, controller) => { + controller.enqueue(chunk.slice(0, 7)); + if (chunk.length > 7) { + controller.enqueue(chunk.slice(7)); + } + }, + }); + const written: string[] = []; + + await relayStream( + source, + [splitAtSeven], + { + write: (text) => { + written.push(text); + }, + }, + () => {}, + ); + + expect(written.join("")).toBe("hello \u{1F600} bye"); + for (const text of written) { + expect(Buffer.from(text, "utf8").toString("utf8")).toBe(text); + } + }); + test("does not read ahead of a destination that has not drained", async () => { let pulled = 0; const source: AsyncIterable = { diff --git a/packages/cli/src/spawn.ts b/packages/cli/src/spawn.ts index 1a83dada..c2bae5b1 100644 --- a/packages/cli/src/spawn.ts +++ b/packages/cli/src/spawn.ts @@ -93,14 +93,20 @@ export function makeSpawnChild( } /** The engine reads both pipes. A grandchild that inherited them can - * hold them open after the child exits, so they are destroyed after - * the drain grace, with an error that says why. */ + * hold them open after the child exits, so a pipe the engine has + * waited on with no output for the drain grace is closed, with an + * error that says why. Time the engine spends not reading, because + * its destination is slow, does not count. */ function engineReadChild( child: ChildProcess, kill: SpawnedChild["kill"], processEnded: Promise, drainGraceMs: number, ): SpawnedChild { + const exited = processEnded.then( + () => undefined, + () => undefined, + ); const sources = [child.stdout, child.stderr].filter( (source): source is Readable => source !== null, ); @@ -110,33 +116,72 @@ function engineReadChild( return new Promise((resolve) => source.once("close", resolve)); }), ); - const afterDrain = async (): Promise => { - const drainDeadline = setTimeout(() => { - for (const source of sources) { - source.destroy( - new Error( - `the pipe was still open ${drainGraceMs}ms after the child exited, so it was closed`, - ), - ); - } - }, drainGraceMs); - await closed; - clearTimeout(drainDeadline); - }; + const readable = (source: Readable | null) => + source === null ? undefined : closeWhenIdle(source, exited, drainGraceMs); + const stdout = readable(child.stdout); + const stderr = readable(child.stderr); return { ended: processEnded.then( async (result) => { - await afterDrain(); + await closed; return result; }, async (cause: unknown) => { - await afterDrain(); + await closed; throw cause; }, ), kill, - ...(child.stdout === null ? {} : { stdout: child.stdout }), - ...(child.stderr === null ? {} : { stderr: child.stderr }), + ...(stdout === undefined ? {} : { stdout }), + ...(stderr === undefined ? {} : { stderr }), + }; +} + +function closeWhenIdle( + source: Readable, + exited: Promise, + drainGraceMs: number, +): AsyncIterable { + let childExited = false; + let waiting = false; + let deadline: ReturnType | undefined; + const arm = () => { + if (childExited && waiting && deadline === undefined) { + deadline = setTimeout(() => { + source.destroy( + new Error( + `the pipe stayed open with no output for ${drainGraceMs}ms after the child exited, so it was closed`, + ), + ); + }, drainGraceMs); + } + }; + const disarm = () => { + clearTimeout(deadline); + deadline = undefined; + }; + void exited.then(() => { + childExited = true; + arm(); + }); + return { + [Symbol.asyncIterator]: () => { + const chunks: AsyncIterator = source[Symbol.asyncIterator](); + return { + next: async () => { + waiting = true; + arm(); + try { + return await chunks.next(); + } finally { + waiting = false; + disarm(); + } + }, + return: async () => + (await chunks.return?.()) ?? { done: true, value: undefined }, + }; + }, }; } diff --git a/packages/cli/tests/spawn-engine-output.test.ts b/packages/cli/tests/spawn-engine-output.test.ts index af700672..a45a300a 100644 --- a/packages/cli/tests/spawn-engine-output.test.ts +++ b/packages/cli/tests/spawn-engine-output.test.ts @@ -22,7 +22,10 @@ function redacting(secret: string) { return new TransformStream({ transform: (chunk, controller) => { const text = (held + chunk).replaceAll(secret, "[redacted]"); - const keep = Math.min(text.length, secret.length - 1); + let keep = Math.min(text.length, secret.length - 1); + while (keep > 0 && !text.endsWith(secret.slice(0, keep))) { + keep -= 1; + } held = text.slice(text.length - keep); if (text.length > keep) { controller.enqueue(text.slice(0, text.length - keep)); @@ -188,9 +191,37 @@ setInterval(() => process.stdout.write("tick\\n"), 10); expect(result.exitCode).toBe(0); expect(result.stdout).toBe("done\n"); expect(result.stderr).toContain( - "spawn: the child's stdout failed, so its later output was dropped: the pipe was still open 200ms after the child exited, so it was closed", + "spawn: the child's stdout failed, so its later output was dropped: the pipe stayed open with no output for 200ms after the child exited, so it was closed", ); }); + + test("a slow reader still gets the child's last output after the drain grace", async () => { + const script = `process.stdout.write("x".repeat(300_000) + "LAST-LINE\\n");`; + const child = makeSpawnChild( + { write: () => {} }, + { drainGraceMs: 200 }, + )({ + command: NODE, + args: ["-e", script], + cwd: process.cwd(), + env: process.env, + output: "engine", + }); + let received = ""; + for await (const chunk of child.stdout ?? []) { + received += chunk; + await new Promise((resolve) => setTimeout(resolve, 150)); + } + let errors = ""; + for await (const chunk of child.stderr ?? []) { + errors += chunk; + } + + expect(await child.ended).toEqual({ exitCode: 0, signal: null }); + expect(errors).toBe(""); + expect(received.length).toBe(300_000 + "LAST-LINE\n".length); + expect(received.endsWith("LAST-LINE\n")).toBe(true); + }); }, 20_000, ); From d71da9bac7d15ab23665fa8b165eb8cf4729e05d Mon Sep 17 00:00:00 2001 From: willbot Date: Mon, 2 Nov 2026 18:04:08 +0100 Subject: [PATCH 4/6] fix(engine): transformed child output works under Bun, which has no ReadableStream.from Bun 1.3.13 has no ReadableStream.from, so every ctx.spawn with transformOutput failed under Bun with CLI.INTERNAL_ERROR. The relay now builds the stream with new ReadableStream({ pull }) and a high-water mark of zero, which reads the child only when the pipe asks for a chunk, as before. The other APIs on this path (TransformStream, WritableStream, pipeThrough, pipeTo, AbortSignal.any) exist in Bun 1.3.13. Signed-off-by: willbot Signed-off-by: Will Madden --- packages/cli-engine/src/execution/spawn.ts | 24 ++++++++++++++++- packages/cli-engine/tests/spawn.test.ts | 31 ++++++++++++++++++++++ 2 files changed, 54 insertions(+), 1 deletion(-) diff --git a/packages/cli-engine/src/execution/spawn.ts b/packages/cli-engine/src/execution/spawn.ts index 19dc4153..6ceeca1c 100644 --- a/packages/cli-engine/src/execution/spawn.ts +++ b/packages/cli-engine/src/execution/spawn.ts @@ -340,7 +340,7 @@ export async function relayStream( await [...transforms, wholeCharacters()] .reduce( (readable, transform) => readable.pipeThrough(transform), - ReadableStream.from(endOnError(source, onSourceError)), + readableFrom(endOnError(source, onSourceError)), ) .pipeTo( new WritableStream({ @@ -352,6 +352,28 @@ export async function relayStream( ); } +/** ReadableStream.from, which Bun does not have. A high-water mark of + * zero reads the source only when the pipe asks for a chunk. */ +function readableFrom(source: AsyncIterable): ReadableStream { + const chunks = source[Symbol.asyncIterator](); + return new ReadableStream( + { + pull: async (controller) => { + const next = await chunks.next(); + if (next.done === true) { + controller.close(); + } else { + controller.enqueue(next.value); + } + }, + cancel: async () => { + await chunks.return?.(); + }, + }, + { highWaterMark: 0 }, + ); +} + /** The credential the engine put in the child's environment. The * command never sees it, so only the engine can redact it. */ function injectedSecrets( diff --git a/packages/cli-engine/tests/spawn.test.ts b/packages/cli-engine/tests/spawn.test.ts index 4f9e5214..cfe73c02 100644 --- a/packages/cli-engine/tests/spawn.test.ts +++ b/packages/cli-engine/tests/spawn.test.ts @@ -2395,6 +2395,37 @@ describe("relayStream", () => { ); }); + test("works where ReadableStream.from does not exist, as in Bun", async () => { + const from = Object.getOwnPropertyDescriptor(ReadableStream, "from"); + Reflect.deleteProperty(ReadableStream, "from"); + try { + const source: AsyncIterable = { + async *[Symbol.asyncIterator]() { + yield "one hun"; + yield "ter2 two\n"; + }, + }; + const written: string[] = []; + + await relayStream( + source, + [redacting("hunter2")()], + { + write: (text) => { + written.push(text); + }, + }, + () => {}, + ); + + expect(written.join("")).toBe("one [redacted] two\n"); + } finally { + if (from !== undefined) { + Object.defineProperty(ReadableStream, "from", from); + } + } + }); + test("never splits a surrogate pair across two writes", async () => { const source: AsyncIterable = { async *[Symbol.asyncIterator]() { From 955a8dea00bfdeb1d4d65ea58783aeebd557b7ff Mon Sep 17 00:00:00 2001 From: willbot Date: Mon, 2 Nov 2026 18:18:03 +0100 Subject: [PATCH 5/6] fix(cli): a failing output stream stops a transformed child, and the spawn tests check what they claim The runtime's drained() now rejects when the host stream emits error, so the relay stops and ctx.spawn rejects instead of writing on to a failed stream. A stream that closes still resolves it. A host stream with no off() is never waited on, so no error or close listener is left behind. Tests: drained() stays pending until drain and handles close and error, through assembleRuntime. The shipped-adapter abort test sends SIGTERM only after the child's first line reaches the transform, and requires that line in stdout. A transform whose promise rejects later ends the child like one that throws. The real-child tests check each stream's text, not the order between the streams, and the abort test waits for the child's pid file before its grace runs out. Signed-off-by: willbot Signed-off-by: Will Madden --- packages/cli-engine/README.md | 2 +- packages/cli-engine/src/runtime.ts | 13 ++- .../cli-engine/tests/spawn-real-child.test.ts | 13 ++- packages/cli-engine/tests/spawn.test.ts | 70 ++++++++++++ packages/cli/src/runtime.ts | 38 ++++--- packages/cli/tests/runtime-drained.test.ts | 101 ++++++++++++++++++ .../cli/tests/spawn-engine-output.test.ts | 48 +++------ 7 files changed, 232 insertions(+), 53 deletions(-) create mode 100644 packages/cli/tests/runtime-drained.test.ts diff --git a/packages/cli-engine/README.md b/packages/cli-engine/README.md index d5bc314c..fd434f78 100644 --- a/packages/cli-engine/README.md +++ b/packages/cli-engine/README.md @@ -99,7 +99,7 @@ Each stream keeps its own order; there is no order between the two. The engine r When the command declares `needs: { credentials: "child" }`, the engine puts the access token in the child's `PRISMA_SERVICE_TOKEN`. The command never sees that value, so when the command passes `transformOutput`, the engine replaces the token with `[redacted]` itself, before your transform runs. Without `transformOutput` the engine does not read the child's output and does not redact the token, in any format, json included. -The engine still owns the child. Stdin stays inherited, Ctrl-C reaches the child as before, and a SIGTERM or an abort of `ctx.signal` ends it with SIGTERM, then SIGKILL after a grace period. If a transform throws or errors, the engine stops reading that stream, ends the child the same way, and `ctx.spawn` rejects with that error once the child has exited. With `PRISMA_DEBUG=1`, a stream that fails, or that a grandchild keeps open with no output after the child exits, leaves a line saying its later output was dropped. A slow reader does not cause this: the grace period counts only time the engine spends waiting for output that does not come. +The engine still owns the child. Stdin stays inherited, Ctrl-C reaches the child as before, and a SIGTERM or an abort of `ctx.signal` ends it with SIGTERM, then SIGKILL after a grace period. If a transform throws or errors, or the destination fails, the engine stops reading that stream, ends the child the same way, and `ctx.spawn` rejects with that error once the child has exited. With `PRISMA_DEBUG=1`, a stream that fails, or that a grandchild keeps open with no output after the child exits, leaves a line saying its later output was dropped. A slow reader does not cause this: the grace period counts only time the engine spends waiting for output that does not come. ### Checking for support diff --git a/packages/cli-engine/src/runtime.ts b/packages/cli-engine/src/runtime.ts index 216002dc..62b2976b 100644 --- a/packages/cli-engine/src/runtime.ts +++ b/packages/cli-engine/src/runtime.ts @@ -13,7 +13,8 @@ export interface OutputStream { * terminal resized mid-run is respected by the next thing drawn. */ readonly columns?: number; /** Resolves once the stream can take more text without buffering it - * in memory. Absent means the stream never asks the writer to wait. */ + * in memory, and rejects when the stream fails. Absent means the + * stream never asks the writer to wait. */ readonly drained?: () => Promise; } @@ -164,8 +165,14 @@ export interface HostOutputStream { write(text: string): unknown; isTTY?: boolean; writableNeedDrain?: boolean; - once?(event: "drain" | "error" | "close", listener: () => void): unknown; - off?(event: "drain" | "error" | "close", listener: () => void): unknown; + once?( + event: "drain" | "error" | "close", + listener: (cause?: unknown) => void, + ): unknown; + off?( + event: "drain" | "error" | "close", + listener: (cause?: unknown) => void, + ): unknown; } /** diff --git a/packages/cli-engine/tests/spawn-real-child.test.ts b/packages/cli-engine/tests/spawn-real-child.test.ts index 82f033c8..7639addb 100644 --- a/packages/cli-engine/tests/spawn-real-child.test.ts +++ b/packages/cli-engine/tests/spawn-real-child.test.ts @@ -321,11 +321,13 @@ describe.skipIf(process.platform === "win32")( const result = await cli.run(["converge", "--format", "human"]); expect(result.exitCode).toBe(4); - expect(chunks).toEqual([ - ["stdout", "out-1\n"], - ["stderr", "err-1\n"], - ["stdout", "out-2\n"], - ]); + const text = (stream: ChildStreamName) => + chunks + .filter(([name]) => name === stream) + .map(([, chunk]) => chunk) + .join(""); + expect(text("stdout")).toBe("out-1\nout-2\n"); + expect(text("stderr")).toBe("err-1\n"); expect(result.stdout).toBe("out-1\nout-2\n"); expect(result.stderr).toBe("err-1\n"); }); @@ -353,6 +355,7 @@ describe.skipIf(process.platform === "win32")( commands: { converge: watcher }, spawn: termAfterReady(pipingSpawn, pidFile), delay: async () => { + await waitForFile(pidFile); await new Promise((resolve) => setTimeout(resolve, 300)); }, }); diff --git a/packages/cli-engine/tests/spawn.test.ts b/packages/cli-engine/tests/spawn.test.ts index cfe73c02..6530b6ae 100644 --- a/packages/cli-engine/tests/spawn.test.ts +++ b/packages/cli-engine/tests/spawn.test.ts @@ -2060,6 +2060,51 @@ describe("transformed child output", () => { expect((caught as Error).message).toBe("redaction failed"); }); + test("a transform whose promise rejects later ends the child, and ctx.spawn rejects with its error", async () => { + let caught: unknown; + const command = defineCommand({ + help: { summary: "Fails asynchronously while transforming" }, + maySpawn: true, + handler: async (_args, ctx) => { + try { + await ctx.spawn({ + command: "alchemy", + transformOutput: () => + new TransformStream({ + transform: async (chunk, controller) => { + await new Promise((resolve) => setTimeout(resolve, 0)); + if (chunk === "second") { + throw new Error("redaction failed later"); + } + controller.enqueue(chunk); + }, + }), + }); + } catch (cause) { + caught = cause; + } + return ok(exitWithChildStatus()); + }, + }); + const cli = createTestCli({ + commands: { converge: command }, + now: CLOCK, + delay: (_ms, signal) => signalDone(signal), + spawnScript: async (_request, child) => { + child.write("stdout", "first"); + child.write("stdout", "second"); + await child.nextKill(); + return { exitCode: null, signal: "SIGTERM" }; + }, + }); + + const result = await cli.run(["converge", "--format", "human"]); + + expect(result.spawns[0]?.kills).toEqual(["SIGTERM"]); + expect(result.stdout).toBe("first"); + expect((caught as Error).message).toBe("redaction failed later"); + }); + test("a throwing transformOutput rejects ctx.spawn before any child starts", async () => { let caught: unknown; const command = defineCommand({ @@ -2426,6 +2471,31 @@ describe("relayStream", () => { } }); + test("rejects when the destination fails while the relay waits for it to drain", async () => { + const source: AsyncIterable = { + async *[Symbol.asyncIterator]() { + yield "one\n"; + yield "two\n"; + }, + }; + const written: string[] = []; + + await expect( + relayStream( + source, + [], + { + write: (text) => { + written.push(text); + }, + drained: () => Promise.reject(new Error("EPIPE")), + }, + () => {}, + ), + ).rejects.toThrow("EPIPE"); + expect(written).toEqual(["one\n"]); + }); + test("never splits a surrogate pair across two writes", async () => { const source: AsyncIterable = { async *[Symbol.asyncIterator]() { diff --git a/packages/cli/src/runtime.ts b/packages/cli/src/runtime.ts index 825f3869..c88f764a 100644 --- a/packages/cli/src/runtime.ts +++ b/packages/cli/src/runtime.ts @@ -147,23 +147,37 @@ export function makeStdin(proc: HostProcess): InputStream { }; } -/** A stream that errors or closes instead of draining resolves too, - * so a writer waiting on it is never stuck. */ +/** A stream that closes instead of draining resolves too, so a writer + * waiting on it is never stuck. A stream that errors rejects, so the + * writer stops instead of writing to a failed stream. A stream that + * cannot remove its listeners is never waited on. */ function drained(stream: HostOutputStream): Promise { const { once, off } = stream; - if (stream.writableNeedDrain !== true || once === undefined) { + if ( + stream.writableNeedDrain !== true || + once === undefined || + off === undefined + ) { return Promise.resolve(); } - return new Promise((resolve) => { - const done = () => { - off?.call(stream, "drain", done); - off?.call(stream, "error", done); - off?.call(stream, "close", done); - resolve(); + return new Promise((resolve, reject) => { + const settle = (outcome: () => void) => { + off.call(stream, "drain", onDrain); + off.call(stream, "error", onError); + off.call(stream, "close", onDrain); + outcome(); }; - once.call(stream, "drain", done); - once.call(stream, "error", done); - once.call(stream, "close", done); + const onDrain = () => { + settle(resolve); + }; + const onError = (cause?: unknown) => { + settle(() => { + reject(cause instanceof Error ? cause : new Error(String(cause))); + }); + }; + once.call(stream, "drain", onDrain); + once.call(stream, "error", onError); + once.call(stream, "close", onDrain); }); } diff --git a/packages/cli/tests/runtime-drained.test.ts b/packages/cli/tests/runtime-drained.test.ts new file mode 100644 index 00000000..eefc071a --- /dev/null +++ b/packages/cli/tests/runtime-drained.test.ts @@ -0,0 +1,101 @@ +import { EventEmitter } from "node:events"; +import { describe, expect, test } from "vitest"; +import { assembleRuntime, type HostProcess } from "../src/runtime"; + +class HostStream extends EventEmitter { + writableNeedDrain = true; + write(_text: string): boolean { + return false; + } +} + +async function runtimeWith(stdout: HostStream) { + const proc: HostProcess = { + argv: ["node", "bin.js"], + version: "v22.18.0", + versions: { node: "22.18.0" }, + platform: "linux", + arch: "x64", + env: {}, + cwd: () => "/projects/acme", + stdout, + stderr: new HostStream(), + stdin: { + async *[Symbol.asyncIterator]() {}, + }, + on: () => undefined, + off: () => undefined, + exit: (code: number): never => { + throw new Error(`process.exit(${code}) reached the test`); + }, + }; + return await assembleRuntime(proc); +} + +async function settledWithin(promise: Promise, ms: number) { + let settled = false; + void promise.then( + () => { + settled = true; + }, + () => { + settled = true; + }, + ); + await new Promise((resolve) => setTimeout(resolve, ms)); + return settled; +} + +const LISTENED = ["drain", "error", "close"] as const; + +describe("runtime.stdout.drained", () => { + test("stays pending until the host stream emits drain, then resolves", async () => { + const stdout = new HostStream(); + const runtime = await runtimeWith(stdout); + + const waiting = runtime.stdout.drained?.() ?? Promise.resolve(); + + expect(await settledWithin(waiting, 20)).toBe(false); + stdout.writableNeedDrain = false; + stdout.emit("drain"); + await expect(waiting).resolves.toBeUndefined(); + for (const event of LISTENED) { + expect(stdout.listenerCount(event)).toBe(0); + } + }); + + test("resolves when the host stream closes instead of draining", async () => { + const stdout = new HostStream(); + const runtime = await runtimeWith(stdout); + + const waiting = runtime.stdout.drained?.() ?? Promise.resolve(); + stdout.emit("close"); + + await expect(waiting).resolves.toBeUndefined(); + for (const event of LISTENED) { + expect(stdout.listenerCount(event)).toBe(0); + } + }); + + test("rejects when the host stream fails", async () => { + const stdout = new HostStream(); + const runtime = await runtimeWith(stdout); + + const waiting = runtime.stdout.drained?.() ?? Promise.resolve(); + stdout.emit("error", new Error("EPIPE")); + + await expect(waiting).rejects.toThrow("EPIPE"); + for (const event of LISTENED) { + expect(stdout.listenerCount(event)).toBe(0); + } + }); + + test("resolves at once when the host stream does not need to drain", async () => { + const stdout = new HostStream(); + stdout.writableNeedDrain = false; + const runtime = await runtimeWith(stdout); + + await expect(runtime.stdout.drained?.()).resolves.toBeUndefined(); + expect(stdout.listenerCount("drain")).toBe(0); + }); +}); diff --git a/packages/cli/tests/spawn-engine-output.test.ts b/packages/cli/tests/spawn-engine-output.test.ts index a45a300a..ad0d18d3 100644 --- a/packages/cli/tests/spawn-engine-output.test.ts +++ b/packages/cli/tests/spawn-engine-output.test.ts @@ -14,7 +14,7 @@ import { describe, expect, test } from "vitest"; import { makeSpawnChild } from "../src/spawn"; const NODE = process.execPath; -const TICKS = /^(tick\n)+$/; +const READY_THEN_TICKS = /^ready\n(tick\n)*$/; function redacting(secret: string) { return (_stream: ChildStreamName) => { @@ -109,42 +109,26 @@ describe.skipIf(process.platform === "win32")( expect(result.stderr).toContain("err-1\n"); }); - test("an abort ends the child, and its output written so far still arrives", async () => { + test("a SIGTERM ends the child, and its output written so far still arrives", async () => { const controller = new AbortController(); - const command = defineCommand({ - help: { summary: "Aborts, then runs a child" }, - maySpawn: true, - handler: async (_args, ctx) => { - ctx.report({ kind: "status", subject: "run", status: "ready" }); - await new Promise((resolve) => setTimeout(resolve, 0)); - await ctx.spawn({ - command: NODE, - args: [ - "-e", - "setInterval(() => process.stdout.write('tick\\n'), 10)", - ], - transformOutput: redacting("hunter2"), - }); - return ok(exitWithChildStatus()); - }, - }); - const cli = createTestCli({ - commands: { converge: command }, - spawn: makeSpawnChild({ write: () => {} }, { drainGraceMs: 200 }), - }); + const abortOnReady = (_stream: ChildStreamName) => + new TransformStream({ + transform: (chunk, transformController) => { + transformController.enqueue(chunk); + if (chunk.includes("ready\n")) { + controller.abort("SIGTERM"); + } + }, + }); + const script = + "process.stdout.write('ready\\n'); setInterval(() => process.stdout.write('tick\\n'), 10);"; + const cli = cliFor(nodeCommand(script, abortOnReady)); - const result = await cli.run(["converge"], { - abort: controller.signal, - onEvent: (event) => { - if (event.kind === "status") { - controller.abort(); - } - }, - }); + const result = await cli.run(["converge"], { abort: controller.signal }); expect(result.spawns[0]?.kills[0]).toBe("SIGTERM"); expect(result.exitCode).toBe(143); - expect(result.stdout === "" || TICKS.test(result.stdout)).toBe(true); + expect(result.stdout).toMatch(READY_THEN_TICKS); }); test("a throwing transform ends a child that ignores SIGTERM, and ctx.spawn rejects with its error", async () => { From 01a2072685ade04db16e36b894103186be6348ab Mon Sep 17 00:00:00 2001 From: willbot Date: Mon, 2 Nov 2026 18:27:22 +0100 Subject: [PATCH 6/6] fix(engine): a host stream that can wait for drain must also remove its listeners HostOutputStream now has once and off as a pair, so the type refuses a stream that could wait for drain but not clean up after it. The bin's drained() therefore waits on every stream that has once, which keeps backpressure, and never leaves a listener behind. The drained tests run against both stdout and stderr, each with its own host stream. The README says the debug line covers only a failed read of the child's output, not a transform or destination failure. Signed-off-by: willbot Signed-off-by: Will Madden --- packages/cli-engine/README.md | 2 +- packages/cli-engine/src/runtime.ts | 24 ++++----- packages/cli/src/runtime.ts | 9 +--- packages/cli/tests/runtime-drained.test.ts | 57 ++++++++++++---------- 4 files changed, 48 insertions(+), 44 deletions(-) diff --git a/packages/cli-engine/README.md b/packages/cli-engine/README.md index fd434f78..eb661c54 100644 --- a/packages/cli-engine/README.md +++ b/packages/cli-engine/README.md @@ -99,7 +99,7 @@ Each stream keeps its own order; there is no order between the two. The engine r When the command declares `needs: { credentials: "child" }`, the engine puts the access token in the child's `PRISMA_SERVICE_TOKEN`. The command never sees that value, so when the command passes `transformOutput`, the engine replaces the token with `[redacted]` itself, before your transform runs. Without `transformOutput` the engine does not read the child's output and does not redact the token, in any format, json included. -The engine still owns the child. Stdin stays inherited, Ctrl-C reaches the child as before, and a SIGTERM or an abort of `ctx.signal` ends it with SIGTERM, then SIGKILL after a grace period. If a transform throws or errors, or the destination fails, the engine stops reading that stream, ends the child the same way, and `ctx.spawn` rejects with that error once the child has exited. With `PRISMA_DEBUG=1`, a stream that fails, or that a grandchild keeps open with no output after the child exits, leaves a line saying its later output was dropped. A slow reader does not cause this: the grace period counts only time the engine spends waiting for output that does not come. +The engine still owns the child. Stdin stays inherited, Ctrl-C reaches the child as before, and a SIGTERM or an abort of `ctx.signal` ends it with SIGTERM, then SIGKILL after a grace period. If a transform throws or errors, or the destination fails, the engine stops reading that stream, ends the child the same way, and `ctx.spawn` rejects with that error once the child has exited. With `PRISMA_DEBUG=1`, a child output stream whose read fails, or that a grandchild keeps open with no output after the child exits, leaves a line saying its later output was dropped. A transform or destination failure leaves no such line, because `ctx.spawn` rejects with its error. A slow reader does not cause this: the grace period counts only time the engine spends waiting for output that does not come. ### Checking for support diff --git a/packages/cli-engine/src/runtime.ts b/packages/cli-engine/src/runtime.ts index 62b2976b..1771b70b 100644 --- a/packages/cli-engine/src/runtime.ts +++ b/packages/cli-engine/src/runtime.ts @@ -159,21 +159,23 @@ export interface Host { readonly arch: string; } +type HostStreamEvent = "drain" | "error" | "close"; +type HostStreamListener = (cause?: unknown) => void; + /** The parts of a Node writable the bin reads off process.stdout and - * process.stderr. */ -export interface HostOutputStream { + * process.stderr. A stream that can wait for drain has both `once` and + * `off`, so waiting never leaves a listener behind. */ +export type HostOutputStream = { write(text: string): unknown; isTTY?: boolean; writableNeedDrain?: boolean; - once?( - event: "drain" | "error" | "close", - listener: (cause?: unknown) => void, - ): unknown; - off?( - event: "drain" | "error" | "close", - listener: (cause?: unknown) => void, - ): unknown; -} +} & ( + | { + once(event: HostStreamEvent, listener: HostStreamListener): unknown; + off(event: HostStreamEvent, listener: HostStreamListener): unknown; + } + | { once?: undefined; off?: undefined } +); /** * The minimal process surface a bin adapts a Runtime from — Node's diff --git a/packages/cli/src/runtime.ts b/packages/cli/src/runtime.ts index c88f764a..3b4a7b1c 100644 --- a/packages/cli/src/runtime.ts +++ b/packages/cli/src/runtime.ts @@ -149,15 +149,10 @@ export function makeStdin(proc: HostProcess): InputStream { /** A stream that closes instead of draining resolves too, so a writer * waiting on it is never stuck. A stream that errors rejects, so the - * writer stops instead of writing to a failed stream. A stream that - * cannot remove its listeners is never waited on. */ + * writer stops instead of writing to a failed stream. */ function drained(stream: HostOutputStream): Promise { const { once, off } = stream; - if ( - stream.writableNeedDrain !== true || - once === undefined || - off === undefined - ) { + if (stream.writableNeedDrain !== true || once === undefined) { return Promise.resolve(); } return new Promise((resolve, reject) => { diff --git a/packages/cli/tests/runtime-drained.test.ts b/packages/cli/tests/runtime-drained.test.ts index eefc071a..a7662cba 100644 --- a/packages/cli/tests/runtime-drained.test.ts +++ b/packages/cli/tests/runtime-drained.test.ts @@ -9,7 +9,11 @@ class HostStream extends EventEmitter { } } -async function runtimeWith(stdout: HostStream) { +function hostStreams() { + return { stdout: new HostStream(), stderr: new HostStream() }; +} + +async function runtimeWith(streams: ReturnType) { const proc: HostProcess = { argv: ["node", "bin.js"], version: "v22.18.0", @@ -18,8 +22,8 @@ async function runtimeWith(stdout: HostStream) { arch: "x64", env: {}, cwd: () => "/projects/acme", - stdout, - stderr: new HostStream(), + stdout: streams.stdout, + stderr: streams.stderr, stdin: { async *[Symbol.asyncIterator]() {}, }, @@ -48,54 +52,57 @@ async function settledWithin(promise: Promise, ms: number) { const LISTENED = ["drain", "error", "close"] as const; -describe("runtime.stdout.drained", () => { +describe.each(["stdout", "stderr"] as const)("runtime.%s.drained", (name) => { test("stays pending until the host stream emits drain, then resolves", async () => { - const stdout = new HostStream(); - const runtime = await runtimeWith(stdout); + const streams = hostStreams(); + const host = streams[name]; + const runtime = await runtimeWith(streams); - const waiting = runtime.stdout.drained?.() ?? Promise.resolve(); + const waiting = runtime[name].drained?.() ?? Promise.resolve(); expect(await settledWithin(waiting, 20)).toBe(false); - stdout.writableNeedDrain = false; - stdout.emit("drain"); + host.writableNeedDrain = false; + host.emit("drain"); await expect(waiting).resolves.toBeUndefined(); for (const event of LISTENED) { - expect(stdout.listenerCount(event)).toBe(0); + expect(host.listenerCount(event)).toBe(0); } }); test("resolves when the host stream closes instead of draining", async () => { - const stdout = new HostStream(); - const runtime = await runtimeWith(stdout); + const streams = hostStreams(); + const host = streams[name]; + const runtime = await runtimeWith(streams); - const waiting = runtime.stdout.drained?.() ?? Promise.resolve(); - stdout.emit("close"); + const waiting = runtime[name].drained?.() ?? Promise.resolve(); + host.emit("close"); await expect(waiting).resolves.toBeUndefined(); for (const event of LISTENED) { - expect(stdout.listenerCount(event)).toBe(0); + expect(host.listenerCount(event)).toBe(0); } }); test("rejects when the host stream fails", async () => { - const stdout = new HostStream(); - const runtime = await runtimeWith(stdout); + const streams = hostStreams(); + const host = streams[name]; + const runtime = await runtimeWith(streams); - const waiting = runtime.stdout.drained?.() ?? Promise.resolve(); - stdout.emit("error", new Error("EPIPE")); + const waiting = runtime[name].drained?.() ?? Promise.resolve(); + host.emit("error", new Error("EPIPE")); await expect(waiting).rejects.toThrow("EPIPE"); for (const event of LISTENED) { - expect(stdout.listenerCount(event)).toBe(0); + expect(host.listenerCount(event)).toBe(0); } }); test("resolves at once when the host stream does not need to drain", async () => { - const stdout = new HostStream(); - stdout.writableNeedDrain = false; - const runtime = await runtimeWith(stdout); + const streams = hostStreams(); + streams[name].writableNeedDrain = false; + const runtime = await runtimeWith(streams); - await expect(runtime.stdout.drained?.()).resolves.toBeUndefined(); - expect(stdout.listenerCount("drain")).toBe(0); + await expect(runtime[name].drained?.()).resolves.toBeUndefined(); + expect(streams[name].listenerCount("drain")).toBe(0); }); });