From 009d4e886514ccd55c2b3e9bc42438b44b562b5f Mon Sep 17 00:00:00 2001 From: Mohammed Alkindi Date: Wed, 7 Oct 2026 16:08:15 -0700 Subject: [PATCH] fix(interactive): process every stdin line and resolve on EOF processRequests asked one question at a time, so lines typed or pasted while a request was running were dropped, and EOF left the promise pending or rejected it. It now iterates the readline interface, which buffers lines and ends on EOF, and prompts only while input is not paused so readline keeps its backpressure. --- typescript/package.json | 2 +- typescript/src/interactive/interactive.ts | 16 ++- typescript/test/interactive-stdin.test.ts | 139 ++++++++++++++++++++++ 3 files changed, 154 insertions(+), 3 deletions(-) create mode 100644 typescript/test/interactive-stdin.test.ts 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" }); + } + }); +});