diff --git a/typescript/package.json b/typescript/package.json index dd10ea6b..fd9a1442 100644 --- a/typescript/package.json +++ b/typescript/package.json @@ -21,7 +21,7 @@ }, "scripts": { "build": "tsc -p src", - "test": "npm run build && tsc -p test && node --test out/validate.test.js out/zod.test.js out/program.test.js tests/model.test.mjs", + "test": "npm run build && tsc -p test && node --test out/validate.test.js out/zod.test.js out/program.test.js out/interactive-stdin.test.js tests/model.test.mjs", "build-all": "npm run build --workspaces", "prepare": "npm run build-all", "prepublishOnly": "node -e \"require('fs').copyFileSync('../SECURITY.md','SECURITY.md')\"", diff --git a/typescript/src/interactive/interactive.ts b/typescript/src/interactive/interactive.ts index 16f1c541..6e85ca05 100644 --- a/typescript/src/interactive/interactive.ts +++ b/typescript/src/interactive/interactive.ts @@ -21,14 +21,26 @@ export async function processRequests(interactivePrompt: string, inputFileName: } else { const stdio = readline.createInterface({ input: process.stdin, output: process.stdout }); - while (true) { - const input = await stdio.question(interactivePrompt); + // Iterating the interface buffers lines that arrive while a request is running and ends on EOF. + // Track closing, since EOF can arrive mid-request and prompt() throws on a closed interface. + // Track pausing too: the iterator pauses input when its buffer fills, and prompt() would resume it. + let closed = false; + let paused = false; + stdio.once("close", () => closed = true); + stdio.on("pause", () => paused = true); + stdio.on("resume", () => paused = false); + stdio.setPrompt(interactivePrompt); + stdio.prompt(); + for await (const input of stdio) { if (input.toLowerCase() === "quit" || input.toLowerCase() === "exit") { break; } else if (input.length) { await processRequest(input); } + if (!closed && !paused) { + stdio.prompt(); + } } stdio.close(); } diff --git a/typescript/test/interactive-stdin.test.ts b/typescript/test/interactive-stdin.test.ts new file mode 100644 index 00000000..c7185d55 --- /dev/null +++ b/typescript/test/interactive-stdin.test.ts @@ -0,0 +1,139 @@ +import { describe, it, afterEach } from "node:test"; +import assert from "node:assert/strict"; +import { once } from "node:events"; +import { PassThrough } from "node:stream"; +import { processRequests } from "../dist/interactive/index.js"; + +// processRequests reads requests from process.stdin and writes prompts to process.stdout. Each test +// swaps both for PassThrough streams so it controls exactly when input lines and EOF arrive. +const stdinDescriptor = Object.getOwnPropertyDescriptor(process, "stdin")!; +const stdoutDescriptor = Object.getOwnPropertyDescriptor(process, "stdout")!; + +function useFakeStdio() { + const stdin = new PassThrough(); + const stdout = new PassThrough(); + stdout.resume(); // discard prompts + Object.defineProperty(process, "stdin", { value: stdin, configurable: true }); + Object.defineProperty(process, "stdout", { value: stdout, configurable: true }); + return stdin; +} + +function deferred() { + let resolve!: () => void; + const promise = new Promise(r => resolve = r); + return { promise, resolve }; +} + +// Reports how a promise settled, or "pending" if it has not settled within `ms`, so a hang shows up +// as an assertion failure instead of a stuck test. +async function settled(promise: Promise, ms = 1000): Promise { + let timer: NodeJS.Timeout | undefined; + const timeout = new Promise(resolve => { timer = setTimeout(() => resolve("pending"), ms); }); + try { + return await Promise.race([ + promise.then(() => "resolved", (e: Error) => `rejected: ${e.message}`), + timeout, + ]); + } + finally { + clearTimeout(timer); + } +} + +// Starts processRequests with a handler that records each request and holds "one" in flight +// until release() is called. +function startWithSlowFirstRequest() { + const started = deferred(); + const release = deferred(); + const calls: string[] = []; + const done = processRequests("> ", undefined, async request => { + calls.push(request); + if (request === "one") { + started.resolve(); + await release.promise; + } + }); + return { calls, done, started: started.promise, release: release.resolve }; +} + +describe("processRequests reading from stdin", () => { + afterEach(() => { + Object.defineProperty(process, "stdin", stdinDescriptor); + Object.defineProperty(process, "stdout", stdoutDescriptor); + }); + + it("processes a line typed while the previous request is still running", async () => { + const stdin = useFakeStdio(); + const { calls, done, started, release } = startWithSlowFirstRequest(); + stdin.write("one\n"); + await started; + stdin.write("two\n"); + await new Promise(setImmediate); + release(); + await new Promise(setImmediate); + stdin.write("quit\n"); + const result = await settled(done); + assert.deepEqual({ calls, result }, { calls: ["one", "two"], result: "resolved" }); + }); + + it("leaves input unread while more lines are waiting than readline buffers", async () => { + const stdin = useFakeStdio(); + // Every request is held until released, so input can only drain as fast as requests finish. + const gates: (() => void)[] = []; + let hold = true; + const calls: string[] = []; + const done = processRequests("> ", undefined, request => { + calls.push(request); + return hold ? new Promise(resolve => gates.push(resolve)) : Promise.resolve(); + }); + stdin.write("one\n"); + await new Promise(setImmediate); + for (let i = 0; i < 3000; i++) { + stdin.write(`line ${i}\n`); + } + await new Promise(setImmediate); + gates.shift()!(); + for (let i = 0; i < 5; i++) { + await new Promise(setImmediate); + } + const unread = stdin.readableLength; + hold = false; + gates.forEach(resolve => resolve()); + stdin.end(); + const result = await settled(done, 5000); + assert.ok(unread > 0, "all waiting input was read into memory"); + assert.deepEqual({ count: calls.length, last: calls[calls.length - 1], result }, { count: 3001, last: "line 2999", result: "resolved" }); + }); + + it("processes every line of a multi-line chunk, such as a paste or a pipe", async () => { + const stdin = useFakeStdio(); + const calls: string[] = []; + const done = processRequests("> ", undefined, async request => { calls.push(request); }); + stdin.write("one\n\ntwo\nthree\nquit\n"); + const result = await settled(done); + assert.deepEqual({ calls, result }, { calls: ["one", "two", "three"], result: "resolved" }); + }); + + it("resolves when input ends while a request is still running", async () => { + const stdin = useFakeStdio(); + const { calls, done, started, release } = startWithSlowFirstRequest(); + stdin.write("one\n"); + await started; + stdin.end("two\n"); + await once(stdin, "end"); + release(); + const result = await settled(done); + assert.deepEqual({ calls, result }, { calls: ["one", "two"], result: "resolved" }); + }); + + it("stops at quit or exit in the middle of the input", async () => { + for (const word of ["quit", "EXIT"]) { + const stdin = useFakeStdio(); + const calls: string[] = []; + const done = processRequests("> ", undefined, async request => { calls.push(request); }); + stdin.write(`one\n${word}\nthree\n`); + const result = await settled(done); + assert.deepEqual({ calls, result }, { calls: ["one"], result: "resolved" }); + } + }); +});