From 8be8ef3e27dd3e40bfe740bc5ee18fe8fb6d66f7 Mon Sep 17 00:00:00 2001 From: tt-a1i <53142663+tt-a1i@users.noreply.github.com> Date: Sun, 21 Jun 2026 06:39:03 +0800 Subject: [PATCH] fix(cli): handle truncated remote input files (#5473) Co-authored-by: Shaojin Wen --- .../remoteInput/RemoteInputWatcher.test.ts | 56 ++++++++++++ .../cli/src/remoteInput/RemoteInputWatcher.ts | 86 ++++++++++++++++++- 2 files changed, 141 insertions(+), 1 deletion(-) diff --git a/packages/cli/src/remoteInput/RemoteInputWatcher.test.ts b/packages/cli/src/remoteInput/RemoteInputWatcher.test.ts index eade1dd793..331d78ac8a 100644 --- a/packages/cli/src/remoteInput/RemoteInputWatcher.test.ts +++ b/packages/cli/src/remoteInput/RemoteInputWatcher.test.ts @@ -114,6 +114,62 @@ describe('RemoteInputWatcher', () => { expect(submitted).toEqual(['after-bad-line']); }); + it('reads commands written after the input file is truncated', async () => { + watcher = new RemoteInputWatcher(inputFile); + const submitted: string[] = []; + watcher.setSubmitFn((text) => { + submitted.push(text); + }); + + fs.appendFileSync( + inputFile, + JSON.stringify({ type: 'submit', text: 'before-truncate' }) + '\n', + ); + await watcher.checkForNewInput(); + const consumedSize = fs.statSync(inputFile).size; + + const afterTruncate = 'after-truncate-with-a-longer-command'; + fs.writeFileSync( + inputFile, + JSON.stringify({ type: 'submit', text: afterTruncate }) + '\n', + ); + expect(fs.statSync(inputFile).size).toBeGreaterThan(consumedSize); + await watcher.checkForNewInput(); + + expect(submitted).toEqual(['before-truncate', afterTruncate]); + }); + + it('reads commands after truncation rewrites the file to the same size', async () => { + watcher = new RemoteInputWatcher(inputFile); + const submitted: string[] = []; + watcher.setSubmitFn((text) => { + submitted.push(text); + }); + + const beforeTruncate = 'before-truncate'; + const afterTruncate = 'after--truncate'; + const fixedMtime = new Date('2026-06-20T00:00:00.000Z'); + expect(afterTruncate).toHaveLength(beforeTruncate.length); + + fs.appendFileSync( + inputFile, + JSON.stringify({ type: 'submit', text: beforeTruncate }) + '\n', + ); + fs.utimesSync(inputFile, fixedMtime, fixedMtime); + await watcher.checkForNewInput(); + const consumedSize = fs.statSync(inputFile).size; + + fs.writeFileSync( + inputFile, + JSON.stringify({ type: 'submit', text: afterTruncate }) + '\n', + ); + fs.utimesSync(inputFile, fixedMtime, fixedMtime); + expect(fs.statSync(inputFile).size).toBe(consumedSize); + await watcher.checkForNewInput(); + + expect(submitted).toEqual([beforeTruncate, afterTruncate]); + }); + it('stops watching after shutdown', async () => { watcher = new RemoteInputWatcher(inputFile); const submitted: string[] = []; diff --git a/packages/cli/src/remoteInput/RemoteInputWatcher.ts b/packages/cli/src/remoteInput/RemoteInputWatcher.ts index 4a30551036..5dd61133ac 100644 --- a/packages/cli/src/remoteInput/RemoteInputWatcher.ts +++ b/packages/cli/src/remoteInput/RemoteInputWatcher.ts @@ -4,7 +4,16 @@ * SPDX-License-Identifier: Apache-2.0 */ -import { createReadStream, watchFile, unwatchFile, statSync } from 'node:fs'; +import { createHash } from 'node:crypto'; +import { + closeSync, + createReadStream, + openSync, + readSync, + statSync, + unwatchFile, + watchFile, +} from 'node:fs'; import { createInterface } from 'node:readline'; import { createDebugLogger } from '@qwen-code/qwen-code-core'; @@ -51,6 +60,7 @@ export class RemoteInputWatcher { private processing = false; private active = true; private bytesRead = 0; + private consumedPrefixHash: string | null = null; private reading = false; private filePath: string; private retryTimer: ReturnType | null = null; @@ -95,8 +105,10 @@ export class RemoteInputWatcher { try { const stat = statSync(this.filePath); this.bytesRead = stat.size; + this.consumedPrefixHash = this.hashFilePrefix(this.bytesRead); } catch { this.bytesRead = 0; + this.consumedPrefixHash = null; } watchFile(this.filePath, { interval: this.pollIntervalMs }, () => { @@ -128,8 +140,25 @@ export class RemoteInputWatcher { return Promise.resolve(); } + // Size alone misses truncate+rewrite that lands at the same or a larger + // size. Append-only writes preserve the consumed prefix hash; rewrites do not. + if (currentSize < this.bytesRead) { + debugLogger.debug( + 'RemoteInput: input file shrank, resetting read offset', + ); + this.bytesRead = 0; + this.consumedPrefixHash = null; + } else if (this.hasConsumedPrefixChanged()) { + debugLogger.debug( + 'RemoteInput: input file prefix changed, resetting read offset', + ); + this.bytesRead = 0; + this.consumedPrefixHash = null; + } + if (currentSize <= this.bytesRead) return Promise.resolve(); + const nextConsumedPrefixHash = this.hashFilePrefix(currentSize); this.reading = true; const stream = createReadStream(this.filePath, { start: this.bytesRead, @@ -179,6 +208,9 @@ export class RemoteInputWatcher { return new Promise((resolve) => { rl.on('close', () => { this.bytesRead = currentSize; + if (nextConsumedPrefixHash !== null) { + this.consumedPrefixHash = nextConsumedPrefixHash; + } this.reading = false; this.processQueue(); resolve(); @@ -186,6 +218,58 @@ export class RemoteInputWatcher { }); } + private hasConsumedPrefixChanged(): boolean { + if (this.bytesRead === 0) { + return false; + } + if (this.consumedPrefixHash === null) { + debugLogger.warn( + 'RemoteInput: missing consumed prefix hash, resetting read offset', + ); + return true; + } + + const currentHash = this.hashFilePrefix(this.bytesRead); + if (currentHash === null) { + debugLogger.warn( + 'RemoteInput: failed to hash consumed prefix, resetting read offset', + ); + return true; + } + return currentHash !== this.consumedPrefixHash; + } + + private hashFilePrefix(size: number): string | null { + if (size <= 0) return null; + + let fd: number | null = null; + try { + fd = openSync(this.filePath, 'r'); + const hash = createHash('sha256'); + const buffer = Buffer.allocUnsafe(Math.min(64 * 1024, size)); + let remaining = size; + let position = 0; + + while (remaining > 0) { + const bytesToRead = Math.min(buffer.length, remaining); + const bytesRead = readSync(fd, buffer, 0, bytesToRead, position); + if (bytesRead <= 0) return null; + hash.update(buffer.subarray(0, bytesRead)); + remaining -= bytesRead; + position += bytesRead; + } + + return hash.digest('base64'); + } catch (err) { + debugLogger.warn('RemoteInput: failed to hash file prefix:', err); + return null; + } finally { + if (fd !== null) { + closeSync(fd); + } + } + } + private async processQueue(): Promise { if (this.processing || !this.submitFn || this.queue.length === 0) return;