diff --git a/README.md b/README.md index 575cf61356..31238bae52 100644 --- a/README.md +++ b/README.md @@ -135,6 +135,11 @@ rea update `rea --update` is an alias for this verified workflow and accepts the same global output flags. Use `rea update` when combining it with other command controls. +Updater commands retain at most 4 MiB of stdout and stderr combined. Oversized +metadata replies are rejected after the command exits; verbose installs continue +with bounded diagnostics. Human progress output respects terminal backpressure +and waits for submitted writes to finish. + Bun global updates require Bun 1.2.17 or later. With an older Bun release, upgrade Bun manually before running `rea update`. diff --git a/src/application/UpdateCommandProgress.test.ts b/src/application/UpdateCommandProgress.test.ts new file mode 100644 index 0000000000..bcab02941d --- /dev/null +++ b/src/application/UpdateCommandProgress.test.ts @@ -0,0 +1,91 @@ +import { setImmediate as nextTurn } from "node:timers/promises"; +import { PassThrough, Writable } from "node:stream"; +import { expect, it } from "vitest"; +import { UpdateCommandProgress } from "./UpdateCommandProgress.js"; + +it.each(["success", "failure"] as const)( + "joins the final progress acknowledgement: %s", + async (outcome) => { + let acknowledge!: (error?: Error | null) => void; + let started!: () => void; + const writing = new Promise((resolve) => { + started = resolve; + }); + const destination = new Writable({ + write(_chunk, _encoding, callback) { + acknowledge = callback; + started(); + }, + }); + const progress = new UpdateCommandProgress(destination); + const source = new PassThrough(); + progress.pipe(source); + source.end("final progress"); + await writing; + let closed = false; + const pending = progress.close().then((result) => { + closed = true; + return result; + }); + await Promise.resolve(); + expect(closed).toBe(false); + acknowledge( + outcome === "failure" ? new Error("final progress refused") : undefined, + ); + const failure = await pending; + if (outcome === "success") expect(failure).toBeUndefined(); + else expect(failure?.message).toBe("final progress refused"); + expect(destination.writableEnded).toBe(false); + await nextTurn(); + expect(destination.listenerCount("error")).toBe(0); + expect(progress.close()).toBe(progress.close()); + }, +); + +it("applies backpressure and resumes draining after a destination failure", async () => { + let acknowledge!: (error?: Error | null) => void; + const destination = new Writable({ + highWaterMark: 1024, + write(_chunk, _encoding, callback) { + acknowledge = callback; + }, + }); + const progress = new UpdateCommandProgress(destination); + const source = new PassThrough({ highWaterMark: 1024 }); + const ended = new Promise((resolve) => { + source.once("end", resolve); + }); + progress.pipe(source); + const chunk = Buffer.alloc(64 * 1024); + source.write(chunk); + expect(source.isPaused()).toBe(true); + source.end(chunk); + expect(destination.writableLength).toBe(chunk.length); + acknowledge(new Error("progress destination refused")); + await ended; + expect((await progress.close())?.message).toBe( + "progress destination refused", + ); + expect(destination.writableEnded).toBe(false); +}); + +it("observes late write rejection after an independent destination failure", async () => { + let acknowledge!: (error?: Error | null) => void; + const destination = new Writable({ + write(_chunk, _encoding, callback) { + acknowledge = callback; + }, + }); + const progress = new UpdateCommandProgress(destination); + const source = new PassThrough(); + progress.pipe(source); + source.end("pending progress"); + destination.emit("error", new Error("independent destination failure")); + expect((await progress.close())?.message).toBe( + "independent destination failure", + ); + expect(destination.listenerCount("error")).toBe(1); + acknowledge(new Error("late write rejection")); + await nextTurn(); + expect(destination.listenerCount("error")).toBe(0); +}); diff --git a/src/application/UpdateCommandProgress.ts b/src/application/UpdateCommandProgress.ts new file mode 100644 index 0000000000..6e5203d4db --- /dev/null +++ b/src/application/UpdateCommandProgress.ts @@ -0,0 +1,68 @@ +import { Writable, type Readable } from "node:stream"; +import { finished } from "node:stream/promises"; + +/** Borrow a progress destination while owning backpressure and write acknowledgements. */ +export class UpdateCommandProgress { + readonly #streams: Readable[] = []; + readonly #sink: Writable; + readonly #finished: Promise; + #closing: Promise | undefined; + #failed = false; + #pendingWrites = 0; + #closed = false; + readonly #onDestinationError = (cause: Error): void => { + this.#sink.destroy(cause); + }; + + constructor(private readonly destination: Writable) { + this.#sink = new Writable({ + write: (chunk: Buffer, encoding, callback) => { + this.#pendingWrites++; + destination.write(chunk, encoding, (cause) => { + this.#pendingWrites--; + callback(cause); + // A destination can emit its write error after invoking the callback. + setImmediate(() => this.#releaseDestination()); + }); + }, + }); + this.#finished = finished(this.#sink, { cleanup: true }).then( + () => undefined, + (cause: unknown) => { + this.#failed = true; + for (const stream of this.#streams) { + stream.unpipe(this.#sink); + stream.resume(); + } + return cause instanceof Error ? cause : new Error(String(cause)); + }, + ); + destination.on("error", this.#onDestinationError); + } + + pipe(stream: Readable): void { + this.#streams.push(stream); + if (this.#failed) stream.resume(); + else stream.pipe(this.#sink, { end: false }); + } + + close(): Promise { + this.#closing ??= this.#close(); + return this.#closing; + } + + async #close(): Promise { + for (const stream of this.#streams) stream.unpipe(this.#sink); + this.#sink.end(); + try { + return await this.#finished; + } finally { + this.#closed = true; + setImmediate(() => this.#releaseDestination()); + } + } + #releaseDestination(): void { + if (this.#closed && this.#pendingWrites === 0) + this.destination.off("error", this.#onDestinationError); + } +} diff --git a/src/application/UpdateRuntime.ts b/src/application/UpdateRuntime.ts index b4986cbcf5..9b54597ad7 100644 --- a/src/application/UpdateRuntime.ts +++ b/src/application/UpdateRuntime.ts @@ -3,6 +3,8 @@ import { homeDirectoryFromEnvironment } from "../config/homeDirectory.js"; import { basename, dirname, join, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import spawn from "cross-spawn"; +import { ProviderProcessSupervisor } from "../process/ProviderProcess.js"; +import { UpdateCommandProgress } from "./UpdateCommandProgress.js"; import { lt, valid } from "semver"; import { PRODUCT_IDENTITY } from "../identity.js"; @@ -34,54 +36,100 @@ export interface BunInstallationHost { isFile(path: string): Promise; } -/** Run an updater subprocess with complete diagnostics and isolated machine output. */ +/** Aggregate raw-byte retention policy for package-manager replies and diagnostics. */ +export const UPDATE_COMMAND_OUTPUT_BYTES = 4 * 1024 * 1024; + +/** Complete metadata queries share launch and collection with diagnostic-only installs. */ export const runUpdateCommand = ( command: readonly string[], selectedEnvironment: Readonly, output: UpdateOutput = "structured", workingDirectory: string | undefined = undefined, ): Promise> => - new Promise((resolveResult) => { - const [executable, ...args] = command; - if (executable === undefined) { - resolveResult(err("No executable supplied.")); - return; - } - // A fresh installed entry point must not inherit a package runner's invocation identity. - const environment = withoutPackageRunnerIdentity(selectedEnvironment); - const child = spawn(executable, args, { - stdio: ["ignore", "pipe", "pipe"], - env: environment, - windowsHide: true, - ...(workingDirectory === undefined ? {} : { cwd: workingDirectory }), - }); - let stdout = ""; - let stderr = ""; - child.stdout?.setEncoding("utf8"); - child.stderr?.setEncoding("utf8"); - child.stdout?.on("data", (chunk: string) => { - stdout += chunk; - if (output === "human") process.stderr.write(chunk); - }); - child.stderr?.on("data", (chunk: string) => { - stderr += chunk; - if (output === "human") process.stderr.write(chunk); - }); - child.once("error", (cause) => resolveResult(err(cause.message))); - child.once("close", (code, signal) => - resolveResult( - code === 0 - ? ok(stdout.trim()) - : err( - stderr.trim() || - stdout.trim() || - (signal === null - ? `Process exited with code ${String(code)}.` - : `Process terminated by ${signal}.`), - ), - ), - ); + runUpdateProcess( + command, + selectedEnvironment, + output, + workingDirectory, + "complete", + ); + +const runUpdateProcess = async ( + command: readonly string[], + selectedEnvironment: Readonly, + output: UpdateOutput, + workingDirectory: string | undefined, + retention: "complete" | "diagnostics", +): Promise> => { + const [executable, ...args] = command; + if (executable === undefined) return err("No executable supplied."); + const environment = withoutPackageRunnerIdentity(selectedEnvironment); + const child = spawn(executable, args, { + stdio: ["ignore", "pipe", "pipe"], + env: environment, + windowsHide: true, + ...(workingDirectory === undefined ? {} : { cwd: workingDirectory }), }); + let processFailure: string | undefined; + const supervisor = new ProviderProcessSupervisor( + { process: child, ownsProcessLifetime: true }, + { + maxDiagnosticBytes: UPDATE_COMMAND_OUTPUT_BYTES, + onDiagnostic: (event) => { + if (event.type === "error") processFailure ??= event.message; + }, + }, + ); + const progress = + output === "human" ? new UpdateCommandProgress(process.stderr) : undefined; + if (progress !== undefined) { + for (const stream of [child.stdout, child.stderr]) + if (stream !== null) progress.pipe(stream); + } + try { + // Drain even an oversized reply. Verbosity must not interrupt a package install. + // Keep the existing launch/signal policy; this collector owns no process-tree authority. + await supervisor.waitForOutputClose(); + const progressFailure = await progress?.close(); + const snapshot = supervisor.snapshot(); + if (processFailure !== undefined) return err(processFailure); + if (snapshot.outputIncomplete === true) + return err("Update command output did not close completely."); + if (retention === "complete" && snapshot.diagnosticTruncated === true) + return err( + `Update command output exceeded the ${UPDATE_COMMAND_OUTPUT_BYTES}-byte complete-output retention budget; reply was not accepted.`, + ); + if (snapshot.exitCode !== 0 || snapshot.signal !== null) { + const detail = + snapshot.stderr.text.trim() || + snapshot.stdout.text.trim() || + (typeof snapshot.signal === "string" + ? `Process terminated by ${snapshot.signal}.` + : typeof snapshot.exitCode === "number" + ? `Process exited with code ${snapshot.exitCode}.` + : "Update command failed without a recorded exit status."); + return err( + snapshot.diagnosticTruncated + ? `${detail}\n[Update diagnostics truncated at ${UPDATE_COMMAND_OUTPUT_BYTES} bytes; later output omitted.]` + : detail, + ); + } + if ( + retention === "complete" && + (snapshot.stdout.invalidUtf8 || snapshot.stderr.invalidUtf8) + ) + return err( + "Update command returned invalid UTF-8; complete output is unavailable.", + ); + if (progressFailure !== undefined) + return err(`Update progress output failed: ${progressFailure.message}`); + return ok(retention === "diagnostics" ? "" : snapshot.stdout.text.trim()); + } finally { + // Reached only after physical close; no child or inherited pipe remains pending. + supervisor.dispose(); + await progress?.close(); + } +}; const requireCommandOutput = async ( command: readonly string[], @@ -341,11 +389,12 @@ export const systemUpdateHost = ( }, installVersion: async (installation, version, output) => { const context = packageManagerContext(installation, environment); - const result = await runUpdateCommand( + const result = await runUpdateProcess( updateInstallCommand(installation, version), context.environment, output, context.workingDirectory, + "diagnostics", ); return result.ok ? ok(undefined) : result; }, diff --git a/tests/acceptance/setup/updateAliasCli.test.ts b/tests/acceptance/setup/updateAliasCli.test.ts index 3dceafc903..9fd0536d38 100644 --- a/tests/acceptance/setup/updateAliasCli.test.ts +++ b/tests/acceptance/setup/updateAliasCli.test.ts @@ -26,6 +26,7 @@ cliTest( else if(args[0]==='prefix') console.log(join(process.env.USERPROFILE,'unowned')); else if(args[0]==='view') { if(process.env.REA_TEST_NPM_FAILURE==='1') {process.stderr.write('owned fake registry denied');process.exitCode=2;} + else if(process.env.REA_TEST_NPM_OVERSIZED==='1') process.stdout.write(JSON.stringify(${JSON.stringify(PRODUCT_IDENTITY.packageVersion)})+' '.repeat(4 * 1024 * 1024)); else console.log(JSON.stringify(${JSON.stringify(PRODUCT_IDENTITY.packageVersion)})); } else {process.stderr.write('No installation or unexpected package-manager commands are allowed');process.exitCode=97;} `, @@ -37,18 +38,31 @@ cliTest( : '#!/bin/sh\nexec "$REA_TEST_NODE" "$REA_TEST_NPM_SCRIPT" "$@"\n', { mode: 0o700 }, ); - const run = (arguments_: readonly string[], failure = false) => + const run = ( + arguments_: readonly string[], + failure = false, + oversized = false, + ) => cli.run({ arguments: arguments_, cwd: home, timeoutMs: 20_000, environment: { USERPROFILE: home, - PATH: bin + delimiter + (process.env.PATH ?? ""), + PATH: [ + bin, + ...(process.platform === "win32" + ? [join(process.env.SystemRoot ?? "C:\\Windows", "System32")] + : ["/usr/bin", "/bin"]), + ].join(delimiter), + npm_config_prefix: join(home, "prefix"), + npm_config_cache: join(home, "cache"), + npm_config_user_agent: "npm/test", REA_TEST_NODE: process.execPath, REA_TEST_NPM_SCRIPT: script, REA_TEST_NPM_LOG: log, REA_TEST_NPM_FAILURE: failure ? "1" : "0", + REA_TEST_NPM_OVERSIZED: oversized ? "1" : "0", CI: "1", NO_UPDATE_NOTIFIER: "1", }, @@ -82,6 +96,13 @@ cliTest( reason: "version-check", remediation: expect.stringContaining("owned fake registry denied"), }); + const oversized = await run(["update", "--json"], false, true); + expect(oversized.exitCode).toBe(1); + expect(oversized.json).toMatchObject({ + status: "failed", + reason: "version-check", + remediation: expect.stringContaining("complete-output retention budget"), + }); const before = await readFile(log, "utf8"); const ambiguous = await run(["providers", "--update", "--json"]); expect(ambiguous.exitCode).toBe(1); diff --git a/tests/process-global/updateMaintenance.test.ts b/tests/process-global/updateMaintenance.test.ts index a57864da14..11babb5273 100644 --- a/tests/process-global/updateMaintenance.test.ts +++ b/tests/process-global/updateMaintenance.test.ts @@ -12,6 +12,7 @@ import { } from "../../src/application/UpdateMaintenance.js"; import { runUpdateCommand, + UPDATE_COMMAND_OUTPUT_BYTES, systemUpdateHost, } from "../../src/application/UpdateRuntime.js"; import { createTestTempDirectory } from "../fixtures/temporaryDirectory.js"; @@ -444,65 +445,93 @@ describe("update subprocess boundaries", () => { }); }); +const createBunUpdateFixture = async ( + root: string, + bunVersion: string, + installOutputBytes: number, + installExitCode: number, +) => { + const globalDirectory = join(root, "bun", "install", "global"); + const packageRoot = join( + globalDirectory, + "node_modules", + PRODUCT_IDENTITY.packageName, + ); + const versionPath = join(root, "installed-version"); + const logPath = join(root, "bun-invocations.jsonl"); + await mkdir(join(packageRoot, "scripts"), { recursive: true }); + await writeFile( + join(globalDirectory, "package.json"), + '{"dependencies":{"rea-agents":"3.2.1"}}\n', + ); + await writeFile(versionPath, "3.2.1"); + await writeFile( + join(packageRoot, "scripts", "rea.mjs"), + `import { readFileSync } from "node:fs";\nprocess.stdout.write(readFileSync(${JSON.stringify(versionPath)}, "utf8") + "\\n");\n`, + ); + const binDirectory = join(root, "bin"); + const recorder = join(binDirectory, "record-bun.mjs"); + await mkdir(binDirectory); + await writeFile( + recorder, + [ + 'import { appendFileSync, writeFileSync } from "node:fs";', + "const args = process.argv.slice(2);", + `appendFileSync(${JSON.stringify(logPath)}, JSON.stringify({ args, cwd: process.cwd(), globalDirectory: process.env.BUN_INSTALL_GLOBAL_DIR }) + "\\n");`, + `if (JSON.stringify(args) === '["--version"]') console.log(${JSON.stringify(bunVersion)});`, + `else if (JSON.stringify(args) === ${JSON.stringify(JSON.stringify(["info", PRODUCT_IDENTITY.packageName, "dist-tags.latest", "--json"]))}) process.stdout.write('"3.3.0"\\n');`, + `else if (JSON.stringify(args) === ${JSON.stringify(JSON.stringify(["add", "--global", "--ignore-scripts", `${PRODUCT_IDENTITY.packageName}@3.3.0`]))}) process.stderr.write('x'.repeat(${installOutputBytes}), () => { ${installExitCode === 0 ? `writeFileSync(${JSON.stringify(versionPath)}, "3.3.0");` : "process.exitCode = 2;"} });`, + "else process.exitCode = 2;", + "", + ].join("\n"), + ); + if (process.platform === "win32") { + await writeFile( + join(binDirectory, "bun.cmd"), + `@"${process.execPath}" "${recorder}" %*\r\n`, + ); + } else { + await writeFile( + join(binDirectory, "bun"), + `#!/bin/sh\nexec ${JSON.stringify(process.execPath)} ${JSON.stringify(recorder)} "$@"\n`, + { mode: 0o755 }, + ); + } + return { globalDirectory, packageRoot, versionPath, logPath, binDirectory }; +}; + describe("Bun global installation update", () => { it.each([ - ["1.4.2", "updated", "absolute"], - ["1.2.17", "updated", "absolute"], - ["1.2.17", "updated", "relative"], - ["1.2.14", "failed", "absolute"], - ["1.2.17-canary.1", "failed", "absolute"], - ["unknown", "failed", "absolute"], - ])( + ["1.4.2", "updated", "absolute", UPDATE_COMMAND_OUTPUT_BYTES + 1, 0], + ["1.4.2", "install-failed", "absolute", UPDATE_COMMAND_OUTPUT_BYTES + 1, 2], + ["1.2.17", "updated", "absolute", 0, 0], + ["1.2.17", "updated", "relative", 0, 0], + ["1.2.14", "failed", "absolute", 0, 0], + ["1.2.17-canary.1", "failed", "absolute", 0, 0], + ["unknown", "failed", "absolute", 0, 0], + ] as const)( "handles Bun %s with outcome %s using %s global-directory configuration", - async (bunVersion, status, directoryMode) => { + async ( + bunVersion, + status, + directoryMode, + installOutputBytes, + installExitCode, + ) => { // A recording `bun` on PATH crosses the real subprocess boundary without registry access. const root = await createTestTempDirectory("rea-update-bun-"); - const globalDirectory = join(root, "bun", "install", "global"); - const packageRoot = join( + const { globalDirectory, - "node_modules", - PRODUCT_IDENTITY.packageName, + packageRoot, + versionPath, + logPath, + binDirectory, + } = await createBunUpdateFixture( + root, + bunVersion, + installOutputBytes, + installExitCode, ); - const versionPath = join(root, "installed-version"); - const logPath = join(root, "bun-invocations.jsonl"); - await mkdir(join(packageRoot, "scripts"), { recursive: true }); - await writeFile( - join(globalDirectory, "package.json"), - '{"dependencies":{"rea-agents":"3.2.1"}}\n', - ); - await writeFile(versionPath, "3.2.1"); - await writeFile( - join(packageRoot, "scripts", "rea.mjs"), - `import { readFileSync } from "node:fs";\nprocess.stdout.write(readFileSync(${JSON.stringify(versionPath)}, "utf8") + "\\n");\n`, - ); - const binDirectory = join(root, "bin"); - const recorder = join(binDirectory, "record-bun.mjs"); - await mkdir(binDirectory); - await writeFile( - recorder, - [ - 'import { appendFileSync, writeFileSync } from "node:fs";', - "const args = process.argv.slice(2);", - `appendFileSync(${JSON.stringify(logPath)}, JSON.stringify({ args, cwd: process.cwd(), globalDirectory: process.env.BUN_INSTALL_GLOBAL_DIR }) + "\\n");`, - `if (JSON.stringify(args) === '["--version"]') console.log(${JSON.stringify(bunVersion)});`, - `else if (JSON.stringify(args) === ${JSON.stringify(JSON.stringify(["info", PRODUCT_IDENTITY.packageName, "dist-tags.latest", "--json"]))}) process.stdout.write('"3.3.0"\\n');`, - `else if (JSON.stringify(args) === ${JSON.stringify(JSON.stringify(["add", "--global", "--ignore-scripts", `${PRODUCT_IDENTITY.packageName}@3.3.0`]))}) writeFileSync(${JSON.stringify(versionPath)}, "3.3.0");`, - "else process.exitCode = 2;", - "", - ].join("\n"), - ); - if (process.platform === "win32") { - await writeFile( - join(binDirectory, "bun.cmd"), - `@"${process.execPath}" "${recorder}" %*\r\n`, - ); - } else { - await writeFile( - join(binDirectory, "bun"), - `#!/bin/sh\nexec ${JSON.stringify(process.execPath)} ${JSON.stringify(recorder)} "$@"\n`, - { mode: 0o755 }, - ); - } const pathKey = Object.keys(process.env).find((key) => key.toUpperCase() === "PATH") ?? "PATH"; @@ -550,18 +579,26 @@ describe("Bun global installation update", () => { return; } - expect(result).toMatchObject({ - status: "updated", - installMethod: "bun", - installedVersion: "3.3.0", - command: [ - "bun", - "add", - "--global", - "--ignore-scripts", - `${PRODUCT_IDENTITY.packageName}@3.3.0`, - ], - }); + if (status === "install-failed") { + expect(result).toMatchObject({ + status: "failed", + reason: "install", + remediation: expect.stringContaining("diagnostics truncated"), + }); + expect(await readFile(versionPath, "utf8")).toBe("3.2.1"); + } else + expect(result).toMatchObject({ + status: "updated", + installMethod: "bun", + installedVersion: "3.3.0", + command: [ + "bun", + "add", + "--global", + "--ignore-scripts", + `${PRODUCT_IDENTITY.packageName}@3.3.0`, + ], + }); // `bun info` needs a project in its working directory, and the caller's // directory must never be selected, so all commands run in the owner. expect(invocations).toEqual([ diff --git a/tests/process-global/updateOutput.test.ts b/tests/process-global/updateOutput.test.ts new file mode 100644 index 0000000000..9b004168a0 --- /dev/null +++ b/tests/process-global/updateOutput.test.ts @@ -0,0 +1,145 @@ +import { spawn, type ChildProcess } from "node:child_process"; +import { setImmediate as nextTurn } from "node:timers/promises"; +import { Writable } from "node:stream"; +import { afterEach, beforeEach, expect, it, vi } from "vitest"; +import { + runUpdateCommand, + UPDATE_COMMAND_OUTPUT_BYTES, +} from "../../src/application/UpdateRuntime.js"; + +const execution = vi.hoisted(() => ({ spawn: vi.fn() })); +vi.mock("cross-spawn", () => ({ default: execution.spawn })); +let child: ChildProcess | undefined; +beforeEach(() => { + execution.spawn.mockReset(); + execution.spawn.mockImplementation( + ( + executable: string, + args: readonly string[], + options: Parameters[2], + ) => { + if (executable !== process.execPath) + throw new Error("Only owned Node fixtures may execute"); + child = spawn(executable, [...args], options); + return child; + }, + ); +}); +afterEach(async () => { + vi.restoreAllMocks(); + if ( + child !== undefined && + child.exitCode === null && + child.signalCode === null + ) { + const closed = new Promise((resolve) => { + child!.once("close", resolve); + }); + child.kill("SIGKILL"); + await closed; + } + child = undefined; +}); +const command = (script: string) => [process.execPath, "-e", script]; + +it.each(["stdout", "stderr"])( + "rejects incomplete %s metadata after draining the command", + async (stream) => { + const result = await runUpdateCommand( + command( + `process.stdout.write('"6.5.0"');process.${stream}.write(' '.repeat(${UPDATE_COMMAND_OUTPUT_BYTES}))`, + ), + process.env, + ); + expect(result).toMatchObject({ + ok: false, + error: expect.stringContaining("complete-output retention budget"), + }); + expect(child?.exitCode).toBe(0); + }, +); + +it("accepts exactly the combined byte budget and preserves split UTF-8", async () => { + const result = await runUpdateCommand( + command( + `process.stdout.write('x'.repeat(${UPDATE_COMMAND_OUTPUT_BYTES - 3}));const b=Buffer.from('界');process.stdout.write(b.subarray(0,1));setTimeout(()=>process.stdout.write(b.subarray(1)),10)`, + ), + process.env, + ); + expect(result.ok).toBe(true); + if (result.ok) { + expect(Buffer.byteLength(result.value)).toBe(UPDATE_COMMAND_OUTPUT_BYTES); + expect(result.value.endsWith("界")).toBe(true); + } +}); + +it("refuses malformed UTF-8 instead of interpreting replacement text", async () => { + expect( + await runUpdateCommand( + command("process.stdout.write(Buffer.from([255]))"), + process.env, + ), + ).toMatchObject({ + ok: false, + error: expect.stringContaining("invalid UTF-8"), + }); +}); + +it("waits for physical close after an early process error", async () => { + let settled = false; + const pending = runUpdateCommand( + command("setTimeout(()=>process.stdout.write('late output'),50)"), + process.env, + ).then((result) => { + settled = true; + return result; + }); + child!.emit("error", new Error("controlled process error")); + await Promise.resolve(); + expect(settled).toBe(false); + expect(await pending).toEqual({ + ok: false, + error: "controlled process error", + }); + expect(child?.exitCode).toBe(0); +}); + +it("joins the final human progress write before reporting success", async () => { + let acknowledge!: (cause?: Error | null) => void; + let started!: () => void; + const writing = new Promise((resolve) => { + started = resolve; + }); + const destination = new Writable({ + write(_chunk, _encoding, callback) { + acknowledge = callback; + started(); + }, + }); + vi.spyOn(process, "stderr", "get").mockReturnValue( + destination as typeof process.stderr, + ); + let settled = false; + const pending = runUpdateCommand( + command("process.stdout.write('final progress')"), + process.env, + "human", + ).then((result) => { + settled = true; + return result; + }); + const closed = new Promise((resolve) => { + child!.once("close", resolve); + }); + await writing; + await closed; + expect(settled).toBe(false); + acknowledge(new Error("final progress refused")); + expect(await pending).toMatchObject({ + ok: false, + error: expect.stringContaining("final progress refused"), + }); + expect(destination.writableEnded).toBe(false); + await nextTurn(); + expect(destination.listenerCount("error")).toBe(0); +});