From 8ebb1d2aec6fc620d24e2133fcc92f43fc49dcf7 Mon Sep 17 00:00:00 2001 From: Monkey-wusky Date: Thu, 16 Jul 2026 18:25:55 +0800 Subject: [PATCH] fix(sessions): complete tail-read windows despite short positional reads (#108655) * fix(sessions): complete tail-read windows despite short positional reads Replace single-shot handle.read() calls in two session tail-read paths with readFileWindowFully so multibyte-range positional reads do not silently return incomplete tail data on network filesystems. - readRecentTranscriptTailLinesAsync: tail-window read for recent session messages now loops until the requested window fills - readLastMessagePreviewFromOpenTranscriptAsync: last-message preview tail read now loops until the requested window fills The same readFileWindowFully helper was introduced in #108253 and expanded with a sync variant in #108127 (both by sunlit-deng). * fix(sessions): complete sync title preview reads --------- Co-authored-by: Peter Steinberger --- src/gateway/session-utils.fs.test.ts | 159 +++++++++++++++++++++++++++ src/gateway/session-utils.fs.ts | 12 +- 2 files changed, 167 insertions(+), 4 deletions(-) diff --git a/src/gateway/session-utils.fs.test.ts b/src/gateway/session-utils.fs.test.ts index bb1a8afe6228..3f8d81b399d3 100644 --- a/src/gateway/session-utils.fs.test.ts +++ b/src/gateway/session-utils.fs.test.ts @@ -2036,4 +2036,163 @@ describe("oversized transcript line guards", () => { expect(asyncResult.lastMessagePreview).toBe("Bot says hello"); }); }); + +describe("short read resilience", () => { + let tmpDir: string; + let storePath: string; + + registerTempSessionStore("openclaw-short-read-test-", (nextTmpDir, nextStorePath) => { + tmpDir = nextTmpDir; + storePath = nextStorePath; + }); + + function installAsyncShortReadProxy(maxPerCall = 16) { + const realOpen = fs.promises.open.bind(fs.promises); + let shortReadCalls = 0; + vi.spyOn(fs.promises, "open").mockImplementation(async (...args: unknown[]) => { + const handle = await realOpen(...(args as Parameters)); + const realRead = handle.read.bind(handle); + return new Proxy(handle, { + get(target, prop, receiver) { + if (prop !== "read") { + return Reflect.get(target, prop, receiver); + } + return (buf: Buffer, offset: number, length: number, position: number | null) => { + const cappedLength = position !== null && position > 0 ? maxPerCall : length; + if (cappedLength < length) { + shortReadCalls += 1; + } + return realRead(buf, offset, Math.min(length, cappedLength), position); + }; + }, + }); + }); + return () => shortReadCalls; + } + + function installSyncShortReadProxy(maxPerCall = 16) { + const realReadSync = fs.readSync.bind(fs); + let shortReadCalls = 0; + const readSpy = vi.spyOn(fs, "readSync").mockImplementation((( + fd: number, + buffer: NodeJS.ArrayBufferView, + offset: number, + length: number, + position: fs.ReadPosition | null, + ) => { + const cappedLength = typeof position === "number" && position > 0 ? maxPerCall : length; + if (cappedLength < length) { + shortReadCalls += 1; + } + return realReadSync(fd, buffer, offset, Math.min(length, cappedLength), position); + }) as typeof fs.readSync); + return { readSpy, getShortReadCalls: () => shortReadCalls }; + } + + function buildLargeTitleTranscript(sessionId: string) { + return [ + { type: "session", version: 1, id: sessionId }, + { message: { role: "user", content: "head title" } }, + ...Array.from({ length: 40 }, (_, index) => ({ + message: { role: "assistant", content: `filler ${index} ${"x".repeat(512)}` }, + })), + { message: { role: "assistant", content: "tail preview" } }, + ]; + } + + test("readRecentSessionMessagesAsync survives 16-byte tail read caps", async () => { + const sessionId = "test-short-read-recent"; + const lines = [ + { type: "session", version: 1, id: sessionId }, + ...Array.from({ length: 30 }, (_, i) => ({ + message: { + role: i % 2 ? "assistant" : "user", + content: `message ${i}: ${"data ".repeat(80)}`, + }, + })), + ]; + writeTranscript(tmpDir, sessionId, lines); + + const expected = await readRecentSessionMessagesAsync(sessionId, storePath, undefined, { + maxMessages: 20, + maxBytes: 8192, + }); + const getShortReadCalls = installAsyncShortReadProxy(16); + try { + const actual = await readRecentSessionMessagesAsync(sessionId, storePath, undefined, { + maxMessages: 20, + maxBytes: 8192, + }); + expect(actual).toEqual(expected); + expect(getShortReadCalls()).toBeGreaterThan(1); + } finally { + vi.restoreAllMocks(); + } + }); + + test("readRecentSessionMessagesAsync honors maxBytes under short reads", async () => { + const sessionId = "test-short-read-byte-cap"; + const lines = [ + { type: "session", version: 1, id: sessionId }, + ...Array.from({ length: 20 }, (_, i) => ({ + message: { + role: i % 2 ? "assistant" : "user", + content: `line ${String(i).padStart(2, "0")}: ${"payload ".repeat(30)}`, + }, + })), + ]; + writeTranscript(tmpDir, sessionId, lines); + + const normal = await readRecentSessionMessagesAsync(sessionId, storePath, undefined, { + maxMessages: 20, + maxBytes: 4096, + }); + + const getShortReadCalls = installAsyncShortReadProxy(64); + try { + const short = await readRecentSessionMessagesAsync(sessionId, storePath, undefined, { + maxMessages: 20, + maxBytes: 4096, + }); + expect(short).toEqual(normal); + expect(getShortReadCalls()).toBeGreaterThan(1); + } finally { + vi.restoreAllMocks(); + } + }); + + test("reads async title fields across short positional tail reads", async () => { + const sessionId = "test-short-read-title-async"; + writeTranscript(tmpDir, sessionId, buildLargeTitleTranscript(sessionId)); + + const getShortReadCalls = installAsyncShortReadProxy(16); + try { + await expect( + readSessionTitleFieldsFromTranscriptAsync(sessionId, storePath), + ).resolves.toEqual({ + firstUserMessage: "head title", + lastMessagePreview: "tail preview", + }); + expect(getShortReadCalls()).toBeGreaterThan(1); + } finally { + vi.restoreAllMocks(); + } + }); + + test("reads sync title fields across short positional tail reads", () => { + const sessionId = "test-short-read-title-sync"; + writeTranscript(tmpDir, sessionId, buildLargeTitleTranscript(sessionId)); + + const { readSpy, getShortReadCalls } = installSyncShortReadProxy(16); + try { + expect(readSessionTitleFieldsFromTranscript(sessionId, storePath)).toEqual({ + firstUserMessage: "head title", + lastMessagePreview: "tail preview", + }); + expect(getShortReadCalls()).toBeGreaterThan(1); + } finally { + readSpy.mockRestore(); + } + }); +}); /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/src/gateway/session-utils.fs.ts b/src/gateway/session-utils.fs.ts index f0e57755a58c..b93b9f8df984 100644 --- a/src/gateway/session-utils.fs.ts +++ b/src/gateway/session-utils.fs.ts @@ -19,6 +19,7 @@ import { scanSessionTranscriptTree, selectSessionTranscriptTreePathNodes, } from "../config/sessions/transcript-tree.js"; +import { readFileWindowFully, readFileWindowFullySync } from "../infra/file-read.js"; import { jsonUtf8Bytes } from "../infra/json-utf8-bytes.js"; import { hasInterSessionUserProvenance } from "../sessions/input-provenance.js"; import { extractAssistantVisibleText } from "../shared/chat-message-content.js"; @@ -255,7 +256,7 @@ async function readRecentTranscriptTailLinesAsync( const handle = await fs.promises.open(filePath, "r"); try { const buffer = Buffer.alloc(readLen); - const { bytesRead } = await handle.read(buffer, 0, readLen, readStart); + const bytesRead = await readFileWindowFully(handle, buffer, readStart); if (bytesRead <= 0) { return []; } @@ -1174,9 +1175,12 @@ function readLastMessagePreviewFromOpenTranscript(params: { const readStart = Math.max(0, params.size - LAST_MSG_MAX_BYTES); const readLen = Math.min(params.size, LAST_MSG_MAX_BYTES); const buf = Buffer.alloc(readLen); - fs.readSync(params.fd, buf, 0, readLen, readStart); + const bytesRead = readFileWindowFullySync(params.fd, buf, readStart); + if (bytesRead <= 0) { + return null; + } - const chunk = buf.toString("utf-8"); + const chunk = buf.toString("utf-8", 0, bytesRead); const lines = chunk.split(/\r?\n/).filter((l) => l.trim()); return extractLastMessagePreviewFromTranscriptLines(lines.slice(-LAST_MSG_MAX_LINES)); } @@ -1188,7 +1192,7 @@ async function readLastMessagePreviewFromOpenTranscriptAsync(params: { const readStart = Math.max(0, params.size - LAST_MSG_MAX_BYTES); const readLen = Math.min(params.size, LAST_MSG_MAX_BYTES); const buffer = Buffer.alloc(readLen); - const { bytesRead } = await params.handle.read(buffer, 0, readLen, readStart); + const bytesRead = await readFileWindowFully(params.handle, buffer, readStart); if (bytesRead <= 0) { return null; }