From 88dbb5ed13511e535dee40f7d3c578a912c608b3 Mon Sep 17 00:00:00 2001 From: ALV0612 Date: Wed, 2 Sep 2026 16:13:05 +0700 Subject: [PATCH 1/4] fix(stdin): retry transient nonblocking reads --- plugins/codex/scripts/lib/fs.mjs | 43 ++++++++- tests/fs.test.mjs | 153 +++++++++++++++++++++++++++++++ 2 files changed, 193 insertions(+), 3 deletions(-) create mode 100644 tests/fs.test.mjs diff --git a/plugins/codex/scripts/lib/fs.mjs b/plugins/codex/scripts/lib/fs.mjs index 027522442..3ee455ddf 100644 --- a/plugins/codex/scripts/lib/fs.mjs +++ b/plugins/codex/scripts/lib/fs.mjs @@ -32,9 +32,46 @@ export function isProbablyText(buffer) { return true; } -export function readStdinIfPiped() { - if (process.stdin.isTTY) { +const stdinRetryWaitArray = new Int32Array(new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT)); + +function waitForStdinRetry(delayMs) { + Atomics.wait(stdinRetryWaitArray, 0, 0, delayMs); +} + +function isTransientReadError(error) { + return error?.code === "EAGAIN" || error?.code === "EWOULDBLOCK"; +} + +export function readStdinIfPiped({ + stdin = process.stdin, + readFileSync = fs.readFileSync, + waitForRetry = waitForStdinRetry, + maxAttempts = 8, + initialRetryDelayMs = 10, + maxRetryDelayMs = 500 +} = {}) { + if (stdin.isTTY) { return ""; } - return fs.readFileSync(0, "utf8"); + + try { + Reflect.get(stdin, "_handle")?.setBlocking?.(true); + } catch { + // Some stdin handle types do not support changing blocking mode. + } + + let retryDelayMs = initialRetryDelayMs; + for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { + try { + return readFileSync(0, "utf8"); + } catch (error) { + if (!isTransientReadError(error) || attempt === maxAttempts) { + throw error; + } + waitForRetry(retryDelayMs); + retryDelayMs = Math.min(retryDelayMs * 2, maxRetryDelayMs); + } + } + + throw new Error("Unreachable stdin read state."); } diff --git a/tests/fs.test.mjs b/tests/fs.test.mjs new file mode 100644 index 000000000..c4b20f10c --- /dev/null +++ b/tests/fs.test.mjs @@ -0,0 +1,153 @@ +import fs from "node:fs"; +import test from "node:test"; +import assert from "node:assert/strict"; + +import { readStdinIfPiped } from "../plugins/codex/scripts/lib/fs.mjs"; + +function readOptions(overrides = {}) { + return { + stdin: { isTTY: false }, + readFileSync() { + return "piped prompt"; + }, + waitForRetry() { + throw new Error("unexpected retry wait"); + }, + ...overrides + }; +} + +test("readStdinIfPiped recovers when a non-blocking read is transiently unavailable", (context) => { + const transientError = Object.assign(new Error("resource temporarily unavailable"), { + code: "EAGAIN" + }); + const events = []; + let reads = 0; + + context.mock.method(fs, "readFileSync", (fd, encoding) => { + events.push(["read", fd, encoding]); + reads += 1; + if (reads === 1) { + throw transientError; + } + return "deterministic piped prompt"; + }); + + const input = readStdinIfPiped({ + stdin: { + isTTY: false, + _handle: { + setBlocking(value) { + events.push(["setBlocking", value]); + } + } + }, + waitForRetry(delayMs) { + events.push(["wait", delayMs]); + } + }); + + assert.equal(input, "deterministic piped prompt"); + assert.deepEqual(events, [ + ["setBlocking", true], + ["read", 0, "utf8"], + ["wait", 10], + ["read", 0, "utf8"] + ]); +}); + +test("readStdinIfPiped treats EWOULDBLOCK as transient and bounds backoff", () => { + const transientError = Object.assign(new Error("would block"), { + code: "EWOULDBLOCK" + }); + const waits = []; + let reads = 0; + + assert.throws( + () => + readStdinIfPiped( + readOptions({ + readFileSync() { + reads += 1; + throw transientError; + }, + waitForRetry(delayMs) { + waits.push(delayMs); + }, + maxAttempts: 5, + initialRetryDelayMs: 10, + maxRetryDelayMs: 25 + }) + ), + (error) => error === transientError + ); + assert.equal(reads, 5); + assert.deepEqual(waits, [10, 20, 25, 25]); +}); + +test("readStdinIfPiped surfaces non-transient read errors without retrying", () => { + const readError = Object.assign(new Error("bad descriptor"), { code: "EBADF" }); + let reads = 0; + + assert.throws( + () => + readStdinIfPiped( + readOptions({ + readFileSync() { + reads += 1; + throw readError; + } + }) + ), + (error) => error === readError + ); + assert.equal(reads, 1); +}); + +test("readStdinIfPiped ignores unsupported blocking mode and reads the pipe", () => { + let reads = 0; + const input = readStdinIfPiped( + readOptions({ + stdin: { + isTTY: false, + _handle: { + setBlocking() { + throw new Error("not supported"); + } + } + }, + readFileSync(fd, encoding) { + reads += 1; + assert.equal(fd, 0); + assert.equal(encoding, "utf8"); + return "ordinary pipe"; + } + }) + ); + + assert.equal(input, "ordinary pipe"); + assert.equal(reads, 1); +}); + +test("readStdinIfPiped returns empty input for a TTY without touching fd 0", () => { + let setBlockingCalls = 0; + + const input = readStdinIfPiped( + readOptions({ + stdin: { + isTTY: true, + _handle: { + setBlocking() { + setBlockingCalls += 1; + } + } + }, + readFileSync() { + throw new Error("TTY stdin must not be read"); + } + }) + ); + + assert.equal(input, ""); + assert.equal(setBlockingCalls, 0); +}); From 9c3ea57050f862e6dd40c5ab5a2752c6d0d5f57b Mon Sep 17 00:00:00 2001 From: ALV0612 Date: Wed, 2 Sep 2026 17:13:15 +0700 Subject: [PATCH 2/4] fix(stdin): preserve chunks across transient reads --- plugins/codex/scripts/lib/fs.mjs | 22 ++-- tests/fs.test.mjs | 176 +++++++++---------------------- 2 files changed, 65 insertions(+), 133 deletions(-) diff --git a/plugins/codex/scripts/lib/fs.mjs b/plugins/codex/scripts/lib/fs.mjs index 3ee455ddf..b0519566f 100644 --- a/plugins/codex/scripts/lib/fs.mjs +++ b/plugins/codex/scripts/lib/fs.mjs @@ -33,6 +33,7 @@ export function isProbablyText(buffer) { } const stdinRetryWaitArray = new Int32Array(new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT)); +const STDIN_CHUNK_SIZE = 64 * 1024; function waitForStdinRetry(delayMs) { Atomics.wait(stdinRetryWaitArray, 0, 0, delayMs); @@ -44,11 +45,12 @@ function isTransientReadError(error) { export function readStdinIfPiped({ stdin = process.stdin, - readFileSync = fs.readFileSync, + readSync = fs.readSync, waitForRetry = waitForStdinRetry, maxAttempts = 8, initialRetryDelayMs = 10, - maxRetryDelayMs = 500 + maxRetryDelayMs = 500, + chunkSize = STDIN_CHUNK_SIZE } = {}) { if (stdin.isTTY) { return ""; @@ -60,18 +62,24 @@ export function readStdinIfPiped({ // Some stdin handle types do not support changing blocking mode. } + const chunks = []; + let transientAttempts = 0; let retryDelayMs = initialRetryDelayMs; - for (let attempt = 1; attempt <= maxAttempts; attempt += 1) { + while (true) { + const chunk = Buffer.allocUnsafe(chunkSize); try { - return readFileSync(0, "utf8"); + const bytesRead = readSync(0, chunk, 0, chunk.length, null); + if (bytesRead === 0) { + return Buffer.concat(chunks).toString("utf8"); + } + chunks.push(chunk.subarray(0, bytesRead)); } catch (error) { - if (!isTransientReadError(error) || attempt === maxAttempts) { + transientAttempts += 1; + if (!isTransientReadError(error) || transientAttempts >= maxAttempts) { throw error; } waitForRetry(retryDelayMs); retryDelayMs = Math.min(retryDelayMs * 2, maxRetryDelayMs); } } - - throw new Error("Unreachable stdin read state."); } diff --git a/tests/fs.test.mjs b/tests/fs.test.mjs index c4b20f10c..ae96f7ddf 100644 --- a/tests/fs.test.mjs +++ b/tests/fs.test.mjs @@ -1,153 +1,77 @@ -import fs from "node:fs"; import test from "node:test"; import assert from "node:assert/strict"; import { readStdinIfPiped } from "../plugins/codex/scripts/lib/fs.mjs"; -function readOptions(overrides = {}) { - return { - stdin: { isTTY: false }, - readFileSync() { - return "piped prompt"; - }, - waitForRetry() { - throw new Error("unexpected retry wait"); - }, - ...overrides +function scriptedRead(steps, events = []) { + return (fd, buffer, offset, length, position) => { + events.push(["read", fd, offset, length, position]); + const step = steps.shift(); + if (step instanceof Error) throw step; + if (step == null) return 0; + const bytes = Buffer.from(step); + bytes.copy(buffer, offset); + return bytes.length; }; } -test("readStdinIfPiped recovers when a non-blocking read is transiently unavailable", (context) => { - const transientError = Object.assign(new Error("resource temporarily unavailable"), { - code: "EAGAIN" - }); - const events = []; - let reads = 0; - - context.mock.method(fs, "readFileSync", (fd, encoding) => { - events.push(["read", fd, encoding]); - reads += 1; - if (reads === 1) { - throw transientError; - } - return "deterministic piped prompt"; - }); +function transient(code = "EAGAIN") { + return Object.assign(new Error("resource temporarily unavailable"), { code }); +} +test("readStdinIfPiped preserves bytes consumed before a transient read failure", () => { + const events = []; const input = readStdinIfPiped({ - stdin: { - isTTY: false, - _handle: { - setBlocking(value) { - events.push(["setBlocking", value]); - } - } - }, - waitForRetry(delayMs) { - events.push(["wait", delayMs]); - } + stdin: { isTTY: false, _handle: { setBlocking: (value) => events.push(["blocking", value]) } }, + readSync: scriptedRead(["first-", transient(), "second\n", null], events), + waitForRetry: (delay) => events.push(["wait", delay]), + chunkSize: 32 }); - - assert.equal(input, "deterministic piped prompt"); - assert.deepEqual(events, [ - ["setBlocking", true], - ["read", 0, "utf8"], - ["wait", 10], - ["read", 0, "utf8"] - ]); + assert.equal(input, "first-second\n"); + assert.deepEqual(events[0], ["blocking", true]); + assert.deepEqual(events.filter((event) => event[0] === "wait"), [["wait", 10]]); }); test("readStdinIfPiped treats EWOULDBLOCK as transient and bounds backoff", () => { - const transientError = Object.assign(new Error("would block"), { - code: "EWOULDBLOCK" - }); + const error = transient("EWOULDBLOCK"); const waits = []; - let reads = 0; - - assert.throws( - () => - readStdinIfPiped( - readOptions({ - readFileSync() { - reads += 1; - throw transientError; - }, - waitForRetry(delayMs) { - waits.push(delayMs); - }, - maxAttempts: 5, - initialRetryDelayMs: 10, - maxRetryDelayMs: 25 - }) - ), - (error) => error === transientError - ); - assert.equal(reads, 5); + assert.throws(() => readStdinIfPiped({ + stdin: { isTTY: false }, + readSync() { throw error; }, + waitForRetry: (delay) => waits.push(delay), + maxAttempts: 5, + initialRetryDelayMs: 10, + maxRetryDelayMs: 25 + }), (actual) => actual === error); assert.deepEqual(waits, [10, 20, 25, 25]); }); test("readStdinIfPiped surfaces non-transient read errors without retrying", () => { - const readError = Object.assign(new Error("bad descriptor"), { code: "EBADF" }); - let reads = 0; - - assert.throws( - () => - readStdinIfPiped( - readOptions({ - readFileSync() { - reads += 1; - throw readError; - } - }) - ), - (error) => error === readError - ); - assert.equal(reads, 1); + const error = Object.assign(new Error("bad descriptor"), { code: "EBADF" }); + let waits = 0; + assert.throws(() => readStdinIfPiped({ + stdin: { isTTY: false }, + readSync() { throw error; }, + waitForRetry() { waits += 1; } + }), (actual) => actual === error); + assert.equal(waits, 0); }); -test("readStdinIfPiped ignores unsupported blocking mode and reads the pipe", () => { - let reads = 0; - const input = readStdinIfPiped( - readOptions({ - stdin: { - isTTY: false, - _handle: { - setBlocking() { - throw new Error("not supported"); - } - } - }, - readFileSync(fd, encoding) { - reads += 1; - assert.equal(fd, 0); - assert.equal(encoding, "utf8"); - return "ordinary pipe"; - } - }) - ); - +test("readStdinIfPiped ignores unsupported blocking mode and reads all chunks", () => { + const input = readStdinIfPiped({ + stdin: { isTTY: false, _handle: { setBlocking() { throw new Error("unsupported"); } } }, + readSync: scriptedRead(["ordinary ", "pipe", null]), + waitForRetry() { throw new Error("unexpected retry"); } + }); assert.equal(input, "ordinary pipe"); - assert.equal(reads, 1); }); test("readStdinIfPiped returns empty input for a TTY without touching fd 0", () => { - let setBlockingCalls = 0; - - const input = readStdinIfPiped( - readOptions({ - stdin: { - isTTY: true, - _handle: { - setBlocking() { - setBlockingCalls += 1; - } - } - }, - readFileSync() { - throw new Error("TTY stdin must not be read"); - } - }) - ); - + let reads = 0; + const input = readStdinIfPiped({ + stdin: { isTTY: true }, + readSync() { reads += 1; return 0; } + }); assert.equal(input, ""); - assert.equal(setBlockingCalls, 0); + assert.equal(reads, 0); }); From fdba42d1a113bb485d377255cc337fd5b9d60dd4 Mon Sep 17 00:00:00 2001 From: ALV0612 Date: Wed, 2 Sep 2026 17:36:40 +0700 Subject: [PATCH 3/4] fix(stdin): reset retry budget after progress --- plugins/codex/scripts/lib/fs.mjs | 2 ++ tests/fs.test.mjs | 17 +++++++++++++++++ 2 files changed, 19 insertions(+) diff --git a/plugins/codex/scripts/lib/fs.mjs b/plugins/codex/scripts/lib/fs.mjs index b0519566f..8cb1219ee 100644 --- a/plugins/codex/scripts/lib/fs.mjs +++ b/plugins/codex/scripts/lib/fs.mjs @@ -73,6 +73,8 @@ export function readStdinIfPiped({ return Buffer.concat(chunks).toString("utf8"); } chunks.push(chunk.subarray(0, bytesRead)); + transientAttempts = 0; + retryDelayMs = initialRetryDelayMs; } catch (error) { transientAttempts += 1; if (!isTransientReadError(error) || transientAttempts >= maxAttempts) { diff --git a/tests/fs.test.mjs b/tests/fs.test.mjs index ae96f7ddf..bd9a7ee3b 100644 --- a/tests/fs.test.mjs +++ b/tests/fs.test.mjs @@ -32,6 +32,23 @@ test("readStdinIfPiped preserves bytes consumed before a transient read failure" assert.deepEqual(events.filter((event) => event[0] === "wait"), [["wait", 10]]); }); +test("readStdinIfPiped resets the transient retry budget after each successful chunk", () => { + const steps = []; + for (let index = 0; index < 10; index += 1) { + steps.push(`chunk-${index};`, transient()); + } + steps.push(null); + const waits = []; + const input = readStdinIfPiped({ + stdin: { isTTY: false }, + readSync: scriptedRead(steps), + waitForRetry: (delay) => waits.push(delay), + maxAttempts: 2 + }); + assert.equal(input, Array.from({ length: 10 }, (_, index) => `chunk-${index};`).join("")); + assert.deepEqual(waits, Array(10).fill(10)); +}); + test("readStdinIfPiped treats EWOULDBLOCK as transient and bounds backoff", () => { const error = transient("EWOULDBLOCK"); const waits = []; From af3e94bff641c321f577d48392acb1235c700f1b Mon Sep 17 00:00:00 2001 From: ALV0612 Date: Wed, 2 Sep 2026 17:47:08 +0700 Subject: [PATCH 4/4] fix(stdin): wait for open pipe progress --- plugins/codex/scripts/lib/fs.mjs | 6 +----- tests/fs.test.mjs | 16 +++++++++------- 2 files changed, 10 insertions(+), 12 deletions(-) diff --git a/plugins/codex/scripts/lib/fs.mjs b/plugins/codex/scripts/lib/fs.mjs index 8cb1219ee..ed3ea75cb 100644 --- a/plugins/codex/scripts/lib/fs.mjs +++ b/plugins/codex/scripts/lib/fs.mjs @@ -47,7 +47,6 @@ export function readStdinIfPiped({ stdin = process.stdin, readSync = fs.readSync, waitForRetry = waitForStdinRetry, - maxAttempts = 8, initialRetryDelayMs = 10, maxRetryDelayMs = 500, chunkSize = STDIN_CHUNK_SIZE @@ -63,7 +62,6 @@ export function readStdinIfPiped({ } const chunks = []; - let transientAttempts = 0; let retryDelayMs = initialRetryDelayMs; while (true) { const chunk = Buffer.allocUnsafe(chunkSize); @@ -73,11 +71,9 @@ export function readStdinIfPiped({ return Buffer.concat(chunks).toString("utf8"); } chunks.push(chunk.subarray(0, bytesRead)); - transientAttempts = 0; retryDelayMs = initialRetryDelayMs; } catch (error) { - transientAttempts += 1; - if (!isTransientReadError(error) || transientAttempts >= maxAttempts) { + if (!isTransientReadError(error)) { throw error; } waitForRetry(retryDelayMs); diff --git a/tests/fs.test.mjs b/tests/fs.test.mjs index bd9a7ee3b..ea093ee8b 100644 --- a/tests/fs.test.mjs +++ b/tests/fs.test.mjs @@ -49,18 +49,20 @@ test("readStdinIfPiped resets the transient retry budget after each successful c assert.deepEqual(waits, Array(10).fill(10)); }); -test("readStdinIfPiped treats EWOULDBLOCK as transient and bounds backoff", () => { - const error = transient("EWOULDBLOCK"); +test("readStdinIfPiped keeps retrying transient reads until an open pipe progresses", () => { const waits = []; - assert.throws(() => readStdinIfPiped({ + const steps = Array(20).fill(null).flatMap(() => [transient("EWOULDBLOCK")]); + steps.push("eventual-data", null); + const input = readStdinIfPiped({ stdin: { isTTY: false }, - readSync() { throw error; }, + readSync: scriptedRead(steps), waitForRetry: (delay) => waits.push(delay), - maxAttempts: 5, initialRetryDelayMs: 10, maxRetryDelayMs: 25 - }), (actual) => actual === error); - assert.deepEqual(waits, [10, 20, 25, 25]); + }); + assert.equal(input, "eventual-data"); + assert.deepEqual(waits.slice(0, 5), [10, 20, 25, 25, 25]); + assert.equal(waits.length, 20); }); test("readStdinIfPiped surfaces non-transient read errors without retrying", () => {