From 91294c275e7bae4bcdf5150d9fe18c70a03e5767 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Tue, 1 Sep 2026 21:03:18 -0400 Subject: [PATCH 1/6] =?UTF-8?q?=F0=9F=90=9B=20Make=20piped=20`xmd=20syntax?= =?UTF-8?q?`=20output=20complete,=20and=20report=20a=20closed=20sink=20(#7?= =?UTF-8?q?15)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `xmd syntax` waits for stdout to accept the whole catalog, so a pipe receives exactly what a regular-file redirect receives however slowly it is read. A sink that closes mid-write is now reported on stderr with exit status 1 rather than ending the process with an unhandled write failure: a broken pipe arrives twice — at the write callback, and again as an `error` event a tick later — so the listener outlives the write. Three rows in `packages/cli/tests/syntax-cli.test.ts` prove it through a real pipeline against an oversize catalog. `runShell()` and `cliShellCommand()` are the test-support seam that composes one. --- packages/cli/src/cli.ts | 32 +++++-- packages/cli/tests/syntax-cli.test.ts | 121 +++++++++++++++++++++++++- packages/test-support/launch.ts | 59 +++++++++++-- specs/executable-mdx-spec.md | 7 ++ 4 files changed, 201 insertions(+), 18 deletions(-) diff --git a/packages/cli/src/cli.ts b/packages/cli/src/cli.ts index c0eb24594..15fdeec51 100755 --- a/packages/cli/src/cli.ts +++ b/packages/cli/src/cli.ts @@ -2429,7 +2429,13 @@ function* dispatch( yield* exit(1); break; } - yield* writeStdoutWhole(rendered); + const written = yield* writeStdoutWhole(rendered); + if (!written.ok) { + console.error( + `xmd syntax: stdout did not accept the whole catalog: ${describeError(written.error)}`, + ); + yield* exit(1); + } break; } case "test-agent": @@ -2540,13 +2546,29 @@ function* dispatch( * 64 KiB, in the middle of a token. Waiting for the callback is what makes the * write finish before anything can exit. * + * A sink that closes mid-write is the failure this reports rather than raises. + * A broken pipe arrives twice — once at this callback, and again as an `error` + * event on `process.stdout` a tick later — so the listener stays attached after + * the outcome is settled. Removing it would leave that second arrival + * unhandled, and an unhandled `error` event ends the process with a stack trace + * before the command can say which output was cut short. + * * Only the catalog goes through this today, because it is the one output this * command writes in a single call and the only one already past the buffer. */ -function* writeStdoutWhole(text: string): Operation { - yield* until( - new Promise((resolve, reject) => { - process.stdout.write(text, (error) => (error ? reject(error) : resolve())); +function* writeStdoutWhole(text: string): Operation> { + return yield* until( + new Promise>((resolve) => { + let settled = false; + const settle = (error?: Error | null) => { + if (settled) { + return; + } + settled = true; + resolve(error ? Err(error) : Ok()); + }; + process.stdout.on("error", settle); + process.stdout.write(text, settle); }), ); } diff --git a/packages/cli/tests/syntax-cli.test.ts b/packages/cli/tests/syntax-cli.test.ts index 909b7b7e7..d47fff3f1 100644 --- a/packages/cli/tests/syntax-cli.test.ts +++ b/packages/cli/tests/syntax-cli.test.ts @@ -3,16 +3,16 @@ * * Two halves, matching the command's own. The profile rows run in process, * because what they check is which declarations `xmd run` installs — a claim - * about assembly, not about argv. The grammar and failure rows shell out, so - * exit status, stdout and stderr are the ones an operator sees. + * about assembly, not about argv. The grammar, failure and delivery rows shell + * out, so exit status, stdout and stderr are the ones an operator sees. * * Catalog behavior itself is Tier SY's; nothing here re-proves selection. */ import { describe, it } from "@executablemd/test-support/bdd"; import { expect } from "@executablemd/test-support/expect"; -import { runCli } from "@executablemd/test-support/launch"; -import { ensureDir, rm, writeTextFile } from "@effectionx/fs"; +import { cliShellCommand, runCli, runShell, shellQuote } from "@executablemd/test-support/launch"; +import { ensureDir, readTextFile, rm, writeTextFile } from "@effectionx/fs"; import { ensure, scoped, until } from "effection"; import type { Operation } from "effection"; import { symlink } from "node:fs/promises"; @@ -463,3 +463,116 @@ describe("Tier SX — the command line", { sanitizeOps: false, sanitizeResources expect(stdout).toMatch(/^\s+plan\s/m); }); }); + +/** + * Tier SX — what a pipe receives. + * + * These rows shell out to a real pipeline rather than capturing a stream this + * process owns, because the subject is the boundary itself: an operating-system + * pipe holds about 64 KiB, and a catalog handed to a fire-and-forget + * `process.stdout.write()` ends mid-token there while the same invocation + * redirected to a file is whole. Nothing smaller than an oversize catalog + * through an actual pipe tells the two apart. + * + * The two completeness rows are answered by Node and Bun, where that write + * truncates at 80 and 64 KiB. Deno's own `process.stdout` flushes a pipe on the + * way out, so it cannot distinguish them — a change verified under Deno alone + * has not been verified. The closing-consumer row fails under all three. + */ +describe( + "Tier SX — the catalog a pipe receives", + { sanitizeOps: false, sanitizeResources: false }, + () => { + it("SX13: the Markdown form arrives whole, and equals a file redirect", function* () { + yield* useWorkspace(OVERSIZE, function* (cwd) { + const { redirected, piped } = yield* deliveries([], cwd); + + expect(piped).toBe(redirected); + expect(piped.lastIndexOf("### ``")).toBeGreaterThan(PIPE_BUFFER); + }); + }); + + it("SX14: the JSON form arrives whole, parses, and equals a file redirect", function* () { + yield* useWorkspace(OVERSIZE, function* (cwd) { + const { redirected, piped } = yield* deliveries(["--json"], cwd); + + expect(piped).toBe(redirected); + const catalog = parseCatalog(piped); + expect(catalog.version).toBe(1); + expect(names(catalog.categories[2].entries)).toContain("ZBeyondTheBuffer"); + expect(piped.lastIndexOf(`"ZBeyondTheBuffer"`)).toBeGreaterThan(PIPE_BUFFER); + }); + }); + + it("SX15: a consumer that closes early fails the command", function* () { + yield* useWorkspace(OVERSIZE, function* (cwd) { + // A pipeline reports its last stage's status, so the command's own + // travels on stderr, which the closing consumer never held. + const { stdout, stderr } = yield* runShell( + `{ ${cliShellCommand(["syntax", "--json"])}; echo "xmd-exit=$?" >&2; } | head -c 100`, + { cwd }, + ).join(); + + expect(stdout.length).toBe(100); + expect(stderr).toContain("xmd-exit=1"); + expect(stderr).toContain("stdout did not accept the whole catalog"); + // The broken pipe is reported, not raised: an unhandled write failure + // ends the process with one of these instead. + expect(stderr).not.toContain("Unhandled 'error' event"); + expect(stderr).not.toContain("Uncaught"); + }); + }); + }, +); + +/** What an operating-system pipe holds before a writer has to wait. */ +const PIPE_BUFFER = 64 * 1024; + +/** + * A workspace whose catalog is past that buffer in both forms. + * + * The built-ins alone render about 89 KiB of JSON but only about 63 KiB of + * Markdown, so the padding is what puts the *default* form past the boundary + * too. `ZBeyondTheBuffer` sorts after every filler, which is how a row names + * bytes that a truncated delivery could not contain. + */ +const OVERSIZE: Record = { + ...Object.fromEntries( + Array.from({ length: 8 }, (_unused, index) => [ + `components/Filler${index}.md`, + described(`filler ${index} ${"padding ".repeat(220)}`), + ]), + ), + "components/ZBeyondTheBuffer.md": described("the entry past the pipe buffer."), +}; + +function described(description: string): string { + return `---\ndescription: ${description}\n---\n\nbody\n`; +} + +/** + * A reader that takes one line at a time, so the writer blocks on a full pipe + * long before the catalog ends. It reproduces its input byte for byte: both + * renderers end every line, `IFS=` keeps the surrounding whitespace, and `-r` + * keeps the backslashes. + */ +const SLOW_READER = `while IFS= read -r line; do printf '%s\\n' "$line"; done`; + +/** The same invocation delivered twice: to a regular file, and through a pipe. */ +function* deliveries( + form: string[], + cwd: string, +): Operation<{ redirected: string; piped: string }> { + const command = cliShellCommand(["syntax", ...form]); + const direct = join(cwd, "direct.out"); + const through = join(cwd, "piped.out"); + + yield* runShell(`${command} > ${shellQuote(direct)}`, { cwd }).expect(); + const redirected = yield* readTextFile(direct); + // Without this the comparison proves nothing: a catalog that fits in one + // pipe buffer arrives whole however it was written. + expect(redirected.length).toBeGreaterThan(PIPE_BUFFER); + + yield* runShell(`${command} | (${SLOW_READER}) > ${shellQuote(through)}`, { cwd }).expect(); + return { redirected, piped: yield* readTextFile(through) }; +} diff --git a/packages/test-support/launch.ts b/packages/test-support/launch.ts index 3fa6b528e..4b606b81e 100644 --- a/packages/test-support/launch.ts +++ b/packages/test-support/launch.ts @@ -46,6 +46,23 @@ export function cliCommand(args: string[]): { command: string; arguments: string return { command, arguments: [...prefix, ...args] }; } +/** + * The same command as one quoted shell word list. + * + * A suite whose subject is what a *pipeline* delivers — a regular-file + * redirect, a reader that closes early — needs the CLI as text it can compose + * around, and the runtime it belongs to is still this package's to know. + */ +export function cliShellCommand(args: string[]): string { + const cli = cliCommand(args); + return [cli.command, ...cli.arguments].map(shellQuote).join(" "); +} + +/** One shell word, quoted — a path a composed line names goes through here too. */ +export function shellQuote(word: string): string { + return `'${word.replaceAll("'", `'\\''`)}'`; +} + /** * What a subprocess needs from this one: where to find executables, and where * each runtime caches what it downloads. `HOME` is deliberately absent — a run @@ -84,9 +101,26 @@ export interface CliRun { * timeout are configured here so every suite launches the same way. */ export function runCli(args: string[], options: CliRunOptions = {}): CliRun { + const launch = cliCommand(args); + const label = `xmd ${args.join(" ")}`; + return { + join: () => bounded(launch, label, options, "join"), + expect: () => bounded(launch, label, options, "expect"), + }; +} + +/** + * Run a shell line the caller composed, in the environment `runCli` uses. + * + * Compose it from `cliShellCommand()`. The exit status reported is the shell's, + * which in a pipeline is the last stage's — a row that needs the CLI's own + * status reports it out of band. + */ +export function runShell(line: string, options: CliRunOptions = {}): CliRun { + const launch = { command: line, shell: true }; return { - join: () => bounded(args, options, "join"), - expect: () => bounded(args, options, "expect"), + join: () => bounded(launch, line, options, "join"), + expect: () => bounded(launch, line, options, "expect"), }; } @@ -96,20 +130,27 @@ interface PartialOutput { stderr: string; } +interface Launch { + command: string; + arguments?: string[]; + shell?: boolean; +} + function* bounded( - args: string[], + launch: Launch, + label: string, options: CliRunOptions, mode: "join" | "expect", ): Operation { const limit = options.timeout ?? DEFAULT_TIMEOUT; - const cli = cliCommand(args); // Accumulated outside the deadline, so a run the deadline abandons still has // an account: a timeout that reports nothing but its duration cannot say // whether the child hung before its first line or after its last. const partial: PartialOutput = { stdout: "", stderr: "" }; const result = yield* timebox(limit, function* () { - const child = yield* exec(cli.command, { - arguments: cli.arguments, + const child = yield* exec(launch.command, { + arguments: launch.arguments, + shell: launch.shell, cwd: options.cwd, env: cliEnv(options), }); @@ -129,7 +170,7 @@ function* bounded( return { ...status, stdout: partial.stdout, stderr: partial.stderr }; }); if (result.timeout) { - throw new Error(abandonedReport(args, limit, partial)); + throw new Error(abandonedReport(label, limit, partial)); } return result.value; } @@ -145,12 +186,12 @@ function text(bytes: Uint8Array): string { * expire under contention, not by the child's own doing — and whatever each * channel received before the run was abandoned. */ -function abandonedReport(args: string[], limit: number, partial: PartialOutput): string { +function abandonedReport(label: string, limit: number, partial: PartialOutput): string { const load = loadavg() .map((average) => average.toFixed(1)) .join(", "); return [ - `xmd ${args.join(" ")} timed out after ${limit}ms (load average ${load})`, + `${label} timed out after ${limit}ms (load average ${load})`, channel("stdout", partial.stdout), channel("stderr", partial.stderr), ].join("\n"); diff --git a/specs/executable-mdx-spec.md b/specs/executable-mdx-spec.md index b32d0cfbc..0dd1a17ca 100644 --- a/specs/executable-mdx-spec.md +++ b/specs/executable-mdx-spec.md @@ -3119,6 +3119,13 @@ and `--json`. It takes no document and no run option, because it runs nothing. An inspection failure is reported on stderr with exit status 1 and no partial catalog on stdout. +**The command has not succeeded until stdout has taken the whole catalog.** A +pipe holds far less than a catalog, so backpressure changes how long the command +takes and never which bytes arrive: a reader that consumes slowly receives +exactly what a regular-file redirect receives, in both forms. A sink that closes +or refuses the write is reported on stderr with exit status 1, rather than +succeeding with what happened to fit. + #### Validating a supplied document: `validateDocument()` Inspection answers about a name and about a directory. `validateDocument()` From 92f0cbeafb55d499c4a9db5032bbdbf6a1452778 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Tue, 1 Sep 2026 21:29:00 -0400 Subject: [PATCH 2/6] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Bound=20the=20stdout?= =?UTF-8?q?=20delivery=20listener=20to=20one=20delivery=20(#715)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The listener `process.stdout.on("error", …)` installed was never removed, so it outlived the operation that owned it: repeated invocations would accumulate listeners on a process-global stream, and a completed delivery could absorb a later, unrelated stdout failure. Delivery moves to `stdout-delivery.ts` behind a narrow `DeliverySink`, and its listener now lives for exactly one delivery. A try/finally detaches on every completion path — success, either failure order, cancellation, and a `write` that refuses outright. The one path that waits is a failed arrival whose paired event is still owed, and `errored` holding this delivery's own failure is what says so: Node and Deno destroy the stream with the very error they are about to emit, while Bun reports the failure once and holds nothing. Waiting unconditionally would hang there. Tier SD covers the lifetime a real pipe cannot show. SX13–SX15 and the test-support pipeline seam are unchanged. --- packages/cli/src/cli.ts | 43 +---- packages/cli/src/stdout-delivery.ts | 103 +++++++++++ packages/cli/tests/stdout-delivery.test.ts | 192 +++++++++++++++++++++ specs/executable-mdx-spec.md | 15 ++ 4 files changed, 315 insertions(+), 38 deletions(-) create mode 100644 packages/cli/src/stdout-delivery.ts create mode 100644 packages/cli/tests/stdout-delivery.test.ts diff --git a/packages/cli/src/cli.ts b/packages/cli/src/cli.ts index 15fdeec51..3f2a97321 100755 --- a/packages/cli/src/cli.ts +++ b/packages/cli/src/cli.ts @@ -130,6 +130,7 @@ import { runUpgrade } from "./upgrade.ts"; import type { UpgradeAssembly } from "./upgrade.ts"; import { componentSearchPath, resolveTestTarget } from "./test-target.ts"; import { renderSyntaxJson, renderSyntaxMarkdown, syntaxCatalog } from "./syntax.ts"; +import { deliverWhole } from "./stdout-delivery.ts"; import { testingExecutionHost } from "./testing-host.ts"; import { unsupportedRepositories } from "./run-repositories.ts"; import type { RepositoryInstaller } from "./run-repositories.ts"; @@ -2429,7 +2430,10 @@ function* dispatch( yield* exit(1); break; } - const written = yield* writeStdoutWhole(rendered); + // Only the catalog goes through delivery today, because it is the one + // output this command writes in a single call and the only one already + // past a pipe buffer. + const written = yield* deliverWhole(rendered, process.stdout); if (!written.ok) { console.error( `xmd syntax: stdout did not accept the whole catalog: ${describeError(written.error)}`, @@ -2536,43 +2540,6 @@ function* dispatch( } } -/** - * Write to stdout and wait for it to reach the operating system. - * - * `process.stdout` is asynchronous when it is a pipe and synchronous when it is - * a file or a terminal. A large document handed to `write` is therefore still - * sitting in a buffer when the run ends, and the process exits without it: - * `xmd syntax --json > file` is whole, `xmd syntax --json | jq` stops at about - * 64 KiB, in the middle of a token. Waiting for the callback is what makes the - * write finish before anything can exit. - * - * A sink that closes mid-write is the failure this reports rather than raises. - * A broken pipe arrives twice — once at this callback, and again as an `error` - * event on `process.stdout` a tick later — so the listener stays attached after - * the outcome is settled. Removing it would leave that second arrival - * unhandled, and an unhandled `error` event ends the process with a stack trace - * before the command can say which output was cut short. - * - * Only the catalog goes through this today, because it is the one output this - * command writes in a single call and the only one already past the buffer. - */ -function* writeStdoutWhole(text: string): Operation> { - return yield* until( - new Promise>((resolve) => { - let settled = false; - const settle = (error?: Error | null) => { - if (settled) { - return; - } - settled = true; - resolve(error ? Err(error) : Ok()); - }; - process.stdout.on("error", settle); - process.stdout.write(text, settle); - }), - ); -} - export function* runXmd( args: string[], installService: HostServiceInstaller, diff --git a/packages/cli/src/stdout-delivery.ts b/packages/cli/src/stdout-delivery.ts new file mode 100644 index 000000000..949d68d04 --- /dev/null +++ b/packages/cli/src/stdout-delivery.ts @@ -0,0 +1,103 @@ +/** + * Handing one rendered result to a stream, and waiting for the stream to take + * all of it. + * + * A stream is asynchronous when it is a pipe and synchronous when it is a file + * or a terminal. A large text handed to `write` is therefore still sitting in a + * buffer when a run ends, and the process exits without it: `xmd syntax --json + * > file` is whole, `xmd syntax --json | jq` stops at about 64 KiB, in the + * middle of a token. Waiting for the write callback is what makes the write + * finish before anything can exit. + */ + +import { Err, Ok, until } from "effection"; +import type { Operation, Result } from "effection"; + +/** + * What delivery asks of a stream. + * + * Narrower than a writable stream so a suite can supply the arrival orders a + * real pipe produces on one runtime and not another, and can watch the + * listener come and go. + */ +export interface DeliverySink { + write(text: string, callback: (error?: Error | null) => void): unknown; + on(event: "error", listener: (error: Error) => void): unknown; + off(event: "error", listener: (error: Error) => void): unknown; + /** The error the stream was destroyed with and has not emitted yet. */ + readonly errored?: Error | null; +} + +/** + * Deliver `text`, and report whether the stream took all of it. + * + * A broken pipe can arrive twice: at the write callback, and again as an + * `error` event. An `error` event nobody is listening for ends the process with + * a stack trace, which is why this listens — and the first arrival is the + * verdict, whichever it was, so the duplicate is absorbed rather than reported + * a second time. + * + * **The listener lives for one delivery and no longer.** It survives a first + * failed arrival only until the paired one lands, and `errored` is what says + * whether one is still owed: Node and Deno destroy the stream with the very + * error they are about to emit, so finding this delivery's own failure there + * means the event is still coming, while Bun reports the failure once and holds + * nothing. Every other path — success, cancellation, a `write` that refuses + * outright — detaches without waiting at all. + */ +export function* deliverWhole(text: string, sink: DeliverySink): Operation> { + let observe: (error: Error) => void = () => {}; + try { + return yield* until( + new Promise>((resolve) => { + let settled = false; + let calledBack = false; + let observed = false; + let failure: Error | undefined; + + const settle = () => { + if (settled || !calledBack) { + return; + } + // Identity, not presence: an error the stream was already holding + // before this delivery has been emitted already, and waiting for it + // would wait forever. + if (!observed && failure !== undefined && sink.errored === failure) { + return; + } + settled = true; + resolve(failure === undefined ? Ok() : Err(failure)); + }; + + const record = (error?: Error | null) => { + if (error && failure === undefined) { + failure = error; + } + }; + + observe = (error: Error) => { + observed = true; + record(error); + settle(); + }; + + sink.on("error", observe); + + try { + sink.write(text, (error) => { + calledBack = true; + record(error); + settle(); + }); + } catch (error) { + // A stream that refuses the call outright never calls back, so + // nothing else will settle this. + settled = true; + resolve(Err(error instanceof Error ? error : new Error(String(error)))); + } + }), + ); + } finally { + sink.off("error", observe); + } +} diff --git a/packages/cli/tests/stdout-delivery.test.ts b/packages/cli/tests/stdout-delivery.test.ts new file mode 100644 index 000000000..8226112bf --- /dev/null +++ b/packages/cli/tests/stdout-delivery.test.ts @@ -0,0 +1,192 @@ +/** + * Tier SD — delivering a rendered result to a stream. + * + * Tier SX proves what a real pipe receives. These rows prove the *lifetime* of + * the one listener delivery installs, which a real pipe cannot show: a stream + * is a process-global object, so a listener that outlived its delivery would + * accumulate across invocations and could absorb a later, unrelated failure. + * + * The sink is an `EventEmitter`, and that is load-bearing rather than + * convenient: emitting `error` with nothing listening throws. A row that emits + * the trailing event therefore fails outright if delivery has already detached, + * instead of quietly asserting nothing. + */ + +import { describe, it } from "@executablemd/test-support/bdd"; +import { expect } from "@executablemd/test-support/expect"; +import { scoped, spawn, withResolvers } from "effection"; +import { EventEmitter } from "node:events"; +import process from "node:process"; +import { deliverWhole } from "../src/stdout-delivery.ts"; +import type { DeliverySink } from "../src/stdout-delivery.ts"; + +/** What the stream does once delivery hands it the text. */ +type SinkPlan = (sink: RecordingSink, callback: (error?: Error | null) => void) => void; + +class RecordingSink extends EventEmitter implements DeliverySink { + /** The error a destroyed stream holds until it emits it. */ + errored: Error | null = null; + readonly written: string[] = []; + readonly started = withResolvers(); + readonly emitted = withResolvers(); + /** How many listeners the trailing event found, which is the whole question. */ + listenersAtEmit = 0; + + constructor(private readonly plan: SinkPlan) { + super(); + } + + write(text: string, callback: (error?: Error | null) => void): boolean { + this.written.push(text); + this.started.resolve(); + this.plan(this, callback); + return true; + } + + emitTrailing(error: Error): void { + this.listenersAtEmit = this.listenerCount("error"); + this.emit("error", error); + this.emitted.resolve(); + } +} + +const EPIPE = new Error("write EPIPE"); +const OTHER = new Error("a later, unrelated stdout failure"); + +/** A stream that takes everything, the way a file redirect does. */ +const ACCEPTS: SinkPlan = (_sink, callback) => callback(); + +/** + * Node and Deno: the callback carries the failure, and the stream holds that + * same error until it emits it. + * + * The event is scheduled rather than emitted here, because a turn is exactly + * what separates the two arrivals on a real pipe — long enough for a delivery + * that settled at the callback to have resumed and detached already. + */ +const CALLBACK_THEN_EVENT: SinkPlan = (sink, callback) => { + sink.errored = EPIPE; + callback(EPIPE); + setTimeout(() => { + sink.errored = null; + sink.emitTrailing(EPIPE); + }, 0); +}; + +/** The reverse arrival order, with a distinct second error to rank the two. */ +const EVENT_THEN_CALLBACK: SinkPlan = (sink, callback) => { + sink.emit("error", EPIPE); + callback(OTHER); +}; + +/** Bun: the failure is reported once, and the stream holds nothing after it. */ +const REPORTED_ONCE: SinkPlan = (_sink, callback) => callback(EPIPE); + +/** A stream still working, or one that never answers at all. */ +const NEVER_ANSWERS: SinkPlan = () => {}; + +const REFUSES: SinkPlan = () => { + throw EPIPE; +}; + +function listeners(sink: EventEmitter): number { + return sink.listenerCount("error"); +} + +describe("Tier SD — the listener lives for one delivery", () => { + it("SD1: a delivery that succeeds leaves the stream as it found it", function* () { + const sink = new RecordingSink(ACCEPTS); + const before = listeners(sink); + + const result = yield* deliverWhole("catalog", sink); + + expect(result.ok).toBe(true); + expect(sink.written).toEqual(["catalog"]); + expect(listeners(sink)).toBe(before); + }); + + it("SD2: the trailing event still finds the listener, which is gone after it", function* () { + const sink = new RecordingSink(CALLBACK_THEN_EVENT); + // A sentinel, so that detaching too early is this row's assertion rather + // than an unhandled `error` event thrown from a timer. + sink.on("error", () => {}); + const before = listeners(sink); + + const result = yield* deliverWhole("catalog", sink); + yield* sink.emitted.operation; + + expect(result.ok).toBe(false); + expect(result.ok ? undefined : result.error).toBe(EPIPE); + // Delivery was still observing when the event landed, and only then let go. + expect(sink.listenersAtEmit).toBe(before + 1); + expect(listeners(sink)).toBe(before); + }); + + it("SD3: the event may arrive first, and the first arrival is the verdict", function* () { + const sink = new RecordingSink(EVENT_THEN_CALLBACK); + const before = listeners(sink); + + const result = yield* deliverWhole("catalog", sink); + + expect(result.ok ? undefined : result.error).toBe(EPIPE); + expect(listeners(sink)).toBe(before); + }); + + it("SD4: a failure the stream reports once settles without waiting for a second", function* () { + const sink = new RecordingSink(REPORTED_ONCE); + const before = listeners(sink); + + // Nothing is owed, because the stream holds no error. Waiting for a + // trailing event here would never return. + const result = yield* deliverWhole("catalog", sink); + + expect(result.ok ? undefined : result.error).toBe(EPIPE); + expect(listeners(sink)).toBe(before); + }); + + it("SD5: a write that refuses outright detaches too", function* () { + const sink = new RecordingSink(REFUSES); + const before = listeners(sink); + + const result = yield* deliverWhole("catalog", sink); + + expect(result.ok ? undefined : result.error).toBe(EPIPE); + expect(listeners(sink)).toBe(before); + }); + + it("SD6: a delivery that never settles detaches when its scope ends", function* () { + const sink = new RecordingSink(NEVER_ANSWERS); + const before = listeners(sink); + + yield* scoped(function* () { + yield* spawn(() => deliverWhole("catalog", sink)); + yield* sink.started.operation; + expect(listeners(sink)).toBe(before + 1); + }); + + expect(listeners(sink)).toBe(before); + }); + + it("SD7: no finished delivery absorbs a later, unrelated failure", function* () { + const sink = new RecordingSink(ACCEPTS); + const seen: Error[] = []; + sink.on("error", (error: Error) => seen.push(error)); + + yield* deliverWhole("catalog", sink); + sink.emit("error", OTHER); + + // One listener, the sentinel's, and it received the error whole. + expect(listeners(sink)).toBe(1); + expect(seen).toEqual([OTHER]); + }); + + it("SD8: the real process.stdout is left as it was found", function* () { + const before = listeners(process.stdout); + + // Empty, so this row writes nothing a reporter could mistake for output. + const result = yield* deliverWhole("", process.stdout); + + expect(result.ok).toBe(true); + expect(listeners(process.stdout)).toBe(before); + }); +}); diff --git a/specs/executable-mdx-spec.md b/specs/executable-mdx-spec.md index 0dd1a17ca..973239181 100644 --- a/specs/executable-mdx-spec.md +++ b/specs/executable-mdx-spec.md @@ -10527,6 +10527,21 @@ itself, so an execution starting anywhere fails the row. Defined in §5.3. | SX9 | Failure | An unusable include exits 1, reports on stderr and prints no catalog | | SX10/SX11 | Formats | Markdown by default, version-1 JSON with `--json`; the catalog is inspection, and `xmd plan` is the command that writes with the same structured value | | SX12 | A package tree | Bare `xmd syntax` succeeds with the default includes in a repository whose `node_modules` holds directory links | +| SX13–SX15 | Delivery | A real pipeline reading a catalog larger than one pipe buffer receives the bytes a regular-file redirect receives, in both forms; a consumer that closes early leaves the command reporting on stderr with exit 1 rather than an unhandled write failure | + +### Tier SD — Delivering a rendered result + +Delivery installs one `error` listener on a stream every command shares, so its +lifetime is the claim. Tier SX cannot show it: a real pipe reports what arrived, +not what stayed attached afterwards. + +| # | Test | Verify | +|---|------|--------| +| SD1 | Success | A delivery that succeeds leaves the stream's listeners as it found them | +| SD2/SD3 | Both arrival orders | The listener is still attached when the trailing event lands and gone once it has; either order settles on the first arrival and absorbs the duplicate | +| SD4 | One arrival | A failure the stream reports once, holding nothing after it, settles without waiting for a second | +| SD5/SD6 | Refusal and cancellation | A `write` that refuses outright and a delivery halted before it settles both detach | +| SD7/SD8 | No residue | No finished delivery observes a later, unrelated failure, and the real `process.stdout` is left as it was found | ### Tier SM — `xmd syntax` end to end From 6c014f79fc47a51009adacadcbe08a322436ec78 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Tue, 1 Sep 2026 21:44:26 -0400 Subject: [PATCH 3/6] =?UTF-8?q?=F0=9F=8F=B7=EF=B8=8F=20Name=20the=20stdout?= =?UTF-8?q?=20delivery=20tier=20SDL=20(#715)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tier SD is the CLI secret-detection opt-out's, and it holds SD1–SD13 (`packages/cli/tests/secret-detection-cli.test.ts`). The delivery evidence took the same prefix, so two unrelated suites answered to one name. The delivery tier and its eight rows become SDL. Nothing else moves: the secret-detection tier keeps SD, and the implementation and SX13–SX15 are untouched. --- packages/cli/tests/stdout-delivery.test.ts | 20 ++++++++++---------- specs/executable-mdx-spec.md | 12 ++++++------ 2 files changed, 16 insertions(+), 16 deletions(-) diff --git a/packages/cli/tests/stdout-delivery.test.ts b/packages/cli/tests/stdout-delivery.test.ts index 8226112bf..b24b63b55 100644 --- a/packages/cli/tests/stdout-delivery.test.ts +++ b/packages/cli/tests/stdout-delivery.test.ts @@ -1,5 +1,5 @@ /** - * Tier SD — delivering a rendered result to a stream. + * Tier SDL — delivering a rendered result to a stream. * * Tier SX proves what a real pipe receives. These rows prove the *lifetime* of * the one listener delivery installs, which a real pipe cannot show: a stream @@ -93,8 +93,8 @@ function listeners(sink: EventEmitter): number { return sink.listenerCount("error"); } -describe("Tier SD — the listener lives for one delivery", () => { - it("SD1: a delivery that succeeds leaves the stream as it found it", function* () { +describe("Tier SDL — the listener lives for one delivery", () => { + it("SDL1: a delivery that succeeds leaves the stream as it found it", function* () { const sink = new RecordingSink(ACCEPTS); const before = listeners(sink); @@ -105,7 +105,7 @@ describe("Tier SD — the listener lives for one delivery", () => { expect(listeners(sink)).toBe(before); }); - it("SD2: the trailing event still finds the listener, which is gone after it", function* () { + it("SDL2: the trailing event still finds the listener, which is gone after it", function* () { const sink = new RecordingSink(CALLBACK_THEN_EVENT); // A sentinel, so that detaching too early is this row's assertion rather // than an unhandled `error` event thrown from a timer. @@ -122,7 +122,7 @@ describe("Tier SD — the listener lives for one delivery", () => { expect(listeners(sink)).toBe(before); }); - it("SD3: the event may arrive first, and the first arrival is the verdict", function* () { + it("SDL3: the event may arrive first, and the first arrival is the verdict", function* () { const sink = new RecordingSink(EVENT_THEN_CALLBACK); const before = listeners(sink); @@ -132,7 +132,7 @@ describe("Tier SD — the listener lives for one delivery", () => { expect(listeners(sink)).toBe(before); }); - it("SD4: a failure the stream reports once settles without waiting for a second", function* () { + it("SDL4: a failure the stream reports once settles without waiting for a second", function* () { const sink = new RecordingSink(REPORTED_ONCE); const before = listeners(sink); @@ -144,7 +144,7 @@ describe("Tier SD — the listener lives for one delivery", () => { expect(listeners(sink)).toBe(before); }); - it("SD5: a write that refuses outright detaches too", function* () { + it("SDL5: a write that refuses outright detaches too", function* () { const sink = new RecordingSink(REFUSES); const before = listeners(sink); @@ -154,7 +154,7 @@ describe("Tier SD — the listener lives for one delivery", () => { expect(listeners(sink)).toBe(before); }); - it("SD6: a delivery that never settles detaches when its scope ends", function* () { + it("SDL6: a delivery that never settles detaches when its scope ends", function* () { const sink = new RecordingSink(NEVER_ANSWERS); const before = listeners(sink); @@ -167,7 +167,7 @@ describe("Tier SD — the listener lives for one delivery", () => { expect(listeners(sink)).toBe(before); }); - it("SD7: no finished delivery absorbs a later, unrelated failure", function* () { + it("SDL7: no finished delivery absorbs a later, unrelated failure", function* () { const sink = new RecordingSink(ACCEPTS); const seen: Error[] = []; sink.on("error", (error: Error) => seen.push(error)); @@ -180,7 +180,7 @@ describe("Tier SD — the listener lives for one delivery", () => { expect(seen).toEqual([OTHER]); }); - it("SD8: the real process.stdout is left as it was found", function* () { + it("SDL8: the real process.stdout is left as it was found", function* () { const before = listeners(process.stdout); // Empty, so this row writes nothing a reporter could mistake for output. diff --git a/specs/executable-mdx-spec.md b/specs/executable-mdx-spec.md index 973239181..a6b2435df 100644 --- a/specs/executable-mdx-spec.md +++ b/specs/executable-mdx-spec.md @@ -10529,7 +10529,7 @@ itself, so an execution starting anywhere fails the row. Defined in §5.3. | SX12 | A package tree | Bare `xmd syntax` succeeds with the default includes in a repository whose `node_modules` holds directory links | | SX13–SX15 | Delivery | A real pipeline reading a catalog larger than one pipe buffer receives the bytes a regular-file redirect receives, in both forms; a consumer that closes early leaves the command reporting on stderr with exit 1 rather than an unhandled write failure | -### Tier SD — Delivering a rendered result +### Tier SDL — Delivering a rendered result Delivery installs one `error` listener on a stream every command shares, so its lifetime is the claim. Tier SX cannot show it: a real pipe reports what arrived, @@ -10537,11 +10537,11 @@ not what stayed attached afterwards. | # | Test | Verify | |---|------|--------| -| SD1 | Success | A delivery that succeeds leaves the stream's listeners as it found them | -| SD2/SD3 | Both arrival orders | The listener is still attached when the trailing event lands and gone once it has; either order settles on the first arrival and absorbs the duplicate | -| SD4 | One arrival | A failure the stream reports once, holding nothing after it, settles without waiting for a second | -| SD5/SD6 | Refusal and cancellation | A `write` that refuses outright and a delivery halted before it settles both detach | -| SD7/SD8 | No residue | No finished delivery observes a later, unrelated failure, and the real `process.stdout` is left as it was found | +| SDL1 | Success | A delivery that succeeds leaves the stream's listeners as it found them | +| SDL2/SDL3 | Both arrival orders | The listener is still attached when the trailing event lands and gone once it has; either order settles on the first arrival and absorbs the duplicate | +| SDL4 | One arrival | A failure the stream reports once, holding nothing after it, settles without waiting for a second | +| SDL5/SDL6 | Refusal and cancellation | A `write` that refuses outright and a delivery halted before it settles both detach | +| SDL7/SDL8 | No residue | No finished delivery observes a later, unrelated failure, and the real `process.stdout` is left as it was found | ### Tier SM — `xmd syntax` end to end From 309bdad2beb29cccd06f3f7ecea467255ac825a5 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Tue, 1 Sep 2026 21:51:43 -0400 Subject: [PATCH 4/6] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Bridge=20the=20stdout?= =?UTF-8?q?=20write=20with=20withResolvers()=20instead=20of=20a=20Promise?= =?UTF-8?q?=20(#715)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `deliverWhole` settled through a `new Promise` handed to `until()`, which forced an `observe` placeholder declared outside the executor so the `finally` could reach it. `withResolvers()` is Effection's own synchronous bridge, so the listener, the write and the detach now sit in one generator. Not `action()`: a file or a terminal calls the write callback synchronously, inside `write`, and an `action()` resolved before its executor has returned never runs the cleanup that executor returns (effection 4.1.0) — five SDL rows fail that way. The detach stays in a `finally`, which is where Effection puts synchronous cleanup. --- packages/cli/src/stdout-delivery.ts | 97 +++++++++++++++-------------- 1 file changed, 49 insertions(+), 48 deletions(-) diff --git a/packages/cli/src/stdout-delivery.ts b/packages/cli/src/stdout-delivery.ts index 949d68d04..4d36f48e5 100644 --- a/packages/cli/src/stdout-delivery.ts +++ b/packages/cli/src/stdout-delivery.ts @@ -10,7 +10,7 @@ * finish before anything can exit. */ -import { Err, Ok, until } from "effection"; +import { Err, Ok, withResolvers } from "effection"; import type { Operation, Result } from "effection"; /** @@ -44,59 +44,60 @@ export interface DeliverySink { * means the event is still coming, while Bun reports the failure once and holds * nothing. Every other path — success, cancellation, a `write` that refuses * outright — detaches without waiting at all. + * + * The outcome is bridged with `withResolvers()` rather than `action()`: a file + * or a terminal calls the write callback synchronously, inside `write`, and an + * `action()` resolved before its executor has returned never runs the cleanup + * that executor returns (effection 4.1.0). */ export function* deliverWhole(text: string, sink: DeliverySink): Operation> { - let observe: (error: Error) => void = () => {}; - try { - return yield* until( - new Promise>((resolve) => { - let settled = false; - let calledBack = false; - let observed = false; - let failure: Error | undefined; + const outcome = withResolvers>("deliverWhole"); + let settled = false; + let calledBack = false; + let observed = false; + let failure: Error | undefined; - const settle = () => { - if (settled || !calledBack) { - return; - } - // Identity, not presence: an error the stream was already holding - // before this delivery has been emitted already, and waiting for it - // would wait forever. - if (!observed && failure !== undefined && sink.errored === failure) { - return; - } - settled = true; - resolve(failure === undefined ? Ok() : Err(failure)); - }; + const settle = () => { + if (settled || !calledBack) { + return; + } + // Identity, not presence: an error the stream was already holding before + // this delivery has been emitted already, and waiting for it would wait + // forever. + if (!observed && failure !== undefined && sink.errored === failure) { + return; + } + settled = true; + outcome.resolve(failure === undefined ? Ok() : Err(failure)); + }; - const record = (error?: Error | null) => { - if (error && failure === undefined) { - failure = error; - } - }; + const record = (error?: Error | null) => { + if (error && failure === undefined) { + failure = error; + } + }; - observe = (error: Error) => { - observed = true; - record(error); - settle(); - }; + const observe = (error: Error) => { + observed = true; + record(error); + settle(); + }; - sink.on("error", observe); - - try { - sink.write(text, (error) => { - calledBack = true; - record(error); - settle(); - }); - } catch (error) { - // A stream that refuses the call outright never calls back, so - // nothing else will settle this. - settled = true; - resolve(Err(error instanceof Error ? error : new Error(String(error)))); - } - }), - ); + sink.on("error", observe); + try { + try { + sink.write(text, (error) => { + calledBack = true; + record(error); + settle(); + }); + } catch (error) { + // A stream that refuses the call outright never calls back, so nothing + // else will settle this. + settled = true; + outcome.resolve(Err(error instanceof Error ? error : new Error(String(error)))); + } + return yield* outcome.operation; } finally { sink.off("error", observe); } From 2d7813139ba746f15f45f8e4314e37eead7beb79 Mon Sep 17 00:00:00 2001 From: Taras Mankovski <74687+taras@users.noreply.github.com> Date: Tue, 1 Sep 2026 22:09:43 -0400 Subject: [PATCH 5/6] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20Observe=20stdout=20err?= =?UTF-8?q?ors=20through=20a=20scope-bound=20resource=20(#715)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The error listener's lifetime is now a resource() of the delivery's own scope, the way @effectionx/node's `on()` binds a listener — attached before the write, detached in the resource's synchronous finally however the scope ends — rather than a try/finally around the wait. `scoped()` is what makes that scope the delivery's rather than the caller's: without it the listener outlives the call and seven SDL rows fail. `settled` is gone: withResolvers() ignores a second resolve, so the guard duplicated it. --- packages/cli/src/stdout-delivery.ts | 94 +++++++++++++++++------------ 1 file changed, 55 insertions(+), 39 deletions(-) diff --git a/packages/cli/src/stdout-delivery.ts b/packages/cli/src/stdout-delivery.ts index 4d36f48e5..011e827ec 100644 --- a/packages/cli/src/stdout-delivery.ts +++ b/packages/cli/src/stdout-delivery.ts @@ -10,7 +10,7 @@ * finish before anything can exit. */ -import { Err, Ok, withResolvers } from "effection"; +import { Err, Ok, resource, scoped, withResolvers } from "effection"; import type { Operation, Result } from "effection"; /** @@ -28,6 +28,27 @@ export interface DeliverySink { readonly errored?: Error | null; } +/** + * The sink's `error` events, observed for exactly as long as the enclosing + * scope lives. + * + * The listener's lifetime is the scope's and nothing else's: it is not removed + * by the event firing, and it does not survive the scope whether the event + * fired, never fired, or the scope was halted first. A stream is a + * process-global object, so a listener bound to anything looser would + * accumulate across deliveries and absorb a later, unrelated failure. + */ +function useErrorObserver(sink: DeliverySink, observe: (error: Error) => void): Operation { + return resource(function* (provide) { + sink.on("error", observe); + try { + yield* provide(); + } finally { + sink.off("error", observe); + } + }); +} + /** * Deliver `text`, and report whether the stream took all of it. * @@ -43,48 +64,45 @@ export interface DeliverySink { * error they are about to emit, so finding this delivery's own failure there * means the event is still coming, while Bun reports the failure once and holds * nothing. Every other path — success, cancellation, a `write` that refuses - * outright — detaches without waiting at all. + * outright — ends the scope, and the observer with it, without waiting at all. * * The outcome is bridged with `withResolvers()` rather than `action()`: a file * or a terminal calls the write callback synchronously, inside `write`, and an * `action()` resolved before its executor has returned never runs the cleanup * that executor returns (effection 4.1.0). */ -export function* deliverWhole(text: string, sink: DeliverySink): Operation> { - const outcome = withResolvers>("deliverWhole"); - let settled = false; - let calledBack = false; - let observed = false; - let failure: Error | undefined; +export function deliverWhole(text: string, sink: DeliverySink): Operation> { + return scoped(function* () { + const outcome = withResolvers>("deliverWhole"); + let calledBack = false; + let observed = false; + let failure: Error | undefined; - const settle = () => { - if (settled || !calledBack) { - return; - } - // Identity, not presence: an error the stream was already holding before - // this delivery has been emitted already, and waiting for it would wait - // forever. - if (!observed && failure !== undefined && sink.errored === failure) { - return; - } - settled = true; - outcome.resolve(failure === undefined ? Ok() : Err(failure)); - }; + const settle = () => { + if (!calledBack) { + return; + } + // Identity, not presence: an error the stream was already holding before + // this delivery has been emitted already, and waiting for it would wait + // forever. + if (!observed && failure !== undefined && sink.errored === failure) { + return; + } + outcome.resolve(failure === undefined ? Ok() : Err(failure)); + }; - const record = (error?: Error | null) => { - if (error && failure === undefined) { - failure = error; - } - }; + const record = (error?: Error | null) => { + if (error && failure === undefined) { + failure = error; + } + }; - const observe = (error: Error) => { - observed = true; - record(error); - settle(); - }; + yield* useErrorObserver(sink, (error) => { + observed = true; + record(error); + settle(); + }); - sink.on("error", observe); - try { try { sink.write(text, (error) => { calledBack = true; @@ -93,12 +111,10 @@ export function* deliverWhole(text: string, sink: DeliverySink): Operation Date: Tue, 1 Sep 2026 22:40:02 -0400 Subject: [PATCH 6/6] =?UTF-8?q?=E2=AC=86=EF=B8=8F=20Run=20CI's=20Bun=20job?= =?UTF-8?q?s=20on=201.4.0=20(#715)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SX15 fails under Bun on Linux and nowhere else: the command exits 0 with a truncated catalog when the consumer closes the pipe. The captured stderr is `xmd-exit=0` and nothing more — the diagnostic never printed, so delivery returned Ok. Bun 1.3.14's `process.stdout.write` callback reports success on a broken pipe there, and no `error` event follows it, so there is nothing for the command to report. macOS reports correctly on both 1.3.14 and 1.4.0, which is why only CI ever saw this. 1.4.0 is Bun rewritten in Rust, with 1,517 new Node.js test-suite passes. This defect is exactly that class. The release notes do not name it, so this commit is the experiment that answers it. The whole Bun corpus passes under 1.4.0 locally: 4713 tests across 286 files, 0 failures. `test-weights.json` still records 1.3.14 as its measurement provenance and wants a remeasure on a runner. --- .github/workflows/ci.yml | 8 ++++---- .github/workflows/measure-test-weights.yml | 2 +- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f427ddb0d..1cde9b4a2 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -99,7 +99,7 @@ jobs: # spawn a literal `bun`. The runner image ships Node but not bun. - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: - bun-version: 1.3.14 + bun-version: 1.4.0 - name: Typecheck run: deno task check @@ -445,7 +445,7 @@ jobs: - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: - bun-version: 1.3.14 + bun-version: 1.4.0 # `deno install` rather than `deno task deps`: the task also caches the # graphs a browser build and a release compile walk, and it reaches them @@ -534,7 +534,7 @@ jobs: # empty spool. - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: - bun-version: 1.3.14 + bun-version: 1.4.0 - uses: denoland/setup-deno@e95548e56dfa95d4e1a28d6f422fafe75c4c26fb # v2.0.3 with: @@ -629,7 +629,7 @@ jobs: - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: - bun-version: 1.3.14 + bun-version: 1.4.0 - run: bun install diff --git a/.github/workflows/measure-test-weights.yml b/.github/workflows/measure-test-weights.yml index 4b4fa5aeb..9f44e51cd 100644 --- a/.github/workflows/measure-test-weights.yml +++ b/.github/workflows/measure-test-weights.yml @@ -38,7 +38,7 @@ jobs: - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: - bun-version: 1.3.14 + bun-version: 1.4.0 # One preparation, in the repository's load-bearing order: `deno install`, # then `pnpm install` beside it, then the browser bundle. The measured