From 2bd981b0179cc1e4cfdb6ab52ab83c0d4c6c07ed Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Sat, 24 Oct 2026 10:50:03 +0800 Subject: [PATCH] refactor(tests): consolidate compiled CLI session ownership --- docs/testing.md | 17 +- .../exportedFunctionBindingsCli.test.ts | 31 +- tests/boundary/cli/compiledCliSession.test.ts | 266 ---------- .../cli/sharedJavaScriptCliSession.test.ts | 457 +++++++++++++++++- tests/support/cli/cliCommandSession.ts | 151 ++++-- tests/support/cli/compiledCliSession.ts | 444 ----------------- tests/support/cli/compiledCliWorker.mjs | 151 ------ 7 files changed, 598 insertions(+), 919 deletions(-) delete mode 100644 tests/boundary/cli/compiledCliSession.test.ts delete mode 100644 tests/support/cli/compiledCliSession.ts delete mode 100644 tests/support/cli/compiledCliWorker.mjs diff --git a/docs/testing.md b/docs/testing.md index 4ec866a6d8..53c62fb3aa 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -233,13 +233,16 @@ the next command starts. IL2CPP metadata, and dylib-resolution commands use that session. File teardown stops the shared CLI process, including startup still in flight, and does not open another after close. -`exportedFunctionBindingsCli.test.ts` keeps its own compiled CLI parity -process for the file and still compares that process's JSON command results -with the MCP session. A command that does not respond is rejected. That -process stays owned until it exits, and another process does not start while -it remains. A process that stays running after SIGKILL remains owned so a -later close can retry. Closing the session reports a failure when the process -does not exit after stdin closes, once the child has been stopped. +`exportedFunctionBindingsCli.test.ts` uses the shared compiled CLI session for +its file and compares its JSON command results with the MCP session. It selects +20-second startup and command deadlines; other callers retain their existing +deadline behavior. Retired processes stay owned until exit, block replacement, +and remain available for a later close to retry after failed termination. +Shared session cleanup closes stdin so the worker can release JavaScript +resources, waits the bounded grace, then escalates to SIGTERM and SIGKILL only +if needed. Confirmed exit succeeds; an unconfirmed exit reports cleanup failure. +When startup or a command fails and cleanup also fails, the reported aggregate +retains both failures and the original failure as its cause. `applicationWorkflowCli.test.ts` keeps one compiled CLI process for its JSON commands, including a non-zero status whose stdout is still a JSON document, and closes that process after the file. diff --git a/tests/acceptance/applications/exportedFunctionBindingsCli.test.ts b/tests/acceptance/applications/exportedFunctionBindingsCli.test.ts index 9a6d0cd66c..bc9a6b3ca5 100644 --- a/tests/acceptance/applications/exportedFunctionBindingsCli.test.ts +++ b/tests/acceptance/applications/exportedFunctionBindingsCli.test.ts @@ -15,9 +15,9 @@ import { javascriptApplicationAnalysisResultSchema } from "../../../src/domain/j import { javaScriptExportShapeComparisonResultSchema } from "../../../src/domain/javascript/javascriptExportShapeComparisonSchemas.js"; import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js"; import { - startCompiledCliSession, - type CompiledCliSession, -} from "../../support/cli/compiledCliSession.js"; + createCliCommandSession, + type CliCommandSession, +} from "../../support/cli/cliCommandSession.js"; const execute = promisify(execFile); const requireFixture = createRequire(import.meta.url); @@ -27,11 +27,10 @@ const comparisonEnvelope = z.object({ const applicationEnvelope = z.object({ normalized_result: javascriptApplicationAnalysisResultSchema, }); -let cliSession: CompiledCliSession | undefined; +let cliSession: CliCommandSession | undefined; -// Source semantics use one real stdio MCP session. Cases that also check the -// compiled CLI adapter share one warm process: it loads the production command -// runtime once and serves the same JSON commands as scripts/rea.mjs. +// Source semantics use one real stdio MCP session. CLI parity uses the shared +// file-scoped session, including its worker rotation and cleanup policy. let mcpConnection: | { readonly client: Client; @@ -1207,8 +1206,22 @@ const expectCountChange = ( }; const runCli = async (args: readonly string[]): Promise => { - cliSession ??= startCompiledCliSession(); - return cliSession.run(args); + cliSession ??= createCliCommandSession({ + startTimeoutMs: 20_000, + requestTimeoutMs: 20_000, + }); + const environment = Object.fromEntries( + Object.entries({ ...process.env, REA_LOG_LEVEL: "silent" }).filter( + (entry): entry is [string, string] => entry[1] !== undefined, + ), + ); + const result = await cliSession.run({ + arguments: args, + cwd: process.cwd(), + environment, + }); + expect(result.exitCode, result.stderr || result.stdout).toBe(0); + return result.json; }; const analyzeVersions = async ( diff --git a/tests/boundary/cli/compiledCliSession.test.ts b/tests/boundary/cli/compiledCliSession.test.ts deleted file mode 100644 index 0499a5c44c..0000000000 --- a/tests/boundary/cli/compiledCliSession.test.ts +++ /dev/null @@ -1,266 +0,0 @@ -import { EventEmitter, PassThrough } from "node:stream"; -import type { ChildProcessWithoutNullStreams } from "node:child_process"; -import { writeFile } from "node:fs/promises"; -import { join } from "node:path"; - -import { expect, onTestFinished, test } from "vitest"; - -import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js"; -import { startCompiledCliSession } from "../../support/cli/compiledCliSession.js"; - -interface FakeCliProcess extends EventEmitter { - stdin: PassThrough; - stdout: PassThrough; - stderr: PassThrough; - pid: number; - exitCode: number | null; - signalCode: NodeJS.Signals | null; - kill(signal?: NodeJS.Signals): boolean; -} - -const fakeProcess = (pid: number): FakeCliProcess => { - const child = new EventEmitter() as FakeCliProcess; - child.stdin = new PassThrough(); - child.stdout = new PassThrough(); - child.stderr = new PassThrough(); - child.pid = pid; - child.exitCode = null; - child.signalCode = null; - child.kill = (signal: NodeJS.Signals = "SIGTERM") => { - if (child.exitCode !== null || child.signalCode !== null) return true; - child.signalCode = signal; - child.emit("exit", null, signal); - child.emit("close", null, signal); - return true; - }; - return child; -}; - -const asCliProcess = (child: FakeCliProcess): ChildProcessWithoutNullStreams => - child as unknown as ChildProcessWithoutNullStreams; - -const ready = (child: FakeCliProcess): void => { - child.stdout.write(`${JSON.stringify({ ready: true })}\n`); -}; - -const exitWhenStdinCloses = (child: FakeCliProcess): void => { - child.stdin.on("finish", () => { - if (child.exitCode !== null || child.signalCode !== null) return; - child.exitCode = 0; - child.emit("exit", 0, null); - child.emit("close", 0, null); - }); -}; - -const replyWhenAsked = ( - child: FakeCliProcess, - stdout: string, - exitCode = 0, -): void => { - let buffer = ""; - child.stdin.setEncoding("utf8"); - child.stdin.on("data", (chunk: string) => { - buffer += chunk; - let newline = buffer.indexOf("\n"); - while (newline !== -1) { - const line = buffer.slice(0, newline); - buffer = buffer.slice(newline + 1); - if (line.length > 0) { - const request = JSON.parse(line) as { id: number }; - child.stdout.write( - `${JSON.stringify({ id: request.id, exitCode, stdout })}\n`, - ); - } - newline = buffer.indexOf("\n"); - } - }); -}; - -test("rejects a compiled CLI command when close wins the ready handshake", async () => { - let child: FakeCliProcess | undefined; - let releaseReady: (() => void) | undefined; - const session = startCompiledCliSession({ - stopTimeoutMs: 50, - spawnWorker: () => { - const started = fakeProcess(60); - child = started; - exitWhenStdinCloses(started); - releaseReady = () => { - ready(started); - }; - return asCliProcess(started); - }, - }); - - const pending = session.run(["early"]); - for (let attempt = 0; child === undefined && attempt < 30; attempt += 1) - await new Promise((resolve) => { - setImmediate(resolve); - }); - if (child === undefined) throw new Error("compiled CLI worker did not start"); - const closing = session.close(); - releaseReady?.(); - await expect(pending).rejects.toThrow("Compiled CLI session is closed"); - await closing; -}); - -test("rejects a compiled CLI command that never replies and serves the next one", async () => { - const children: FakeCliProcess[] = []; - const session = startCompiledCliSession({ - requestTimeoutMs: 30, - spawnWorker: () => { - const child = fakeProcess(70 + children.length); - children.push(child); - exitWhenStdinCloses(child); - ready(child); - if (children.length > 1) - replyWhenAsked(child, JSON.stringify({ ok: true })); - return asCliProcess(child); - }, - }); - - await expect(session.run(["stuck"])).rejects.toThrow(/did not respond/); - expect(children[0]?.signalCode).toBe("SIGKILL"); - await expect(session.run(["next"])).resolves.toEqual({ ok: true }); - expect(children).toHaveLength(2); - await session.close(); -}); - -test("keeps a timed-out compiled CLI worker until it exits", async () => { - const children: FakeCliProcess[] = []; - let releaseExit: (() => void) | undefined; - const session = startCompiledCliSession({ - requestTimeoutMs: 20, - stopTimeoutMs: 40, - spawnWorker: () => { - const child = fakeProcess(90 + children.length); - children.push(child); - if (children.length === 1) { - child.kill = () => { - releaseExit = () => { - if (child.exitCode !== null || child.signalCode !== null) return; - child.signalCode = "SIGKILL"; - child.emit("exit", null, "SIGKILL"); - child.emit("close", null, "SIGKILL"); - }; - return true; - }; - } else { - exitWhenStdinCloses(child); - replyWhenAsked(child, JSON.stringify({ ok: true })); - } - ready(child); - return asCliProcess(child); - }, - }); - - const stalled = session.run(["stuck"]); - const next = session.run(["next"]); - await expect(stalled).rejects.toThrow(/did not respond/); - expect(children).toHaveLength(1); - expect(children[0]?.signalCode).toBeNull(); - releaseExit?.(); - await expect(next).resolves.toEqual({ ok: true }); - expect(children).toHaveLength(2); - await session.close(); -}); - -test("retries a timed-out compiled CLI worker that survives SIGKILL", async () => { - let child: FakeCliProcess | undefined; - const signals: NodeJS.Signals[] = []; - const session = startCompiledCliSession({ - requestTimeoutMs: 20, - stopTimeoutMs: 20, - spawnWorker: () => { - child = fakeProcess(91); - child.kill = (signal: NodeJS.Signals = "SIGTERM") => { - signals.push(signal); - return true; - }; - ready(child); - return asCliProcess(child); - }, - }); - - await expect(session.run(["stuck"])).rejects.toThrow(/did not respond/); - const replacement = session.run(["next"]); - await expect(replacement).rejects.toThrow(/stayed running after SIGKILL/); - await expect(session.close()).rejects.toThrow(/stayed running after SIGKILL/); - expect(signals).toEqual(["SIGKILL", "SIGKILL"]); - const stalledChild = child; - if (stalledChild === undefined) throw new Error("missing stalled worker"); - stalledChild.kill = (signal: NodeJS.Signals = "SIGKILL") => { - signals.push(signal); - stalledChild.signalCode = signal; - stalledChild.emit("exit", null, signal); - stalledChild.emit("close", null, signal); - return true; - }; - await session.close(); - expect(stalledChild.signalCode).toBe("SIGKILL"); -}); - -test("reports a compiled CLI session that needed a signal after stdin closed", async () => { - let child: FakeCliProcess | undefined; - const signals: NodeJS.Signals[] = []; - const session = startCompiledCliSession({ - stopTimeoutMs: 20, - spawnWorker: () => { - child = fakeProcess(81); - child.kill = (signal: NodeJS.Signals = "SIGTERM") => { - signals.push(signal); - if (signal !== "SIGKILL") return true; - if (child?.exitCode !== null || child?.signalCode !== null) return true; - child.signalCode = signal; - child.emit("exit", null, signal); - child.emit("close", null, signal); - return true; - }; - ready(child); - replyWhenAsked(child, JSON.stringify({ ok: true })); - return asCliProcess(child); - }, - }); - - await expect(session.run(["once"])).resolves.toEqual({ ok: true }); - await expect(session.close()).rejects.toThrow( - "Compiled CLI session did not exit after stdin closed", - ); - expect(signals).toEqual(["SIGTERM", "SIGKILL"]); - expect(child?.signalCode).toBe("SIGKILL"); -}); - -test("closes a compiled CLI session that exits after stdin closes", async () => { - const session = startCompiledCliSession({ - spawnWorker: () => { - const child = fakeProcess(82); - exitWhenStdinCloses(child); - ready(child); - replyWhenAsked(child, JSON.stringify({ ok: true })); - return asCliProcess(child); - }, - }); - - await expect(session.run(["once"])).resolves.toEqual({ ok: true }); - await session.close(); -}); - -test("keeps a compiled CLI failure status and serves the next command", async () => { - const session = startCompiledCliSession(); - onTestFinished(() => session.close()); - const root = await createTestTempDirectory("rea-warm-cli-status-"); - await writeFile( - join(root, "app.mjs"), - "export function current() { return 1; }\n", - ); - - await expect( - session.run(["compare-javascript-export-shapes", "{", "--json"]), - ).rejects.toThrow(/status 1:.*invalid_request/); - await expect( - session.run(["analyze-javascript-application", root, "--json"]), - ).resolves.toMatchObject({ - normalized_result: expect.any(Object), - }); - await session.close(); -}); diff --git a/tests/boundary/cli/sharedJavaScriptCliSession.test.ts b/tests/boundary/cli/sharedJavaScriptCliSession.test.ts index 3794f70486..6ef66cc142 100644 --- a/tests/boundary/cli/sharedJavaScriptCliSession.test.ts +++ b/tests/boundary/cli/sharedJavaScriptCliSession.test.ts @@ -1,7 +1,8 @@ import { EventEmitter, PassThrough } from "node:stream"; -import type { ChildProcessWithoutNullStreams } from "node:child_process"; +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import { fileURLToPath } from "node:url"; -import { expect, test } from "vitest"; +import { expect, onTestFinished, test } from "vitest"; import { z } from "zod"; import { javascriptApplicationAnalysisResultSchema } from "../../../src/domain/javascript/javascriptApplicationAnalysis.js"; @@ -126,8 +127,13 @@ test("stops an unready shared CLI process when the session closes", async () => command(["--version"], { PATH: "/usr/bin", REA_LOG_LEVEL: "silent" }), ); await waitFor(() => spawned !== undefined); - await session.close(); - await expect(pending).rejects.toThrow("Shared CLI session is closed"); + const rejected = expect(pending).rejects.toThrow( + "Shared CLI session is closed", + ); + const closing = session.close(); + spawned?.stdout.write(encodeCliSessionFrame({ ready: true })); + await closing; + await rejected; expect(spawned?.signalCode).toBe("SIGTERM"); await expect( session.run( @@ -155,8 +161,11 @@ test("stops an in-flight shared CLI command when the session closes", async () = command(["--version"], { PATH: "/usr/bin", REA_LOG_LEVEL: "silent" }), ); await waitFor(() => received); + const rejected = expect(pending).rejects.toThrow( + "Shared CLI session is closed", + ); await session.close(); - await expect(pending).rejects.toThrow("Shared CLI session is closed"); + await rejected; expect(spawned?.signalCode).toBe("SIGTERM"); }); @@ -249,6 +258,369 @@ const respondingProcess = ( return child; }; +test("recovers after a shared CLI command exceeds its selected deadline", async () => { + const children: FakeCliProcess[] = []; + const session = createCliCommandSession({ + requestTimeoutMs: 20, + shutdownGraceMs: 20, + spawnWorker: () => { + const child = respondingProcess( + 130 + children.length, + (request, current) => { + if (children.length > 1) + reply(current, request.id, { stdout: "recovered" }); + }, + ); + children.push(child); + return asCliProcess(child); + }, + }); + await expect( + session.run(command(["stuck"], { PATH: "/usr/bin" })), + ).rejects.toThrow(/did not respond/); + expect(children[0]?.signalCode).not.toBeNull(); + await expect( + session.run(command(["next"], { PATH: "/usr/bin" })), + ).resolves.toMatchObject({ stdout: "recovered" }); + expect(children).toHaveLength(2); + await session.close(); +}); + +test("times out a shared CLI command whose stdin write callback stays pending", async () => { + const children: FakeCliProcess[] = []; + let releaseWrite: (() => void) | undefined; + let observeWrite: (() => void) | undefined; + const writing = new Promise((resolve) => { + observeWrite = resolve; + }); + const session = createCliCommandSession({ + requestTimeoutMs: 20, + shutdownGraceMs: 20, + spawnWorker: () => { + const child = respondingProcess( + 135 + children.length, + (request, current) => + reply(current, request.id, { stdout: "recovered" }), + ); + if (children.length === 0) { + child.stdin._write = (_chunk, _encoding, callback) => { + releaseWrite = () => callback(); + observeWrite?.(); + }; + } + children.push(child); + return asCliProcess(child); + }, + }); + const failed = expect( + session.run(command(["stuck"], { PATH: "/usr/bin" })), + ).rejects.toThrow(/did not respond/); + await writing; + expect(releaseWrite).toEqual(expect.any(Function)); + await failed; + expect(children[0]?.signalCode).not.toBeNull(); + releaseWrite?.(); + await expect( + session.run(command(["next"], { PATH: "/usr/bin" })), + ).resolves.toMatchObject({ stdout: "recovered" }); + expect(children).toHaveLength(2); + await session.close(); +}); + +test("recovers after a shared CLI worker misses its selected readiness deadline", async () => { + const children: FakeCliProcess[] = []; + const session = createCliCommandSession({ + startTimeoutMs: 20, + shutdownGraceMs: 20, + spawnWorker: () => { + const child = + children.length === 0 + ? fakeProcess(140) + : respondingProcess(141, (request, current) => + reply(current, request.id, { stdout: "ready" }), + ); + children.push(child); + return asCliProcess(child); + }, + }); + await expect( + session.run(command(["stuck"], { PATH: "/usr/bin" })), + ).rejects.toThrow(/did not become ready/); + expect(children[0]?.signalCode).not.toBeNull(); + await expect( + session.run(command(["next"], { PATH: "/usr/bin" })), + ).resolves.toMatchObject({ stdout: "ready" }); + expect(children).toHaveLength(2); + await session.close(); +}); + +test("waits for the timed-out shared CLI worker to exit before starting its replacement", async () => { + const children: FakeCliProcess[] = []; + let releaseExit: (() => void) | undefined; + let observeStop: (() => void) | undefined; + const stopping = new Promise((resolve) => { + observeStop = resolve; + }); + const session = createCliCommandSession({ + requestTimeoutMs: 20, + shutdownGraceMs: 100, + spawnWorker: () => { + const child = respondingProcess( + 150 + children.length, + (request, current) => { + if (children.length > 1) + reply(current, request.id, { stdout: "next" }); + }, + ); + if (children.length === 0) { + const stop = child.kill.bind(child); + child.kill = (signal: NodeJS.Signals = "SIGTERM") => { + releaseExit = () => stop(signal); + observeStop?.(); + return true; + }; + } + children.push(child); + return asCliProcess(child); + }, + }); + const stalled = expect( + session.run(command(["stuck"], { PATH: "/usr/bin" })), + ).rejects.toThrow(/did not respond/); + const next = session.run(command(["next"], { PATH: "/usr/bin" })); + await stopping; + expect(children).toHaveLength(1); + releaseExit?.(); + await stalled; + await expect(next).resolves.toMatchObject({ stdout: "next" }); + expect(children).toHaveLength(2); + await session.close(); +}); + +test("retains a shared CLI worker that survives SIGKILL and retries its exact owner on close", async () => { + const children: FakeCliProcess[] = []; + const signals: NodeJS.Signals[] = []; + const session = createCliCommandSession({ + shutdownGraceMs: 20, + spawnWorker: () => { + const child = respondingProcess( + 160 + children.length, + (request, current) => { + reply(current, request.id, { + exitCode: 1, + stderr: "fatal", + fatal: true, + }); + }, + ); + child.kill = (signal: NodeJS.Signals = "SIGTERM") => { + signals.push(signal); + return true; + }; + children.push(child); + return asCliProcess(child); + }, + }); + await expect( + session.run(command(["fatal"], { PATH: "/usr/bin" })), + ).rejects.toThrow(/did not exit after SIGKILL/); + await expect( + session.run(command(["blocked"], { PATH: "/usr/bin" })), + ).rejects.toThrow(/did not exit after SIGKILL/); + expect(children).toHaveLength(1); + await expect(session.close()).rejects.toThrow(/did not exit after SIGKILL/); + const retained = children[0]; + if (retained === undefined) throw new Error("missing retained worker"); + const stopsBeforeRecovery = signals.length; + retained.kill = (signal: NodeJS.Signals = "SIGTERM") => { + signals.push(signal); + retained.signalCode = signal; + retained.emit("exit", null, signal); + retained.emit("close", null, signal); + return true; + }; + await session.close(); + expect(signals.length).toBeGreaterThan(stopsBeforeRecovery); + expect(retained.signalCode).not.toBeNull(); + expect(children).toHaveLength(1); + await session.close(); +}); + +test.each([ + { + phase: "command", + reason: /did not respond/, + }, + { + phase: "startup", + reason: /did not become ready/, + }, +])( + "preserves the $phase timeout when shared CLI cleanup also fails", + async ({ phase, reason }) => { + const children: FakeCliProcess[] = []; + const signals: NodeJS.Signals[] = []; + const session = createCliCommandSession({ + ...(phase === "command" + ? { requestTimeoutMs: 20 } + : { startTimeoutMs: 20 }), + shutdownGraceMs: 20, + spawnWorker: () => { + const child = + phase === "command" + ? respondingProcess(190, () => undefined) + : fakeProcess(190); + child.kill = (signal: NodeJS.Signals = "SIGTERM") => { + signals.push(signal); + return true; + }; + children.push(child); + return asCliProcess(child); + }, + }); + const failure: unknown = await session + .run(command(["stuck"], { PATH: "/usr/bin" })) + .catch((error: unknown) => error); + expect(failure).toBeInstanceOf(AggregateError); + if (!(failure instanceof AggregateError)) + throw new Error("missing combined command and cleanup failure", { + cause: failure, + }); + expect(failure.message).toMatch(reason); + expect(failure.message).toMatch(/did not exit after SIGKILL/); + expect(failure.errors).toHaveLength(2); + expect(failure.errors[0]).toBeInstanceOf(Error); + expect(failure.errors[0]).toHaveProperty( + "message", + expect.stringMatching(reason), + ); + expect(failure.errors[1]).toHaveProperty( + "message", + expect.stringMatching(/did not exit after SIGKILL/), + ); + expect(failure.cause).toBe(failure.errors[0]); + expect(signals).toEqual(["SIGTERM", "SIGKILL"]); + await expect( + session.run(command(["blocked"], { PATH: "/usr/bin" })), + ).rejects.toThrow(/did not exit after SIGKILL/); + expect(children).toHaveLength(1); + await expect(session.close()).rejects.toThrow(/did not exit after SIGKILL/); + const retained = children[0]; + if (retained === undefined) throw new Error("missing retained worker"); + retained.kill = (signal: NodeJS.Signals = "SIGTERM") => { + signals.push(signal); + retained.signalCode = signal; + retained.emit("exit", null, signal); + retained.emit("close", null, signal); + return true; + }; + await session.close(); + expect(retained.signalCode).not.toBeNull(); + expect(children).toHaveLength(1); + }, +); + +test("confirms shared CLI shutdown after escalating an ignored SIGTERM", async () => { + let child: FakeCliProcess | undefined; + const signals: NodeJS.Signals[] = []; + const session = createCliCommandSession({ + shutdownGraceMs: 20, + spawnWorker: () => { + const started = respondingProcess(170, (request, current) => + reply(current, request.id, { stdout: "done" }), + ); + const stop = started.kill.bind(started); + started.kill = (signal: NodeJS.Signals = "SIGTERM") => { + signals.push(signal); + return signal === "SIGKILL" ? stop(signal) : true; + }; + child = started; + return asCliProcess(started); + }, + }); + await expect( + session.run(command(["once"], { PATH: "/usr/bin" })), + ).resolves.toMatchObject({ stdout: "done" }); + await session.close(); + expect(signals).toEqual(["SIGTERM", "SIGKILL"]); + expect(child?.signalCode).toBe("SIGKILL"); + await session.close(); +}); + +test("lets the shared CLI worker finish its resource cleanup before signaling", async () => { + const events: string[] = []; + const session = createCliCommandSession({ + spawnWorker: () => { + const child = respondingProcess(180, (request, current) => + reply(current, request.id, { stdout: "done" }), + ); + child.stdin.once("finish", () => { + events.push("stdin closed", "resources released"); + child.exitCode = 0; + child.emit("exit", 0, null); + child.emit("close", 0, null); + }); + const stop = child.kill.bind(child); + child.kill = (signal: NodeJS.Signals = "SIGTERM") => { + events.push(signal); + return stop(signal); + }; + return asCliProcess(child); + }, + }); + await expect( + session.run(command(["once"], { PATH: "/usr/bin" })), + ).resolves.toMatchObject({ stdout: "done" }); + await session.close(); + expect(events).toEqual(["stdin closed", "resources released"]); + await session.close(); + expect(events).toHaveLength(2); +}); + +test("shares one shared CLI cleanup attempt across concurrent close calls", async () => { + let cleanupStarts = 0; + let releaseExit: (() => void) | undefined; + let observeCleanup: (() => void) | undefined; + const cleaning = new Promise((resolve) => { + observeCleanup = resolve; + }); + const signals: NodeJS.Signals[] = []; + const session = createCliCommandSession({ + spawnWorker: () => { + const child = respondingProcess(185, (request, current) => + reply(current, request.id, { stdout: "done" }), + ); + child.stdin.once("finish", () => { + cleanupStarts += 1; + releaseExit = () => { + child.exitCode = 0; + child.emit("exit", 0, null); + child.emit("close", 0, null); + }; + observeCleanup?.(); + }); + const stop = child.kill.bind(child); + child.kill = (signal: NodeJS.Signals = "SIGTERM") => { + signals.push(signal); + return stop(signal); + }; + return asCliProcess(child); + }, + }); + await session.run(command(["once"], { PATH: "/usr/bin" })); + const first = session.close(); + const second = session.close(); + expect(second).toBe(first); + await cleaning; + expect(cleanupStarts).toBe(1); + expect(signals).toEqual([]); + releaseExit?.(); + await Promise.all([first, second]); + await session.close(); + expect(cleanupStarts).toBe(1); + expect(signals).toEqual([]); +}); + test("starts another process after the shared CLI process exits", async () => { const children: FakeCliProcess[] = []; const session = createCliCommandSession({ @@ -457,6 +829,81 @@ test("starts a new process after the real worker reports a fatal failure", async await session.close(); }); +test("keeps a malformed JSON command's failure status and serves the next command", async () => { + const session = createCliCommandSession(); + onTestFinished(() => session.close()); + const environment = { + PATH: process.env.PATH ?? "", + REA_LOG_LEVEL: "silent", + }; + const failed = await session.run( + command(["compare-javascript-export-shapes", "{", "--json"], environment), + ); + expect(failed.exitCode).toBe(1); + expect(failed.stdout).toContain("invalid_request"); + const worker = session.processId(); + expect(worker).toEqual(expect.any(Number)); + const root = await createTestTempDirectory("rea-warm-cli-status-"); + await writeFixtureFiles(root, { + "app.mjs": "export function current() { return 1; }\n", + }); + const recovered = await session.run( + command(["analyze-javascript-application", root, "--json"], environment), + ); + expect(recovered.exitCode, recovered.stderr).toBe(0); + expect(recovered.json).toMatchObject({ + normalized_result: expect.any(Object), + }); + expect(session.processId()).toBe(worker); + await session.close(); +}); + +test("closes the real JavaScript CLI worker through its resource cleanup", async () => { + let worker: ChildProcessWithoutNullStreams | undefined; + const session = createCliCommandSession({ + spawnWorker: (environment) => { + worker = spawn( + process.execPath, + [ + fileURLToPath( + new URL( + "../../support/cli/cliCommandSessionWorker.mjs", + import.meta.url, + ), + ), + ], + { + cwd: process.cwd(), + env: environment, + stdio: ["pipe", "pipe", "pipe"], + }, + ); + return worker; + }, + }); + onTestFinished(() => session.close()); + const root = await createTestTempDirectory("rea-warm-cli-graceful-"); + await writeFixtureFiles(root, { + "app.mjs": "export function current() { return 1; }\n", + }); + const analyzed = await session.run( + command(["analyze-javascript-application", root, "--json"], { + PATH: process.env.PATH ?? "", + REA_LOG_LEVEL: "silent", + }), + ); + expect(analyzed.exitCode, analyzed.stderr).toBe(0); + expect(analyzed.json).toMatchObject({ + normalized_result: expect.any(Object), + }); + expect(worker?.exitCode).toBeNull(); + expect(worker?.signalCode).toBeNull(); + await session.close(); + expect(worker?.exitCode).toBe(0); + expect(worker?.signalCode).toBeNull(); + expect(session.processId()).toBeUndefined(); +}); + const requireSharedCli = (cli: { sharedCliRun?: (invocation: CliInvocation) => Promise; }): ((invocation: CliInvocation) => Promise) => { diff --git a/tests/support/cli/cliCommandSession.ts b/tests/support/cli/cliCommandSession.ts index 1805836054..96ed81b521 100644 --- a/tests/support/cli/cliCommandSession.ts +++ b/tests/support/cli/cliCommandSession.ts @@ -61,12 +61,23 @@ interface CliCommandSessionDependencies { readonly commandLimit?: number; /** Resident-set bytes, reported by the worker, that start a new process. */ readonly residentLimitBytes?: number; + /** Optional caller-selected startup and command deadlines; unset by default. */ + readonly startTimeoutMs?: number; + readonly requestTimeoutMs?: number; + /** Grace for EOF cleanup and each signal before escalation. */ + readonly shutdownGraceMs?: number; } interface PendingCommand { readonly id: number; readonly resolve: (result: CliCommandResult) => void; readonly reject: (error: unknown) => void; + readonly timer: ReturnType | undefined; +} + +interface RetiredWorker { + readonly child: ChildProcessWithoutNullStreams; + attempt: Promise | undefined; } type SessionResponse = z.infer; @@ -78,6 +89,11 @@ const positiveLimit = (value: number | undefined, fallback: number): number => ? value : fallback; +const selectedTimeout = (value: number | undefined): number | undefined => + value !== undefined && Number.isSafeInteger(value) && value > 0 + ? value + : undefined; + const envKey = (environment: Readonly>): string => JSON.stringify( Object.keys(environment) @@ -110,13 +126,17 @@ const signalChild = ( const stopChild = async ( child: ChildProcessWithoutNullStreams, + graceMs: number, ): Promise => { if (child.exitCode !== null || child.signalCode !== null) return; + // EOF lets the worker await closeCliJavaScriptResources before exiting. + if (!child.stdin.writableEnded) child.stdin.end(); + if (await waitForExit(child, graceMs)) return; signalChild(child, "SIGTERM"); - if (await waitForExit(child, SHUTDOWN_GRACE_MS)) return; + if (await waitForExit(child, graceMs)) return; if (child.exitCode !== null || child.signalCode !== null) return; signalChild(child, "SIGKILL"); - if (await waitForExit(child, SHUTDOWN_GRACE_MS)) return; + if (await waitForExit(child, graceMs)) return; throw new Error( `shared CLI process ${String(child.pid)} did not exit after SIGKILL`, ); @@ -157,11 +177,14 @@ class SharedCliProcess implements CliCommandSession { readonly #commandLimit: number; readonly #residentLimitBytes: number; + readonly #startTimeoutMs: number | undefined; + readonly #requestTimeoutMs: number | undefined; + readonly #shutdownGraceMs: number; #child: ChildProcessWithoutNullStreams | undefined; #childEnvKey: string | undefined; #commands = 0; #residentBytes: number | undefined; - #retiring: Promise | undefined; + #retiring: RetiredWorker | undefined; #closed = false; #closing: Promise | undefined; #starting: Promise | undefined; @@ -181,6 +204,12 @@ class SharedCliProcess implements CliCommandSession { dependencies.residentLimitBytes, RESIDENT_REUSE_LIMIT_BYTES, ); + this.#startTimeoutMs = selectedTimeout(dependencies.startTimeoutMs); + this.#requestTimeoutMs = selectedTimeout(dependencies.requestTimeoutMs); + this.#shutdownGraceMs = positiveLimit( + dependencies.shutdownGraceMs, + SHUTDOWN_GRACE_MS, + ); } run(request: CliCommandRequest): Promise { @@ -207,45 +236,83 @@ class SharedCliProcess implements CliCommandSession { close(): Promise { this.#closed = true; if (this.#closing !== undefined) return this.#closing; - const attempt = this.#shutdown(); + let attempt: Promise; + attempt = this.#shutdown().catch((error: unknown) => { + if (this.#closing === attempt) this.#closing = undefined; + throw error; + }); this.#closing = attempt; return attempt; } async #shutdown(): Promise { - try { - const owner = this.#child; - if (owner !== undefined) await stopChild(owner); - const retiring = this.#retiring; - if (retiring !== undefined) await retiring; - } catch (error) { - this.#closing = undefined; - throw error; - } const pendingStart = this.#starting; + this.#failReady(closedError()); + this.#failPending(closedError()); + const owner = this.#child; + if (owner !== undefined) this.#retire(owner); + await this.#finishRetirement(); if (pendingStart !== undefined) await pendingStart.catch(() => undefined); } async #invoke(request: CliCommandRequest): Promise { if (this.#closed) throw closedError(); - try { - return await this.#run(request); - } finally { + await this.#finishRetirement(); + const current = this.#child; + if ( + !this.#reusable(current, envKey(request.environment)) && + current !== undefined + ) { + this.#retire(current); await this.#finishRetirement(); } + if (this.#closed) throw closedError(); + let result: CliCommandResult; + try { + if (this.#child === undefined) await this.#start(request.environment); + result = await this.#run(request); + } catch (primary) { + try { + await this.#finishRetirement(); + } catch (cleanup) { + const primaryReason = + primary instanceof Error ? primary.message : String(primary); + const cleanupReason = + cleanup instanceof Error ? cleanup.message : String(cleanup); + throw new AggregateError( + [primary, cleanup], + `${primaryReason}; shared CLI cleanup failed: ${cleanupReason}`, + { cause: primary }, + ); + } + throw primary; + } + await this.#finishRetirement(); + return result; } async #run(request: CliCommandRequest): Promise { - await this.#ensure(request.environment); const owner = this.#child; if (this.#closed || owner === undefined) throw closedError(); const id = this.#nextId; this.#nextId += 1; const result = new Promise((resolve, reject) => { - this.#pending = { id, resolve, reject }; + const timer = + this.#requestTimeoutMs === undefined + ? undefined + : setTimeout(() => { + this.#failTransport( + owner, + new Error( + `Shared CLI command did not respond within ${String(this.#requestTimeoutMs)} ms`, + ), + ); + }, this.#requestTimeoutMs); + timer?.unref(); + this.#pending = { id, resolve, reject, timer }; }); try { - await writeChild( + void writeChild( owner, encodeCliSessionFrame({ id, @@ -253,7 +320,9 @@ class SharedCliProcess implements CliCommandSession { cwd: request.cwd, env: request.environment, }), - ); + ).catch((error: unknown) => { + this.#failTransport(owner, error); + }); } catch (error) { this.#failTransport(owner, error); } @@ -263,11 +332,14 @@ class SharedCliProcess implements CliCommandSession { async #finishRetirement(): Promise { const retiring = this.#retiring; if (retiring === undefined) return; - try { - await retiring; - } finally { - if (this.#retiring === retiring) this.#retiring = undefined; - } + retiring.attempt ??= stopChild( + retiring.child, + this.#shutdownGraceMs, + ).finally(() => { + retiring.attempt = undefined; + }); + await retiring.attempt; + if (this.#retiring === retiring) this.#retiring = undefined; } #reusable( @@ -289,16 +361,6 @@ class SharedCliProcess implements CliCommandSession { ); } - async #ensure(environment: Readonly>): Promise { - if (this.#closed) throw closedError(); - const key = envKey(environment); - const current = this.#child; - if (this.#reusable(current, key)) return; - if (current !== undefined) await stopChild(current); - if (this.#closed) throw closedError(); - await this.#start(environment); - } - #start(environment: Readonly>): Promise { if (this.#closed) return Promise.reject(closedError()); const started = this.#spawnWorker(environment); @@ -311,12 +373,25 @@ class SharedCliProcess implements CliCommandSession { this.#rejectReady = reject; }); this.#starting = ready; + const timer = + this.#startTimeoutMs === undefined + ? undefined + : setTimeout(() => { + this.#failTransport( + started, + new Error( + `Shared CLI worker did not become ready within ${String(this.#startTimeoutMs)} ms`, + ), + ); + }, this.#startTimeoutMs); + timer?.unref(); this.#watch(started); if (this.#closed) { - signalChild(started, "SIGTERM"); this.#failReady(closedError()); + this.#retire(started); } return ready.finally(() => { + clearTimeout(timer); if (this.#starting === ready) this.#starting = undefined; }); } @@ -378,6 +453,7 @@ class SharedCliProcess implements CliCommandSession { return; } this.#pending = undefined; + clearTimeout(current.timer); if (message.fatal === true) { this.#retire(owner); } else { @@ -394,7 +470,7 @@ class SharedCliProcess implements CliCommandSession { } this.#commands = 0; this.#residentBytes = undefined; - this.#retiring = stopChild(owner); + this.#retiring ??= { child: owner, attempt: undefined }; } #onClose( @@ -427,6 +503,7 @@ class SharedCliProcess implements CliCommandSession { #failPending(error: unknown): void { const current = this.#pending; this.#pending = undefined; + clearTimeout(current?.timer); current?.reject(error); } diff --git a/tests/support/cli/compiledCliSession.ts b/tests/support/cli/compiledCliSession.ts deleted file mode 100644 index 086efa6806..0000000000 --- a/tests/support/cli/compiledCliSession.ts +++ /dev/null @@ -1,444 +0,0 @@ -import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; -import { fileURLToPath } from "node:url"; - -import { z } from "zod"; - -import { waitForExit } from "../process/processFixture.js"; - -const START_TIMEOUT_MS = 20_000; -const REQUEST_TIMEOUT_MS = 20_000; -const STOP_TIMEOUT_MS = 1_000; -const STDERR_TAIL_CHARS = 4_000; -const workerPath = fileURLToPath( - new URL("./compiledCliWorker.mjs", import.meta.url), -); -const readyMessage = z.object({ ready: z.literal(true) }).strict(); -const responseMessage = z - .object({ - id: z.number(), - exitCode: z.number(), - stdout: z.string(), - error: z.string().optional(), - }) - .strict(); - -type CliResponse = z.infer; - -interface PendingResponse { - readonly resolve: (response: CliResponse) => void; - readonly reject: (cause: unknown) => void; -} - -interface OwnedStop { - readonly child: ChildProcessWithoutNullStreams; - attempt: Promise | undefined; -} - -interface SessionState { - child: ChildProcessWithoutNullStreams | undefined; - starting: Promise | undefined; - buffer: string; - nextId: number; - readonly pending: Map; - readonly stopping: OwnedStop[]; - queue: Promise; - stderr: string; - closed: boolean; - closing: Promise | undefined; - settleStart: ((cause?: Error) => void) | undefined; - ready: boolean; - readonly requestTimeoutMs: number; - readonly stopTimeoutMs: number; - readonly spawnWorker: () => ChildProcessWithoutNullStreams; -} - -/** One warm compiled CLI process serving production JSON commands. */ -export interface CompiledCliSession { - run(args: readonly string[]): Promise; - close(): Promise; -} - -/** Limits and process factory for session tests. Production callers omit them. */ -export interface CompiledCliSessionOptions { - readonly spawnWorker?: () => ChildProcessWithoutNullStreams; - readonly requestTimeoutMs?: number; - readonly stopTimeoutMs?: number; -} - -/** Start a session that loads the compiled CLI once and reuses that process. */ -export const startCompiledCliSession = ( - options: CompiledCliSessionOptions = {}, -): CompiledCliSession => { - const state = createSessionState(options); - return { - run: (args) => enqueue(state, () => dispatch(state, args)), - close: () => closeSession(state), - }; -}; - -const positiveMilliseconds = ( - value: number | undefined, - fallback: number, -): number => - value !== undefined && Number.isSafeInteger(value) && value > 0 - ? value - : fallback; - -const defaultSpawnWorker = (): ChildProcessWithoutNullStreams => - spawn(process.execPath, [workerPath], { - cwd: process.cwd(), - env: { ...process.env, REA_LOG_LEVEL: "silent" }, - stdio: ["pipe", "pipe", "pipe"], - }); - -const createSessionState = ( - options: CompiledCliSessionOptions, -): SessionState => ({ - child: undefined, - starting: undefined, - buffer: "", - nextId: 1, - pending: new Map(), - stopping: [], - queue: Promise.resolve(), - stderr: "", - closed: false, - closing: undefined, - settleStart: undefined, - ready: false, - requestTimeoutMs: positiveMilliseconds( - options.requestTimeoutMs, - REQUEST_TIMEOUT_MS, - ), - stopTimeoutMs: positiveMilliseconds(options.stopTimeoutMs, STOP_TIMEOUT_MS), - spawnWorker: options.spawnWorker ?? defaultSpawnWorker, -}); - -const enqueue = ( - state: SessionState, - task: () => Promise, -): Promise => { - const run = state.queue.then(task); - state.queue = run.then( - () => undefined, - () => undefined, - ); - return run; -}; - -const dispatch = async ( - state: SessionState, - args: readonly string[], -): Promise => { - await ensureStarted(state); - const child = state.child; - if (child === undefined) - throw new Error("Compiled CLI session is not running"); - const id = state.nextId; - state.nextId += 1; - const response = await request(state, child, id, args); - if (response.exitCode !== 0) throw new Error(failureText(state, response)); - return JSON.parse(response.stdout) as unknown; -}; - -const request = ( - state: SessionState, - child: ChildProcessWithoutNullStreams, - id: number, - args: readonly string[], -): Promise => - new Promise((resolve, reject) => { - let settled = false; - let timer: ReturnType | undefined; - const settle = (finish: () => void): void => { - if (settled) return; - settled = true; - clearTimeout(timer); - state.pending.delete(id); - finish(); - }; - timer = setTimeout(() => { - settle(() => { - reject( - new Error( - `Compiled CLI command did not respond within ${String(state.requestTimeoutMs)} ms`, - ), - ); - retireChild(state, child); - }); - }, state.requestTimeoutMs); - timer.unref(); - state.pending.set(id, { - resolve: (response) => { - settle(() => { - resolve(response); - }); - }, - reject: (cause) => { - settle(() => { - reject(cause); - }); - }, - }); - child.stdin.write(`${JSON.stringify({ id, args })}\n`, (cause) => { - if (cause === undefined || cause === null) return; - settle(() => { - reject(cause); - }); - }); - }); - -const failureText = (state: SessionState, response: CliResponse): string => { - const detail = response.error ?? response.stdout.slice(0, 500); - const stderr = state.stderr.trim(); - return stderr.length === 0 - ? `Compiled CLI command failed with status ${String(response.exitCode)}: ${detail}` - : `Compiled CLI command failed with status ${String(response.exitCode)}: ${detail}\n${stderr}`; -}; - -const ensureStarted = async (state: SessionState): Promise => { - if (state.closed) throw sessionClosed(); - await releaseRetired(state); - if (state.closed) throw sessionClosed(); - if (state.ready && state.child !== undefined) return; - state.starting ??= launch(state); - await state.starting; - if (state.closed) throw sessionClosed(); -}; - -const launch = (state: SessionState): Promise => - new Promise((resolve, reject) => { - const child = state.spawnWorker(); - state.child = child; - let settled = false; - let timer: ReturnType | undefined; - const finish = (cause?: Error): void => { - if (settled) return; - settled = true; - clearTimeout(timer); - if (state.settleStart === finish) state.settleStart = undefined; - if (cause === undefined) resolve(); - else reject(cause); - }; - timer = setTimeout(() => { - signalChild(child, "SIGKILL"); - finish( - new Error( - `Compiled CLI session did not start within ${String(START_TIMEOUT_MS)} ms\n${state.stderr}`, - ), - ); - }, START_TIMEOUT_MS); - timer.unref(); - state.settleStart = finish; - const fail = (cause: unknown): void => { - finish(cause instanceof Error ? cause : new Error(String(cause))); - }; - const currentChild = (): boolean => state.child === child; - child.once("error", fail); - child.once("exit", (code, signal) => { - if (!currentChild()) return; - state.child = undefined; - state.ready = false; - state.starting = undefined; - state.buffer = ""; - const cause = new Error( - `Compiled CLI session exited (${String(code)}, ${String(signal)})\n${state.stderr}`, - ); - fail(cause); - rejectPending(state, cause); - }); - child.stderr.setEncoding("utf8"); - child.stderr.on("data", (chunk: string) => { - if (!currentChild()) return; - state.stderr = `${state.stderr}${chunk}`.slice(-STDERR_TAIL_CHARS); - }); - child.stdout.setEncoding("utf8"); - child.stdout.on("data", (chunk: string) => { - if (!currentChild()) return; - state.buffer += chunk; - drainLines(state, () => { - state.ready = true; - finish(); - }); - }); - }); - -const hasExited = (child: ChildProcessWithoutNullStreams): boolean => - child.exitCode !== null || child.signalCode !== null; - -const signalChild = ( - child: ChildProcessWithoutNullStreams, - signal: NodeJS.Signals, -): void => { - try { - child.kill(signal); - } catch (error) { - const code = - error instanceof Error && "code" in error ? error.code : undefined; - if (code !== "ESRCH") throw error; - } -}; - -const detachChild = ( - state: SessionState, - child: ChildProcessWithoutNullStreams, -): void => { - if (state.child !== child) return; - state.child = undefined; - state.ready = false; - state.starting = undefined; - state.buffer = ""; -}; - -const stopOwned = (state: SessionState, entry: OwnedStop): Promise => { - entry.attempt ??= (async () => { - if (hasExited(entry.child)) return true; - signalChild(entry.child, "SIGKILL"); - return waitForExit(entry.child, state.stopTimeoutMs); - })().finally(() => { - entry.attempt = undefined; - }); - return entry.attempt; -}; - -const trackRetired = ( - state: SessionState, - child: ChildProcessWithoutNullStreams, -): OwnedStop => { - const existing = state.stopping.find((entry) => entry.child === child); - if (existing !== undefined) return existing; - const entry: OwnedStop = { child, attempt: undefined }; - state.stopping.push(entry); - if (!hasExited(child)) void stopOwned(state, entry).catch(() => undefined); - return entry; -}; - -const releaseRetired = async (state: SessionState): Promise => { - let failed = false; - for (const entry of [...state.stopping]) { - if (hasExited(entry.child)) continue; - const stopped = await stopOwned(state, entry).catch(() => false); - if (!stopped || !hasExited(entry.child)) failed = true; - } - const retained = state.stopping.filter((entry) => !hasExited(entry.child)); - state.stopping.length = 0; - state.stopping.push(...retained); - if (failed) - throw new Error("Compiled CLI session stayed running after SIGKILL"); -}; - -const retireChild = ( - state: SessionState, - child: ChildProcessWithoutNullStreams, -): void => { - detachChild(state, child); - if (hasExited(child)) return; - trackRetired(state, child); -}; - -const drainLines = (state: SessionState, onReady: () => void): void => { - let newline = state.buffer.indexOf("\n"); - while (newline !== -1) { - const line = state.buffer.slice(0, newline); - state.buffer = state.buffer.slice(newline + 1); - if (line.length > 0) acceptLine(state, line, onReady); - newline = state.buffer.indexOf("\n"); - } -}; - -const acceptLine = ( - state: SessionState, - line: string, - onReady: () => void, -): void => { - let message: unknown; - try { - message = JSON.parse(line) as unknown; - } catch (cause) { - rejectPending( - state, - cause instanceof Error ? cause : new Error(String(cause)), - ); - return; - } - if (readyMessage.safeParse(message).success) { - onReady(); - return; - } - const parsed = responseMessage.safeParse(message); - if (!parsed.success) { - rejectPending( - state, - new Error("Compiled CLI session returned an unexpected message"), - ); - return; - } - const pending = state.pending.get(parsed.data.id); - state.pending.delete(parsed.data.id); - if (pending === undefined) { - rejectPending( - state, - new Error( - `Compiled CLI session returned unexpected id ${String(parsed.data.id)}`, - ), - ); - return; - } - pending.resolve(parsed.data); -}; - -const rejectPending = (state: SessionState, cause: Error): void => { - for (const pending of state.pending.values()) pending.reject(cause); - state.pending.clear(); -}; - -const closeSession = (state: SessionState): Promise => { - state.closed = true; - if (state.closing !== undefined) return state.closing; - let attempt: Promise; - attempt = shutdown(state).catch((error: unknown) => { - if (state.closing === attempt) state.closing = undefined; - throw error; - }); - state.closing = attempt; - return attempt; -}; - -const shutdown = async (state: SessionState): Promise => { - const child = state.child; - let forced = false; - const stillRunning: OwnedStop[] = []; - if (child !== undefined) { - rejectPending(state, sessionClosed()); - state.settleStart?.(sessionClosed()); - detachChild(state, child); - const stopped = await stopGracefully(state, child); - if (stopped === "forced") forced = true; - if (stopped === "stuck" && !hasExited(child)) - stillRunning.push({ child, attempt: undefined }); - } - try { - await releaseRetired(state); - } finally { - state.stopping.push(...stillRunning); - } - if (state.stopping.length > 0) - throw new Error("Compiled CLI session stayed running after SIGKILL"); - if (forced) - throw new Error("Compiled CLI session did not exit after stdin closed"); -}; - -const stopGracefully = async ( - state: SessionState, - child: ChildProcessWithoutNullStreams, -): Promise<"clean" | "forced" | "stuck"> => { - child.stdin.end(); - if (await waitForExit(child, state.stopTimeoutMs)) return "clean"; - signalChild(child, "SIGTERM"); - if (await waitForExit(child, state.stopTimeoutMs)) return "forced"; - signalChild(child, "SIGKILL"); - if (await waitForExit(child, state.stopTimeoutMs)) return "forced"; - return "stuck"; -}; - -const sessionClosed = (): Error => new Error("Compiled CLI session is closed"); diff --git a/tests/support/cli/compiledCliWorker.mjs b/tests/support/cli/compiledCliWorker.mjs deleted file mode 100644 index fc16840989..0000000000 --- a/tests/support/cli/compiledCliWorker.mjs +++ /dev/null @@ -1,151 +0,0 @@ -import { Writable } from "node:stream"; -import { createInterface } from "node:readline"; - -const { closeCliJavaScriptResources, createCli } = - await import("../../../dist/cli-runtime/cli.js"); -const { createCliResultOutput } = - await import("../../../dist/cli-runtime/cliResultOutput.js"); -const { - cliConfigurationError, - normalizeCliArguments, - renderCliOutputArgumentError, - renderEmptyFilteredCliOutput, - parseCliOutputArguments, -} = await import("../../../dist/cli-runtime/cliOutput.js"); - -const MAX_STDOUT_CHARS = 16 * 1024 * 1024; - -const writeMessage = (message) => { - process.stdout.write(`${JSON.stringify(message)}\n`); -}; - -// One CLI instance serves every command. A fresh instance per command retains -// its command registrations after the call returns; production exits instead. -const outputParts = []; -let outputCharacters = 0; -const destination = new Writable({ - write(chunk, _encoding, callback) { - const text = Buffer.isBuffer(chunk) - ? chunk.toString("utf8") - : String(chunk); - outputCharacters += text.length; - if (outputCharacters > MAX_STDOUT_CHARS) { - callback(new Error("Compiled CLI stdout exceeded 16 MiB")); - return; - } - outputParts.push(text); - callback(); - }, -}); -Object.defineProperty(destination, "isTTY", { value: false }); -const resultOutput = createCliResultOutput( - parseCliOutputArguments(["--json"]), - destination, -); -const cli = createCli(process.env, resultOutput); - -const resetOutput = () => { - outputParts.length = 0; - outputCharacters = 0; -}; - -const runCommand = async (args) => { - const previousExitCode = process.exitCode; - process.exitCode = undefined; - resetOutput(); - try { - const cliArgs = normalizeCliArguments(args); - const outputOptions = parseCliOutputArguments(cliArgs); - const outputArguments = outputOptions.validation; - if (!outputArguments.ok) - return { - exitCode: 1, - stdout: renderCliOutputArgumentError(outputArguments), - }; - const configurationError = cliConfigurationError(process.env); - if (configurationError !== undefined) - return { exitCode: 1, stdout: "", error: configurationError }; - - let wroteOutput = false; - let explicitExit; - resultOutput?.begin(outputOptions); - try { - await cli.serve(cliArgs, { - stdout: (output) => { - if (resultOutput?.handled) return; - if (output.length === 0) return; - wroteOutput = true; - if (resultOutput?.captureSdkOutput(output)) return; - destination.write(output); - }, - exit: (code) => { - explicitExit = code; - }, - }); - } catch (cause) { - return { - exitCode: 1, - stdout: outputParts.join(""), - error: - cause instanceof Error - ? (cause.stack ?? cause.message) - : String(cause), - }; - } - if (!wroteOutput && !resultOutput?.handled) { - const empty = renderEmptyFilteredCliOutput(outputOptions); - if (empty !== undefined) destination.write(empty); - } - await resultOutput?.finishSdkOutput( - explicitExit ?? - (typeof process.exitCode === "number" ? process.exitCode : 0), - ); - // logCliCommand reports some failures only through process.exitCode. - return { - exitCode: - explicitExit ?? - (typeof process.exitCode === "number" ? process.exitCode : 0), - stdout: outputParts.join(""), - }; - } finally { - process.exitCode = previousExitCode; - } -}; - -const acceptRequest = async (line) => { - let request; - try { - request = JSON.parse(line); - } catch (cause) { - writeMessage({ - id: -1, - exitCode: 1, - stdout: "", - error: cause instanceof Error ? cause.message : String(cause), - }); - return; - } - const args = request?.args; - if ( - typeof request?.id !== "number" || - !Array.isArray(args) || - args.some((argument) => typeof argument !== "string") - ) { - writeMessage({ - id: typeof request?.id === "number" ? request.id : -1, - exitCode: 1, - stdout: "", - error: "Compiled CLI session expected an id and string arguments", - }); - return; - } - writeMessage({ id: request.id, ...(await runCommand(args)) }); -}; - -const lines = createInterface({ input: process.stdin, crlfDelay: Infinity }); -writeMessage({ ready: true }); -for await (const line of lines) { - if (line.trim().length === 0) continue; - await acceptRequest(line); -} -await closeCliJavaScriptResources();