From 23e771cdac2e90840c39f30a36953cede1f52e1e Mon Sep 17 00:00:00 2001 From: Vincent Koc Date: Tue, 3 Nov 2026 20:38:23 +0800 Subject: [PATCH] fix(node): stop queued input after stdio teardown --- CHANGELOG.md | 4 ++ src/node.ts | 5 +- test/node.test.ts | 121 ++++++++++++++++++++++------------------------ 3 files changed, 67 insertions(+), 63 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0e300ae..8b8df99 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,10 @@ All notable changes to `@openclaw/libterminal` will be documented in this file. ## Unreleased +### Fixed + +- Discard queued stdio input when an attachment ends or fails, while still restoring the terminal without waiting for an active write. + ### Changed - Run full CodeQL scans daily instead of on every main push, preserving PR and manual analysis and protecting active scheduled scans from cancellation. diff --git a/src/node.ts b/src/node.ts index 3ce190b..873c9b6 100644 --- a/src/node.ts +++ b/src/node.ts @@ -191,11 +191,12 @@ export async function attachLocalStdio( : Promise.race([operation, failure]); const waitForCleanup = (operation: Promise): Promise => aborted ? Promise.race([aborted.promise, operation]) : operation; + let acceptingInput = true; let pendingInput = Promise.resolve(); const writeInput = (data: Buffer | string) => { const bytes = typeof data === "string" ? textEncoder.encode(data) : new Uint8Array(data); pendingInput = pendingInput.then(async () => { - if (!options?.signal?.aborted) { + if (acceptingInput && !options?.signal?.aborted) { await terminal.write?.(bytes); } }); @@ -261,6 +262,8 @@ export async function attachLocalStdio( pendingOutput = undefined; } } finally { + // Queued input must not outlive the attachment, even when an active write is still pending. + acceptingInput = false; stdin.off("data", writeInput); stdin.off("error", rejectFailure); stdout.off("resize", handleResize); diff --git a/test/node.test.ts b/test/node.test.ts index 7be84cb..a9db3bd 100644 --- a/test/node.test.ts +++ b/test/node.test.ts @@ -504,69 +504,66 @@ describe("attachLocalStdio", () => { } }); - it("restores stdio after output completes without waiting forever for pending input", async () => { - const removed: string[] = []; - let paused = false; - let inputListener: (data: Buffer) => void = noop; - let completeOutput!: (result: IteratorResult) => void; - const outputResult = new Promise>((resolve) => { - completeOutput = resolve; - }); - let resolveWrite: () => void = noop; - let writeStarted: () => void = noop; - const writeStartedPromise = new Promise((resolve) => { - writeStarted = resolve; - }); - const stdin = { - isTTY: false, - readableFlowing: null, - on: (event: string, listener: (data: Buffer) => void) => { - if (event === "data") { - inputListener = listener; - } - }, - off: (event: string) => removed.push(`stdin:${event}`), - pause: () => { - paused = true; - }, - } as unknown as NodeJS.ReadStream; - const stdout = { - columns: 80, - rows: 24, - on: () => undefined, - off: (event: string) => removed.push(`stdout:${event}`), - write: (_bytes: Uint8Array, callback: (error?: Error | null) => void) => { - callback(); - return true; - }, - } as unknown as NodeJS.WriteStream; - const output = { - [Symbol.asyncIterator]: () => ({ - next: () => outputResult, - }), - }; - const attached = attachLocalStdio( - { - output, - close: async () => undefined, - write: async () => { - writeStarted(); - await new Promise((resolve) => { - resolveWrite = resolve; - }); + it.each(["output completion", "output failure"])( + "discards queued stdin after %s without waiting for an active write", + async (phase) => { + const stdin = new PassThrough(); + const stdout = new PassThrough(); + let completeOutput!: (result: IteratorResult) => void; + let rejectOutput!: (error: Error) => void; + const outputResult = new Promise>((resolve, reject) => { + completeOutput = resolve; + rejectOutput = reject; + }); + let finishWrite!: () => void; + let startWrite!: () => void; + const writing = new Promise((resolve) => { + startWrite = resolve; + }); + const writes: string[] = []; + const attached = attachLocalStdio( + { + output: { [Symbol.asyncIterator]: () => ({ next: () => outputResult }) }, + close: async () => {}, + write: async (bytes) => { + writes.push(new TextDecoder().decode(bytes)); + startWrite(); + await new Promise((resolve) => { + finishWrite = resolve; + }); + }, }, - }, - { stdin, stdout }, - ); - - inputListener(Buffer.from("pending")); - await writeStartedPromise; - completeOutput({ done: true, value: undefined }); - await attached; - expect(removed).toEqual(["stdin:data", "stdin:error", "stdout:resize", "stdout:error"]); - expect(paused).toBe(true); - resolveWrite(); - }); + { + stdin: stdin as unknown as NodeJS.ReadStream, + stdout: stdout as unknown as NodeJS.WriteStream, + }, + ); + stdin.write("pending"); + stdin.write("queued"); + await writing; + try { + if (phase === "output failure") { + const error = new Error("output failed"); + rejectOutput(error); + await expect(attached).rejects.toBe(error); + } else { + completeOutput({ done: true, value: undefined }); + await attached; + } + expect(stdin.listenerCount("data")).toBe(0); + expect(stdin.listenerCount("error")).toBe(0); + expect(stdout.listenerCount("resize")).toBe(0); + expect(stdout.listenerCount("error")).toBe(0); + expect(stdin.readableFlowing).toBe(false); + } finally { + finishWrite(); + await setImmediate(); + stdin.destroy(); + stdout.destroy(); + } + expect(writes).toEqual(["pending"]); + }, + ); it("rejects and restores stdio when writing stdin fails", async () => { const stdio = testStdio();