import { Writable } from "node:stream"; import { access } from "node:fs/promises"; import { dirname, resolve } from "node:path"; import { fileURLToPath } from "node:url"; import { prepareCliProcessEnvironment } from "../../../scripts/lib/cli-process-environment.mjs"; import { withVerifierCleanup } from "../../../scripts/lib/verifier-cleanup.mjs"; import { createCliSessionFrameReader, encodeCliSessionFrame, } from "./cliCommandSessionFrames.mjs"; // Commands run through the same one-shot owner as scripts/rea.mjs. Output is // captured per command so this process can serve the next invocation. const packageRoot = resolve( dirname(fileURLToPath(import.meta.url)), "../../..", ); const { default: packageJson } = await import( new URL("../../../package.json", import.meta.url), { with: { type: "json" } } ); const runtimeFiles = [ "dist/cli-runtime/cli.js", "dist/cli-runtime/cliOneShot.js", ]; prepareCliProcessEnvironment( process.env, process.platform, packageJson.version, ); 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 { closeCliResources } = await import("../../../dist/cli-runtime/cli.js"); const { runOneShotCli } = await import("../../../dist/cli-runtime/cliOneShot.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(); // EOF drains admitted commands; failure seals input and skips commands not yet started. /** @type {{ kind: "open" } | { kind: "ended" } | { kind: "failed", cause: unknown }} */ let completion = { kind: "open" }; /** @type {(() => void) | undefined} */ let resolveInput; /** @type {Promise} */ const inputStopped = new Promise((resolve) => { resolveInput = resolve; }); const failInput = (cause) => { if (completion.kind !== "failed") completion = { kind: "failed", cause }; process.stdin.pause(); resolveInput?.(); }; process.stdin.once("end", () => { if (completion.kind === "open") completion = { kind: "ended" }; resolveInput?.(); }); const reader = createCliSessionFrameReader((value) => { if (completion.kind !== "open") return; chain = chain .then(async () => { if (completion.kind !== "failed") await handle(value); }) .catch(failInput); }); process.stdin.on("data", (chunk) => { if (completion.kind !== "open") return; try { reader(chunk); } catch (cause) { failInput(cause); } }); try { await withVerifierCleanup(async () => { await inputStopped; await chain; if (completion.kind === "failed") throw completion.cause; }, [closeCliResources]); } catch (cause) { console.error(cause); process.exit(1); } 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); prepareCliProcessEnvironment( process.env, process.platform, packageJson.version, ); 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) { failInput(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, }); } 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; let exitOutput; try { await runOneShotCli(args, { environment: process.env, runtime: { packageName: packageJson.name, packageVersion: packageJson.version, versions: process.versions, platform: process.platform, architecture: process.arch, }, stdoutIsTerminal: destination.isTTY, resultDestination: destination, // The session frames own this process's stdout; a command's reader // cannot close it. closedStdoutStatus: 1, writeStdout: (text) => { process.stdout.write(text); }, writeStderr: (text) => { process.stderr.write(text); }, stdoutClosed: () => false, exitCode: () => typeof process.exitCode === "number" ? process.exitCode : 0, setExitCode: (code) => { process.exitCode = code; }, // The session closes CLI resources once, after its last command. ownCliResources: () => {}, exitRequested: () => exitOutput !== undefined, // This process serves the next command; record the status and finish // the pending result in place of exiting. exit: (code, finishOutput) => { explicitExit = code; exitOutput = finishOutput(code); }, }); await exitOutput; } 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; }; } 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" ); }