diff --git a/docs/testing.md b/docs/testing.md index e0b3f39a77..ab8e501a5f 100644 --- a/docs/testing.md +++ b/docs/testing.md @@ -156,7 +156,15 @@ rejection without a reason. That cleanup stays owned until a later close or call releases it. Another server does not open while the cleanup remains, and a cleanup that already succeeded is not run again. Pass `mcpCall` when a case needs its own server, environment, or lifecycle. CLI -cases still start a fresh process. +cases for that helper share one compiled CLI process per child environment. +The first case pays process startup; later cases reuse the loaded runtime. +The process restarts when the environment changes, after 32 commands, or when +the worker reports a resident set of at least 1 GiB. A fatal worker failure +or a broken session frame retires that process and waits for it to exit before +the next command starts. +`cli.run` still starts a fresh process, including stdin and explicit deadlines. +File teardown stops the shared CLI process, including startup still in flight, +and does not open another after close. The Apple artifact verifier step shares a job-temporary bytecode cache among its fresh CLI/MCP children via `REA_ARTIFACT_NODE_COMPILE_CACHE`. The artifact diff --git a/tests/boundary/cli/sharedJavaScriptCliSession.test.ts b/tests/boundary/cli/sharedJavaScriptCliSession.test.ts new file mode 100644 index 0000000000..3794f70486 --- /dev/null +++ b/tests/boundary/cli/sharedJavaScriptCliSession.test.ts @@ -0,0 +1,577 @@ +import { EventEmitter, PassThrough } from "node:stream"; +import type { ChildProcessWithoutNullStreams } from "node:child_process"; + +import { expect, test } from "vitest"; +import { z } from "zod"; + +import { javascriptApplicationAnalysisResultSchema } from "../../../src/domain/javascript/javascriptApplicationAnalysis.js"; +import { createTestTempDirectory } from "../../fixtures/temporaryDirectory.js"; +import { + cliTest, + type CliInvocation, + type CliResult, +} from "../../support/cli/cliFixture.js"; +import { createCliCommandSession } from "../../support/cli/cliCommandSession.js"; +import { + createCliSessionFrameReader, + encodeCliSessionFrame, + MAX_CLI_SESSION_FRAME_BYTES, +} from "../../support/cli/cliCommandSessionFrames.mjs"; +import { analyzeJavaScriptDirectory } from "../../support/cli/javascriptApplication.js"; +import { writeFixtureFiles } from "../../support/workspace/writeFixtureFiles.js"; + +const requestSchema = z.object({ + id: z.number(), + arguments: z.array(z.string()), +}); + +const command = ( + arguments_: readonly string[], + environment: Readonly>, +): CliInvocation & { + cwd: string; + environment: Readonly>; +} => ({ + arguments: arguments_, + cwd: process.cwd(), + environment, +}); + +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 waitFor = async (ready: () => boolean): Promise => { + for (let attempt = 0; attempt < 30; attempt += 1) { + if (ready()) return; + await new Promise((resolve) => { + setImmediate(resolve); + }); + } + throw new Error("shared CLI process did not reach the expected state"); +}; + +test("reads a CLI session frame split across chunks", () => { + const frames: unknown[] = []; + const reader = createCliSessionFrameReader((frame) => { + frames.push(frame); + }); + const encoded = encodeCliSessionFrame({ ready: true, seq: 1 }); + const second = encodeCliSessionFrame({ id: 2 }); + const combined = Buffer.concat([encoded, second]); + reader(combined.subarray(0, 1)); + expect(frames).toEqual([]); + reader(combined.subarray(1, 3)); + expect(frames).toEqual([]); + reader(combined.subarray(3)); + expect(frames).toEqual([{ ready: true, seq: 1 }, { id: 2 }]); +}); + +test("rejects an oversized CLI session frame", () => { + const reader = createCliSessionFrameReader(() => undefined); + const header = Buffer.alloc(4); + header.writeUInt32BE(MAX_CLI_SESSION_FRAME_BYTES + 1); + expect(() => reader(header)).toThrow(/limit/); +}); + +test("reads a CLI session frame delivered one byte at a time", () => { + const frames: unknown[] = []; + const reader = createCliSessionFrameReader((frame) => { + frames.push(frame); + }); + const encoded = encodeCliSessionFrame({ + id: 7, + note: "x".repeat(64), + }); + for (const byte of encoded) reader(Buffer.from([byte])); + expect(frames).toEqual([{ id: 7, note: "x".repeat(64) }]); +}); + +test("stops an unready shared CLI process when the session closes", async () => { + let spawned: FakeCliProcess | undefined; + const session = createCliCommandSession({ + spawnWorker: () => { + spawned = fakeProcess(41); + return asCliProcess(spawned); + }, + }); + const pending = session.run( + 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"); + expect(spawned?.signalCode).toBe("SIGTERM"); + await expect( + session.run( + command(["--version"], { PATH: "/usr/bin", REA_LOG_LEVEL: "silent" }), + ), + ).rejects.toThrow("Shared CLI session is closed"); + await session.close(); +}); + +test("stops an in-flight shared CLI command when the session closes", async () => { + let received = false; + let spawned: FakeCliProcess | undefined; + const session = createCliCommandSession({ + spawnWorker: () => { + const child = fakeProcess(42); + spawned = child; + child.stdout.write(encodeCliSessionFrame({ ready: true })); + child.stdin.on("data", () => { + received = true; + }); + return asCliProcess(child); + }, + }); + const pending = session.run( + command(["--version"], { PATH: "/usr/bin", REA_LOG_LEVEL: "silent" }), + ); + await waitFor(() => received); + await session.close(); + await expect(pending).rejects.toThrow("Shared CLI session is closed"); + expect(spawned?.signalCode).toBe("SIGTERM"); +}); + +test("reuses one process and restarts it when the environment changes", async () => { + let spawns = 0; + const session = createCliCommandSession({ + spawnWorker: (environment) => { + spawns += 1; + const child = fakeProcess(50 + spawns); + const reader = createCliSessionFrameReader((value) => { + const request = requestSchema.parse(value); + child.stdout.write( + encodeCliSessionFrame({ + id: request.id, + exitCode: 0, + stdout: request.arguments.join(" "), + stderr: environment.MARKER ?? "", + }), + ); + }); + child.stdout.write(encodeCliSessionFrame({ ready: true })); + child.stdin.on("data", (chunk) => { + reader(chunk); + }); + return asCliProcess(child); + }, + }); + const first = await session.run( + command(["one"], { PATH: "a", MARKER: "first" }), + ); + const second = await session.run( + command(["two"], { PATH: "a", MARKER: "first" }), + ); + expect(spawns).toBe(1); + expect(session.processId()).toBe(51); + expect(first.stdout).toBe("one"); + expect(second.stdout).toBe("two"); + expect(first.stderr).toBe("first"); + expect(second.stderr).toBe("first"); + + const changed = await session.run( + command(["three"], { PATH: "b", MARKER: "second" }), + ); + expect(spawns).toBe(2); + expect(session.processId()).toBe(52); + expect(changed.stdout).toBe("three"); + expect(changed.stderr).toBe("second"); + await session.close(); + expect(session.processId()).toBeUndefined(); +}); + +const reply = ( + child: FakeCliProcess, + id: number, + fields: { + exitCode?: number; + stdout?: string; + stderr?: string; + rss?: number; + fatal?: boolean; + }, +): void => { + child.stdout.write( + encodeCliSessionFrame({ + id, + exitCode: fields.exitCode ?? 0, + stdout: fields.stdout ?? "", + stderr: fields.stderr ?? "", + ...(fields.rss === undefined ? {} : { rss: fields.rss }), + ...(fields.fatal === undefined ? {} : { fatal: fields.fatal }), + }), + ); +}; + +const respondingProcess = ( + pid: number, + onRequest: ( + request: { id: number; arguments: string[] }, + child: FakeCliProcess, + ) => void, +): FakeCliProcess => { + const child = fakeProcess(pid); + child.stdout.write(encodeCliSessionFrame({ ready: true })); + const reader = createCliSessionFrameReader((value) => { + onRequest(requestSchema.parse(value), child); + }); + child.stdin.on("data", (chunk: Buffer) => { + reader(chunk); + }); + return child; +}; + +test("starts another process after the shared CLI process exits", async () => { + const children: FakeCliProcess[] = []; + const session = createCliCommandSession({ + spawnWorker: () => { + const child = fakeProcess(70 + children.length); + children.push(child); + child.stdout.write(encodeCliSessionFrame({ ready: true })); + if (children.length === 1) { + child.stdin.on("data", () => { + child.kill("SIGTERM"); + }); + } else { + const reader = createCliSessionFrameReader((value) => { + const request = requestSchema.parse(value); + child.stdout.write( + encodeCliSessionFrame({ + id: request.id, + exitCode: 0, + stdout: "recovered", + stderr: "", + }), + ); + }); + child.stdin.on("data", (chunk) => { + reader(chunk); + }); + } + return asCliProcess(child); + }, + }); + await expect( + session.run(command(["fail"], { PATH: "/usr/bin" })), + ).rejects.toThrow(/Shared CLI process exited/); + await expect( + session.run(command(["again"], { PATH: "/usr/bin" })), + ).resolves.toMatchObject({ + exitCode: 0, + stdout: "recovered", + }); + expect(children).toHaveLength(2); + await session.close(); +}); + +test("retires a fatal worker before starting the next command", async () => { + const events: string[] = []; + const session = createCliCommandSession({ + spawnWorker: () => { + const generation = events.filter((event) => + event.startsWith("spawn"), + ).length; + events.push(`spawn:${generation}`); + const child = respondingProcess(80 + generation, (request, current) => { + events.push(`command:${generation}`); + if (generation === 0) { + reply(current, request.id, { + exitCode: 1, + stderr: "boom", + fatal: true, + }); + return; + } + reply(current, request.id, { stdout: "next" }); + }); + const stop = child.kill.bind(child); + child.kill = (signal?: NodeJS.Signals) => { + events.push(`kill:${generation}`); + return stop(signal); + }; + return asCliProcess(child); + }, + }); + const fatal = session.run(command(["fail"], { PATH: "/usr/bin" })); + const next = session.run(command(["next"], { PATH: "/usr/bin" })); + await expect(fatal).resolves.toMatchObject({ exitCode: 1, stderr: "boom" }); + await expect(next).resolves.toMatchObject({ exitCode: 0, stdout: "next" }); + expect(events).toEqual([ + "spawn:0", + "command:0", + "kill:0", + "spawn:1", + "command:1", + ]); + await session.close(); +}); + +test("retires a transport failure before the next command reuses the worker", async () => { + const events: string[] = []; + const session = createCliCommandSession({ + spawnWorker: () => { + const generation = events.filter((event) => + event.startsWith("spawn"), + ).length; + events.push(`spawn:${generation}`); + const child = respondingProcess(120 + generation, (request, current) => { + events.push(`command:${generation}`); + if (generation === 0) { + current.stdout.write(encodeCliSessionFrame({ id: request.id })); + return; + } + reply(current, request.id, { stdout: "recovered" }); + }); + const stop = child.kill.bind(child); + child.kill = (signal: NodeJS.Signals = "SIGTERM") => { + events.push(`kill:${generation}`); + // Leave signalCode unset until a later turn. A synchronous SIGTERM + // does not update it, so reuse cannot depend on that field. + queueMicrotask(() => { + if (child.exitCode !== null || child.signalCode !== null) return; + stop(signal); + }); + return true; + }; + return asCliProcess(child); + }, + }); + const failed = session.run(command(["bad"], { PATH: "/usr/bin" })); + const next = session.run(command(["next"], { PATH: "/usr/bin" })); + await expect(failed).rejects.toThrow(/exit code/); + await expect(next).resolves.toMatchObject({ stdout: "recovered" }); + expect(events).toEqual([ + "spawn:0", + "command:0", + "kill:0", + "spawn:1", + "command:1", + ]); + await session.close(); +}); + +test("restarts the shared CLI process after the command limit", async () => { + const children: FakeCliProcess[] = []; + const session = createCliCommandSession({ + commandLimit: 2, + spawnWorker: () => { + const child = respondingProcess( + 90 + children.length, + (request, current) => { + reply(current, request.id, { + stdout: String(children.indexOf(current)), + }); + }, + ); + children.push(child); + return asCliProcess(child); + }, + }); + const first = await session.run(command(["one"], { PATH: "/usr/bin" })); + const second = await session.run(command(["two"], { PATH: "/usr/bin" })); + expect(children).toHaveLength(1); + expect(first.stdout).toBe("0"); + expect(second.stdout).toBe("0"); + const third = await session.run(command(["three"], { PATH: "/usr/bin" })); + expect(children).toHaveLength(2); + expect(children[0]?.signalCode).toBe("SIGTERM"); + expect(third.stdout).toBe("1"); + expect(session.processId()).toBe(91); + await session.close(); +}); + +test("restarts the shared CLI process after the resident-set limit", async () => { + const children: FakeCliProcess[] = []; + const session = createCliCommandSession({ + commandLimit: 100, + residentLimitBytes: 100, + spawnWorker: () => { + const generation = children.length; + const child = respondingProcess(100 + generation, (request, current) => { + reply(current, request.id, { + stdout: String(generation), + rss: generation === 0 ? 150 : 10, + }); + }); + children.push(child); + return asCliProcess(child); + }, + }); + await expect( + session.run(command(["one"], { PATH: "/usr/bin" })), + ).resolves.toMatchObject({ stdout: "0" }); + expect(session.processId()).toBe(100); + await expect( + session.run(command(["two"], { PATH: "/usr/bin" })), + ).resolves.toMatchObject({ stdout: "1" }); + expect(children[0]?.signalCode).toBe("SIGTERM"); + expect(session.processId()).toBe(101); + await session.close(); +}); + +test("starts a new process after the real worker reports a fatal failure", async () => { + const session = createCliCommandSession(); + const failed = await session.run({ + arguments: ["--version"], + cwd: "/workspace/this-directory-does-not-exist", + environment: { PATH: process.env.PATH ?? "", REA_LOG_LEVEL: "silent" }, + }); + expect(failed.exitCode).toBe(1); + expect(failed.stderr).toMatch(/ENOENT/); + const recovered = await session.run({ + arguments: ["--version"], + cwd: process.cwd(), + environment: { PATH: process.env.PATH ?? "", REA_LOG_LEVEL: "silent" }, + }); + expect(recovered.exitCode).toBe(0); + expect(recovered.stdout.length).toBeGreaterThan(0); + expect(session.processId()).toEqual(expect.any(Number)); + await session.close(); +}); + +const requireSharedCli = (cli: { + sharedCliRun?: (invocation: CliInvocation) => Promise; +}): ((invocation: CliInvocation) => Promise) => { + const run = cli.sharedCliRun; + if (run === undefined) + throw new Error("cliTest did not provide a shared CLI session"); + return run; +}; + +const normalized = (result: CliResult) => + z + .object({ normalized_result: javascriptApplicationAnalysisResultSchema }) + .parse(result.json).normalized_result; + +const analyzeArguments = (root: string): readonly string[] => [ + "analyze-javascript-application", + root, + "--format", + "json", +]; + +const writeApplication = async (prefix: string): Promise => { + const root = await createTestTempDirectory(prefix); + await writeFixtureFiles(root, { + "package.json": JSON.stringify({ main: "main.cjs" }), + "main.cjs": + 'const { BrowserWindow } = require("electron"); new BrowserWindow().loadFile("index.html");', + "index.html": "page", + }); + return root; +}; + +cliTest( + "matches a fresh CLI process and reuses that process for the same environment", + async ({ cli, cliCommandSession }) => { + const sharedCliRun = requireSharedCli(cli); + const fresh = await cli.run({ + arguments: ["--version"], + environment: { REA_LOG_LEVEL: "silent" }, + }); + const shared = await sharedCliRun({ + arguments: ["--version"], + environment: { REA_LOG_LEVEL: "silent" }, + }); + expect(shared.exitCode).toBe(fresh.exitCode); + expect(shared.stdout).toBe(fresh.stdout); + expect(shared.stderr).toBe(fresh.stderr); + const pid = cliCommandSession.processId(); + expect(pid).toEqual(expect.any(Number)); + const again = await sharedCliRun({ + arguments: ["--version"], + environment: { REA_LOG_LEVEL: "silent" }, + }); + expect(again.stdout).toBe(fresh.stdout); + expect(cliCommandSession.processId()).toBe(pid); + + const freshInvalid = await cli.run({ + arguments: ["--version"], + environment: { REA_LOG_LEVEL: "not-a-level" }, + }); + const sharedInvalid = await sharedCliRun({ + arguments: ["--version"], + environment: { REA_LOG_LEVEL: "not-a-level" }, + }); + expect(sharedInvalid.exitCode).toBe(freshInvalid.exitCode); + expect(sharedInvalid.exitCode).not.toBe(0); + expect(sharedInvalid.stdout).toBe(freshInvalid.stdout); + expect(sharedInvalid.stderr).toBe(freshInvalid.stderr); + expect(cliCommandSession.processId()).not.toBe(pid); + + const restored = await sharedCliRun({ + arguments: ["--version"], + environment: { REA_LOG_LEVEL: "silent" }, + }); + expect(restored.exitCode).toBe(0); + expect(restored.stdout).toBe(fresh.stdout); + expect(cliCommandSession.processId()).not.toBe(pid); + }, +); + +cliTest( + "reuses one CLI process across JavaScript application directories", + async ({ cli, cliCommandSession }) => { + const sharedCliRun = requireSharedCli(cli); + const firstRoot = await writeApplication("rea-shared-cli-"); + const fresh = await cli.run({ + arguments: analyzeArguments(firstRoot), + environment: { REA_LOG_LEVEL: "silent" }, + }); + const shared = await sharedCliRun({ + arguments: analyzeArguments(firstRoot), + environment: { REA_LOG_LEVEL: "silent" }, + }); + expect(shared.exitCode).toBe(0); + expect(normalized(shared)).toEqual(normalized(fresh)); + const pid = cliCommandSession.processId(); + expect(pid).toEqual(expect.any(Number)); + + const secondRoot = await writeApplication("rea-shared-cli-next-"); + const freshNext = await cli.run({ + arguments: analyzeArguments(secondRoot), + environment: { REA_LOG_LEVEL: "silent" }, + }); + const sharedNext = await sharedCliRun({ + arguments: analyzeArguments(secondRoot), + environment: { REA_LOG_LEVEL: "silent" }, + }); + expect(normalized(sharedNext)).toEqual(normalized(freshNext)); + expect(normalized(sharedNext).input_path).not.toBe( + normalized(shared).input_path, + ); + expect(cliCommandSession.processId()).toBe(pid); + + const analyzed = await analyzeJavaScriptDirectory("CLI", cli, secondRoot); + expect(analyzed.input_path).toBe(normalized(freshNext).input_path); + expect(cliCommandSession.processId()).toBe(pid); + }, +); diff --git a/tests/support/cli/cliCommandSession.ts b/tests/support/cli/cliCommandSession.ts new file mode 100644 index 0000000000..1805836054 --- /dev/null +++ b/tests/support/cli/cliCommandSession.ts @@ -0,0 +1,448 @@ +import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process"; +import { fileURLToPath } from "node:url"; + +import { z } from "zod"; + +import { waitForExit } from "../process/processFixture.js"; +import { + createCliSessionFrameReader, + encodeCliSessionFrame, +} from "./cliCommandSessionFrames.mjs"; + +const workerPath = fileURLToPath( + new URL("./cliCommandSessionWorker.mjs", import.meta.url), +); +const SHUTDOWN_GRACE_MS = 1_000; +/** Restart after this many commands so one file cannot retain every heap. */ +const COMMAND_REUSE_LIMIT = 32; +/** Restart when the worker reports a resident set at least this large. */ +const RESIDENT_REUSE_LIMIT_BYTES = 1024 * 1024 * 1024; + +const responseSchema = z.object({ + ready: z.boolean().optional(), + id: z.number().optional(), + exitCode: z.number().nullable().optional(), + stdout: z.string().optional(), + stderr: z.string().optional(), + pid: z.number().optional(), + rss: z.number().finite().nonnegative().optional(), + fatal: z.boolean().optional(), +}); + +/** One compiled CLI command sent to the shared process. */ +export interface CliCommandRequest { + readonly arguments: readonly string[]; + readonly cwd: string; + readonly environment: Readonly>; +} + +/** Captured result from one shared CLI command. */ +export interface CliCommandResult { + readonly exitCode: number | null; + readonly signal: NodeJS.Signals | null; + readonly stdout: string; + readonly stderr: string; + readonly json: unknown | undefined; +} + +/** File-scoped compiled CLI process. */ +export interface CliCommandSession { + run(request: CliCommandRequest): Promise; + /** PID of the live process, when one is running. */ + processId(): number | undefined; + close(): Promise; +} + +interface CliCommandSessionDependencies { + readonly spawnWorker?: ( + environment: Readonly>, + ) => ChildProcessWithoutNullStreams; + /** Commands served by one process before the next command starts a new one. */ + readonly commandLimit?: number; + /** Resident-set bytes, reported by the worker, that start a new process. */ + readonly residentLimitBytes?: number; +} + +interface PendingCommand { + readonly id: number; + readonly resolve: (result: CliCommandResult) => void; + readonly reject: (error: unknown) => void; +} + +type SessionResponse = z.infer; + +const closedError = (): Error => new Error("Shared CLI session is closed"); + +const positiveLimit = (value: number | undefined, fallback: number): number => + value !== undefined && Number.isSafeInteger(value) && value > 0 + ? value + : fallback; + +const envKey = (environment: Readonly>): string => + JSON.stringify( + Object.keys(environment) + .sort() + .map((key) => [key, environment[key]]), + ); + +const parseCliStdout = (stdout: string): unknown | undefined => { + const document = stdout.trim(); + if (document.length === 0) return undefined; + try { + return JSON.parse(document) as unknown; + } catch { + return undefined; + } +}; + +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 stopChild = async ( + child: ChildProcessWithoutNullStreams, +): Promise => { + if (child.exitCode !== null || child.signalCode !== null) return; + signalChild(child, "SIGTERM"); + if (await waitForExit(child, SHUTDOWN_GRACE_MS)) return; + if (child.exitCode !== null || child.signalCode !== null) return; + signalChild(child, "SIGKILL"); + if (await waitForExit(child, SHUTDOWN_GRACE_MS)) return; + throw new Error( + `shared CLI process ${String(child.pid)} did not exit after SIGKILL`, + ); +}; + +const commandResult = (message: SessionResponse): CliCommandResult => ({ + exitCode: message.exitCode ?? null, + signal: null, + stdout: message.stdout ?? "", + stderr: message.stderr ?? "", + json: parseCliStdout(message.stdout ?? ""), +}); + +const writeChild = ( + child: ChildProcessWithoutNullStreams, + frame: Buffer, +): Promise => + new Promise((resolve, reject) => { + child.stdin.write(frame, (error) => { + if (error) reject(error); + else resolve(); + }); + }); + +const defaultSpawnWorker = ( + environment: Readonly>, +): ChildProcessWithoutNullStreams => + spawn(process.execPath, [workerPath], { + cwd: process.cwd(), + env: environment, + stdio: ["pipe", "pipe", "pipe"], + }); + +class SharedCliProcess implements CliCommandSession { + readonly #spawnWorker: ( + environment: Readonly>, + ) => ChildProcessWithoutNullStreams; + + readonly #commandLimit: number; + readonly #residentLimitBytes: number; + #child: ChildProcessWithoutNullStreams | undefined; + #childEnvKey: string | undefined; + #commands = 0; + #residentBytes: number | undefined; + #retiring: Promise | undefined; + #closed = false; + #closing: Promise | undefined; + #starting: Promise | undefined; + #nextId = 1; + #pending: PendingCommand | undefined; + #resolveReady: (() => void) | undefined; + #rejectReady: ((error: unknown) => void) | undefined; + #queue: Promise = Promise.resolve(); + + constructor(dependencies: CliCommandSessionDependencies) { + this.#spawnWorker = dependencies.spawnWorker ?? defaultSpawnWorker; + this.#commandLimit = positiveLimit( + dependencies.commandLimit, + COMMAND_REUSE_LIMIT, + ); + this.#residentLimitBytes = positiveLimit( + dependencies.residentLimitBytes, + RESIDENT_REUSE_LIMIT_BYTES, + ); + } + + run(request: CliCommandRequest): Promise { + const result = this.#queue.then(() => this.#invoke(request)); + this.#queue = result.then( + () => undefined, + () => undefined, + ); + return result; + } + + processId(): number | undefined { + const child = this.#child; + if ( + child === undefined || + child.exitCode !== null || + child.signalCode !== null + ) { + return undefined; + } + return child.pid; + } + + close(): Promise { + this.#closed = true; + if (this.#closing !== undefined) return this.#closing; + const attempt = this.#shutdown(); + 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; + 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(); + } + } + + 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 }; + }); + try { + await writeChild( + owner, + encodeCliSessionFrame({ + id, + arguments: request.arguments, + cwd: request.cwd, + env: request.environment, + }), + ); + } catch (error) { + this.#failTransport(owner, error); + } + return await result; + } + + async #finishRetirement(): Promise { + const retiring = this.#retiring; + if (retiring === undefined) return; + try { + await retiring; + } finally { + if (this.#retiring === retiring) this.#retiring = undefined; + } + } + + #reusable( + current: ChildProcessWithoutNullStreams | undefined, + key: string, + ): current is ChildProcessWithoutNullStreams { + if ( + current === undefined || + current.exitCode !== null || + current.signalCode !== null || + this.#childEnvKey !== key + ) { + return false; + } + if (this.#commands >= this.#commandLimit) return false; + return !( + this.#residentBytes !== undefined && + this.#residentBytes >= this.#residentLimitBytes + ); + } + + 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); + this.#child = started; + this.#childEnvKey = envKey(environment); + this.#commands = 0; + this.#residentBytes = undefined; + const ready = new Promise((resolve, reject) => { + this.#resolveReady = resolve; + this.#rejectReady = reject; + }); + this.#starting = ready; + this.#watch(started); + if (this.#closed) { + signalChild(started, "SIGTERM"); + this.#failReady(closedError()); + } + return ready.finally(() => { + if (this.#starting === ready) this.#starting = undefined; + }); + } + + #watch(started: ChildProcessWithoutNullStreams): void { + const reader = createCliSessionFrameReader((frame) => { + this.#onFrame(started, frame); + }); + started.stdout.on("data", (chunk: Buffer) => { + try { + reader(chunk); + } catch (error) { + this.#failTransport(started, error); + } + }); + started.stderr.resume(); + started.stdin.on("error", (error) => { + this.#failTransport(started, error); + }); + started.once("error", (error) => { + this.#failTransport(started, error); + }); + started.once("close", (code, signal) => { + this.#onClose(started, code, signal); + }); + } + + #onFrame(owner: ChildProcessWithoutNullStreams, value: unknown): void { + if (this.#child !== owner || this.#closed) return; + const parsed = responseSchema.safeParse(value); + if (!parsed.success) { + this.#failTransport( + owner, + new Error("CLI session response was not a command result"), + ); + return; + } + const message = parsed.data; + if (message.ready === true && message.id === undefined) { + const resolve = this.#resolveReady; + this.#resolveReady = undefined; + this.#rejectReady = undefined; + resolve?.(); + return; + } + const current = this.#pending; + if (current === undefined || message.id !== current.id) { + this.#failTransport( + owner, + new Error("CLI session response did not match the command"), + ); + return; + } + if (message.exitCode === undefined) { + this.#failTransport( + owner, + new Error("CLI session response omitted its exit code"), + ); + return; + } + this.#pending = undefined; + if (message.fatal === true) { + this.#retire(owner); + } else { + this.#commands += 1; + if (message.rss !== undefined) this.#residentBytes = message.rss; + } + current.resolve(commandResult(message)); + } + + #retire(owner: ChildProcessWithoutNullStreams): void { + if (this.#child === owner) { + this.#child = undefined; + this.#childEnvKey = undefined; + } + this.#commands = 0; + this.#residentBytes = undefined; + this.#retiring = stopChild(owner); + } + + #onClose( + owner: ChildProcessWithoutNullStreams, + code: number | null, + signal: NodeJS.Signals | null, + ): void { + if (this.#child !== owner) return; + this.#child = undefined; + this.#childEnvKey = undefined; + const error = this.#closed + ? closedError() + : new Error( + `Shared CLI process exited (${String(code ?? signal ?? "unknown")})`, + ); + this.#failReady(error); + this.#failPending(error); + } + + #failTransport(owner: ChildProcessWithoutNullStreams, error: unknown): void { + if (this.#child !== owner) return; + this.#failReady(error); + this.#failPending(error); + // Clear the worker before the queued command runs. SIGTERM does not set + // signalCode until the process actually exits, so #reusable would still + // accept this process and write into it. + this.#retire(owner); + } + + #failPending(error: unknown): void { + const current = this.#pending; + this.#pending = undefined; + current?.reject(error); + } + + #failReady(error: unknown): void { + const reject = this.#rejectReady; + this.#resolveReady = undefined; + this.#rejectReady = undefined; + reject?.(error); + } +} + +/** + * Keep one compiled CLI process for commands that share an environment. + * The process restarts after its command limit, after a reported resident-set + * limit, when the environment changes, or when a command fails fatally. + */ +export const createCliCommandSession = ( + dependencies: CliCommandSessionDependencies = {}, +): CliCommandSession => new SharedCliProcess(dependencies); diff --git a/tests/support/cli/cliCommandSessionFrames.d.mts b/tests/support/cli/cliCommandSessionFrames.d.mts new file mode 100644 index 0000000000..85e6ae65f7 --- /dev/null +++ b/tests/support/cli/cliCommandSessionFrames.d.mts @@ -0,0 +1,13 @@ +/** Largest accepted session message, including the encoded command result. */ +export const MAX_CLI_SESSION_FRAME_BYTES: number; + +/** Length-prefixed JSON frame shared by the session parent and worker. */ +export function encodeCliSessionFrame(value: unknown): Buffer; + +/** + * Parse complete frames from a byte stream. An oversized or malformed frame + * fails the reader instead of resynchronizing on a later length. + */ +export function createCliSessionFrameReader( + onFrame: (value: unknown) => void, +): (chunk: Buffer | string) => void; diff --git a/tests/support/cli/cliCommandSessionFrames.mjs b/tests/support/cli/cliCommandSessionFrames.mjs new file mode 100644 index 0000000000..4f88c70cdd --- /dev/null +++ b/tests/support/cli/cliCommandSessionFrames.mjs @@ -0,0 +1,77 @@ +/** Largest accepted session message, including the encoded command result. */ +export const MAX_CLI_SESSION_FRAME_BYTES = 64 * 1024 * 1024; + +/** Length-prefixed JSON frame shared by the session parent and worker. */ +export const encodeCliSessionFrame = (value) => { + const body = Buffer.from(JSON.stringify(value)); + if (body.length > MAX_CLI_SESSION_FRAME_BYTES) { + throw new Error( + `CLI session frame is ${body.length} bytes; the limit is ${MAX_CLI_SESSION_FRAME_BYTES}`, + ); + } + const frame = Buffer.allocUnsafe(4 + body.length); + frame.writeUInt32BE(body.length, 0); + body.copy(frame, 4); + return frame; +}; + +const frameLimitError = (length) => + new Error( + `CLI session frame is ${length} bytes; the limit is ${MAX_CLI_SESSION_FRAME_BYTES}`, + ); + +/** Read a big-endian length from the start of a chunk list without joining it. */ +const readFrameLength = (chunks) => { + const first = chunks[0]; + if (first.length >= 4) return first.readUInt32BE(0); + const header = Buffer.allocUnsafe(4); + let offset = 0; + for (const chunk of chunks) { + const take = Math.min(chunk.length, 4 - offset); + chunk.copy(header, offset, 0, take); + offset += take; + if (offset === 4) break; + } + return header.readUInt32BE(0); +}; + +/** + * Parse complete frames from a byte stream. An oversized or malformed frame + * fails the reader instead of resynchronizing on a later length. Incomplete + * chunks stay separate until one whole frame is buffered, then that frame is + * joined once. + */ +export const createCliSessionFrameReader = (onFrame) => { + /** @type {Buffer[]} */ + const chunks = []; + let buffered = 0; + return (chunk) => { + const bytes = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + chunks.push(bytes); + buffered += bytes.length; + if (buffered < 4) return; + const length = readFrameLength(chunks); + if (length > MAX_CLI_SESSION_FRAME_BYTES) throw frameLimitError(length); + if (buffered < 4 + length) return; + const pending = Buffer.concat(chunks); + let offset = 0; + try { + while (offset + 4 <= pending.length) { + const frameLength = pending.readUInt32BE(offset); + if (frameLength > MAX_CLI_SESSION_FRAME_BYTES) { + throw frameLimitError(frameLength); + } + const frameEnd = offset + 4 + frameLength; + if (frameEnd > pending.length) break; + const body = pending.toString("utf8", offset + 4, frameEnd); + offset = frameEnd; + onFrame(JSON.parse(body)); + } + } finally { + const rest = pending.subarray(offset); + chunks.length = 0; + buffered = rest.length; + if (rest.length > 0) chunks.push(Buffer.from(rest)); + } + }; +}; diff --git a/tests/support/cli/cliCommandSessionWorker.mjs b/tests/support/cli/cliCommandSessionWorker.mjs new file mode 100644 index 0000000000..63ce69542c --- /dev/null +++ b/tests/support/cli/cliCommandSessionWorker.mjs @@ -0,0 +1,263 @@ +import { Writable } from "node:stream"; +import { access } from "node:fs/promises"; +import { dirname, join, resolve } from "node:path"; +import { fileURLToPath } from "node:url"; + +import { + createCliSessionFrameReader, + encodeCliSessionFrame, +} from "./cliCommandSessionFrames.mjs"; + +// One-shot command sequence matches scripts/rea.mjs. Output is captured per +// command so this process can serve the next invocation without restarting. +const packageRoot = resolve( + dirname(fileURLToPath(import.meta.url)), + "../../..", +); +const { default: packageJson } = await import( + new URL("../../../package.json", import.meta.url), + { with: { type: "json" } } +); + +const prepareCliProcess = () => { + process.env.REA_PACKAGE_VERSION = packageJson.version; + // Windows home lookup throws for an empty USERPROFILE. Dependencies resolve + // home during import, before REA's own resolver can apply its account fallback. + if (process.platform === "win32" && process.env.USERPROFILE === "") { + delete process.env.USERPROFILE; + } + if ( + process.platform === "android" && + process.env.PLAYWRIGHT_BROWSERS_PATH === undefined + ) { + const cacheRoot = + process.env.XDG_CACHE_HOME ?? join(process.env.HOME ?? ".", ".cache"); + process.env.PLAYWRIGHT_BROWSERS_PATH = join(cacheRoot, "ms-playwright"); + } +}; + +const runtimeFiles = [ + "dist/cli-runtime/cli.js", + "dist/cli-runtime/cliOutput.js", + "dist/cli-runtime/cliResultOutput.js", +]; + +prepareCliProcess(); +if (!(await compiledRuntimeExists(runtimeFiles))) { + process.stderr.write( + `REA's compiled runtime is missing. Run \`npm ci && npm run build:cached\` in ${packageRoot} to install dependencies and build REA, then restart it. If this is an installed package, reinstall rea-agents.\n`, + ); + process.exit(1); +} + +const { cliConfigurationError, createCli } = + await import("../../../dist/cli-runtime/cli.js"); +const { createCliResultOutput } = + await import("../../../dist/cli-runtime/cliResultOutput.js"); +const { + normalizeCliArguments, + renderCliOutputArgumentError, + renderEmptyFilteredCliOutput, + validateCliOutputArguments, +} = await import("../../../dist/cli-runtime/cliOutput.js"); + +const writeFrame = (value) => + new Promise((resolveWrite, rejectWrite) => { + process.stdout.write(encodeCliSessionFrame(value), (error) => { + if (error) rejectWrite(error); + else resolveWrite(); + }); + }); + +await writeFrame({ ready: true }); + +let chain = Promise.resolve(); +const reader = createCliSessionFrameReader((value) => { + chain = chain.then( + () => handle(value), + () => handle(value), + ); +}); +process.stdin.on("data", (chunk) => { + try { + reader(chunk); + } catch (error) { + process.stderr.write( + `${error instanceof Error ? error.message : String(error)}\n`, + ); + process.exit(1); + } +}); +process.stdin.on("end", () => { + chain.finally(() => process.exit(0)); +}); + +async function handle(value) { + const request = value; + const savedEnv = { ...process.env }; + const savedCwd = process.cwd(); + try { + if ( + request === null || + typeof request !== "object" || + !Array.isArray(request.arguments) || + typeof request.cwd !== "string" || + request.env === null || + typeof request.env !== "object" + ) { + throw new Error( + "CLI session command is missing its arguments, cwd, or environment", + ); + } + for (const key of Object.keys(process.env)) delete process.env[key]; + Object.assign(process.env, request.env); + prepareCliProcess(); + process.chdir(request.cwd); + const result = await runOneShot(request.arguments); + await writeFrame({ + id: request.id, + pid: process.pid, + rss: process.memoryUsage().rss, + ...result, + }); + } catch (error) { + const id = + request !== null && typeof request === "object" ? request.id : undefined; + await writeFrame({ + id, + pid: process.pid, + exitCode: 1, + stdout: "", + stderr: + error instanceof Error ? (error.stack ?? error.message) : String(error), + fatal: true, + }); + process.exit(1); + } finally { + for (const key of Object.keys(process.env)) delete process.env[key]; + Object.assign(process.env, savedEnv); + process.chdir(savedCwd); + } +} + +async function runOneShot(args) { + const stdoutChunks = []; + const stderrChunks = []; + const originalStdoutWrite = process.stdout.write.bind(process.stdout); + const originalStderrWrite = process.stderr.write.bind(process.stderr); + process.stdout.write = captureWrite(stdoutChunks); + process.stderr.write = captureWrite(stderrChunks); + const destination = new Writable({ + write(chunk, encoding, callback) { + process.stdout.write(chunk, encoding, callback); + }, + }); + destination.isTTY = process.stdout.isTTY === true; + const previousExitCode = process.exitCode; + process.exitCode = undefined; + let explicitExit; + try { + const cliArgs = normalizeCliArguments(args); + const outputArguments = validateCliOutputArguments(cliArgs); + const configurationError = cliConfigurationError(process.env); + if (configurationError !== undefined && !isDoctorCommand(cliArgs)) { + process.stderr.write(`${configurationError}\n`); + process.exitCode = 1; + } else if (!outputArguments.ok) { + process.stdout.write(renderCliOutputArgumentError(outputArguments)); + process.exitCode = 1; + } else { + let wroteOutput = false; + const resultOutput = createCliResultOutput(cliArgs, destination); + try { + await createCli(process.env, resultOutput).serve(cliArgs, { + stdout: (output) => { + if (resultOutput?.handled) { + if (resultOutput.failed) process.stderr.write(output); + return; + } + if (output.length > 0) wroteOutput = true; + process.stdout.write(output); + }, + exit: (code) => { + explicitExit = code; + }, + }); + } catch (cause) { + if (wroteOutput) throw cause; + const recovered = await resultOutput?.recoverFormattingFailure(cause); + if (!recovered) throw cause; + process.exitCode = 1; + } + if (!wroteOutput && !resultOutput?.handled) { + const empty = renderEmptyFilteredCliOutput(cliArgs); + if (empty) process.stdout.write(empty); + } + } + } finally { + process.stdout.write = originalStdoutWrite; + process.stderr.write = originalStderrWrite; + } + const exitCode = + explicitExit ?? + (typeof process.exitCode === "number" ? process.exitCode : 0); + process.exitCode = previousExitCode; + return { + exitCode, + stdout: Buffer.concat(stdoutChunks).toString("utf8"), + stderr: Buffer.concat(stderrChunks).toString("utf8"), + }; +} + +function captureWrite(chunks) { + return function patched(chunk, encoding, callback) { + const encodingName = typeof encoding === "string" ? encoding : "utf8"; + chunks.push( + Buffer.isBuffer(chunk) ? chunk : Buffer.from(String(chunk), encodingName), + ); + const done = typeof encoding === "function" ? encoding : callback; + if (typeof done === "function") done(); + return true; + }; +} + +function isDoctorCommand(arguments_) { + const valuedFlags = new Set([ + "--format", + "--filter-output", + "--token-limit", + "--token-offset", + ]); + for (let index = 0; index < arguments_.length; index += 1) { + const argument = arguments_[index]; + if (argument === "--") return arguments_[index + 1] === "doctor"; + if (valuedFlags.has(argument)) { + index += 1; + continue; + } + if (argument.startsWith("-")) continue; + return argument === "doctor"; + } + return false; +} + +async function compiledRuntimeExists(paths) { + for (const path of paths) { + try { + await access(resolve(packageRoot, path)); + } catch (cause) { + if (isMissing(cause)) return false; + throw cause; + } + } + return true; +} + +function isMissing(cause) { + return ( + typeof cause === "object" && + cause !== null && + "code" in cause && + cause.code === "ENOENT" + ); +} diff --git a/tests/support/cli/cliFixture.ts b/tests/support/cli/cliFixture.ts index 9d093c7da4..db71ac40f0 100644 --- a/tests/support/cli/cliFixture.ts +++ b/tests/support/cli/cliFixture.ts @@ -10,6 +10,10 @@ import { type ProcessResult, type TestProcesses, } from "../process/processFixture.js"; +import { + createCliCommandSession, + type CliCommandSession, +} from "./cliCommandSession.js"; /** Caller-controlled values for one compiled REA CLI invocation. */ export interface CliInvocation { @@ -40,6 +44,12 @@ export interface TestCli { * drivers created with `createTestCli` alone. */ sharedMcpCall?(): Promise; + /** + * File-scoped compiled CLI process for commands with no stdin. Present on + * `cliTest` drivers. A different environment restarts the process. Stdin and + * explicit deadlines stay on `run`, which starts a fresh process. + */ + sharedCliRun?(invocation: CliInvocation): Promise; } /** One file-scoped stdio MCP session. */ @@ -202,6 +212,34 @@ const parseJson = (stdout: string): unknown | undefined => { } }; +/** Child environment for one compiled CLI process, without the parent environment. */ +export const cliProcessEnvironment = ( + environment: NodeJS.ProcessEnv | undefined, + compileCache?: string, +): NodeJS.ProcessEnv => ({ + PATH: process.env.PATH ?? "", + // Native Windows APIs need the system root and a writable temporary + // directory even when the command's application environment is empty. + ...(process.platform === "win32" + ? { + SystemRoot: process.env.SystemRoot, + TEMP: tmpdir(), + TMP: tmpdir(), + } + : {}), + ...cliCompileCacheEnvironment(environment, compileCache), +}); + +/** Keep string entries. A shared process cannot round-trip an absent variable as JSON. */ +export const definedCliEnvironment = ( + environment: NodeJS.ProcessEnv, +): Record => + Object.fromEntries( + Object.entries(environment).flatMap(([key, value]) => + typeof value === "string" ? [[key, value] as const] : [], + ), + ); + /** Reuse CLI compilation without overriding caller-selected cache or coverage. */ export const cliCompileCacheEnvironment = ( environment: NodeJS.ProcessEnv | undefined, @@ -230,19 +268,7 @@ export const createTestCli = ( [resolve("scripts/rea.mjs"), ...arguments_], { cwd: cwd ?? process.cwd(), - env: { - PATH: process.env.PATH ?? "", - // Native Windows APIs need the system root and a writable temporary - // directory even when the command's application environment is empty. - ...(process.platform === "win32" - ? { - SystemRoot: process.env.SystemRoot, - TEMP: tmpdir(), - TMP: tmpdir(), - } - : {}), - ...cliCompileCacheEnvironment(environment, compileCache), - }, + env: cliProcessEnvironment(environment, compileCache), ...(input === undefined ? {} : { input }), ...(timeoutMs === undefined ? {} : { timeoutMs }), }, @@ -254,7 +280,7 @@ export const createTestCli = ( /** Vitest API extended with an owned process registry and compiled CLI driver. */ export const cliTest = processTest.extend<{ $test: { cli: TestCli }; - $file: { mcpSession: SharedMcpSession }; + $file: { mcpSession: SharedMcpSession; cliCommandSession: CliCommandSession }; $worker: { cliCompileCache: string }; }>({ cliCompileCache: [ @@ -280,11 +306,40 @@ export const cliTest = processTest.extend<{ }, { scope: "file" }, ], - cli: async ({ processes, cliCompileCache, mcpSession }, use) => { + cliCommandSession: [ + // oxlint-disable-next-line no-empty-pattern -- Vitest parses fixture dependencies from object destructuring. + async ({}, use) => { + const session = createCliCommandSession(); + try { + await use(session); + } finally { + await session.close(); + } + }, + { scope: "file" }, + ], + cli: async ( + { processes, cliCompileCache, mcpSession, cliCommandSession }, + use, + ) => { const cli = createTestCli(processes, cliCompileCache); await use({ ...cli, sharedMcpCall: () => mcpSession.call(), + sharedCliRun: (invocation) => { + if ( + invocation.input !== undefined || + invocation.timeoutMs !== undefined + ) + return cli.run(invocation); + return cliCommandSession.run({ + arguments: invocation.arguments, + cwd: invocation.cwd ?? process.cwd(), + environment: definedCliEnvironment( + cliProcessEnvironment(invocation.environment, cliCompileCache), + ), + }); + }, }); }, }); diff --git a/tests/support/cli/javascriptApplication.ts b/tests/support/cli/javascriptApplication.ts index 6a04b9592c..186953cb91 100644 --- a/tests/support/cli/javascriptApplication.ts +++ b/tests/support/cli/javascriptApplication.ts @@ -35,7 +35,11 @@ export const resolveJavaScriptApplicationMcpCall = async ( return (await connectLocalToolsMcp()).call; }; -/** Analyze a directory; a supplied MCP caller reuses its test-owned connection. */ +/** + * Analyze a directory. A supplied MCP caller reuses its test-owned connection. + * CLI uses the driver's file-scoped process when present, and a fresh process + * otherwise. + */ export const analyzeJavaScriptDirectory = async ( surface: "CLI" | "stdio MCP", cli: TestCli, @@ -50,7 +54,8 @@ export const analyzeJavaScriptDirectory = async ( ) => { let document: unknown; if (surface === "CLI") { - const response = await cli.run({ + const run = cli.sharedCliRun ?? ((invocation) => cli.run(invocation)); + const response = await run({ arguments: [ "analyze-javascript-application", root,