From af6fa19c9211a91d6e387b26278729e971995f0d Mon Sep 17 00:00:00 2001 From: Akinfenwa Obaseyi Date: Tue, 3 Nov 2026 10:20:03 +0100 Subject: [PATCH] fix(io): wait for delayed stdin under Node --- src/io/source.test.ts | 44 +++++++++++++++++++++++++++++++++++++++++++ src/io/source.ts | 6 +++--- 2 files changed, 47 insertions(+), 3 deletions(-) diff --git a/src/io/source.test.ts b/src/io/source.test.ts index 8c99026bed..8226c7b861 100644 --- a/src/io/source.test.ts +++ b/src/io/source.test.ts @@ -1,4 +1,6 @@ import { afterEach, describe, expect, test } from "bun:test"; +import { spawn } from "node:child_process"; +import { once } from "node:events"; import { mkdtemp, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { join } from "node:path"; @@ -57,6 +59,48 @@ describe("SourceResolver bytes", () => { expect(await resolver.resolveBytes("payload", "-")).toEqual(bytes); }); + test("waits for delayed process stdin under Node", async () => { + const build = await Bun.build({ + entrypoints: [join(import.meta.dir, "source.ts")], + target: "node", + format: "cjs", + }); + expect(build.success).toBe(true); + const directory = await mkdtemp(join(tmpdir(), "agentcore-source-node-")); + tempDirectories.push(directory); + const bundle = join(directory, "source.cjs"); + await writeFile(bundle, await build.outputs[0]!.text()); + + const child = spawn( + "node", + [ + "-e", + ` + const { SourceResolver } = require(process.argv[1]); + new SourceResolver({ stdin: process.stdin }).resolveText("bearer-token", "-") + .then(value => process.stdout.write(value)) + .catch(error => { console.error(error); process.exitCode = 1; }); + `, + bundle, + ], + { stdio: ["pipe", "pipe", "pipe"] }, + ); + let output = ""; + let errors = ""; + child.stdout.on("data", (chunk) => { + output += chunk; + }); + child.stderr.on("data", (chunk) => { + errors += chunk; + }); + setTimeout(() => child.stdin.end("late value"), 150); + + const [code] = await once(child, "close"); + expect(code).toBe(0); + expect(errors).toBe(""); + expect(output).toBe("late value"); + }); + test("rejects a second stdin consumer and identifies both options", async () => { const resolver = new SourceResolver({ stdin: stdin(new TextEncoder().encode("value")) }); diff --git a/src/io/source.ts b/src/io/source.ts index 5454ca8c5d..88a03831a0 100644 --- a/src/io/source.ts +++ b/src/io/source.ts @@ -85,10 +85,10 @@ export class SourceResolver { // Importing Ink under Bun drains the process.stdin stream object before a // headless handler can claim it, while file descriptor 0 still retains the - // bytes. Use a fresh non-closing stream for the real process input; injected - // AppIO streams continue through the ordinary testable path. + // bytes. Node can make fd 0 non-blocking when process.stdin is accessed, so + // use its stream directly there to wait for delayed input. const source = - this.config.stdin === process.stdin + this.config.stdin === process.stdin && process.versions.bun !== undefined ? createReadStream("", { fd: 0, autoClose: false }) : this.config.stdin; const stdin = this.config.signal ? addAbortSignal(this.config.signal, source) : source;