Skip to content

Commit 552470c

Browse files
steipetesteipete
committed
fix(sessions): stream transcript reverse scans
1 parent 42662bc commit 552470c

5 files changed

Lines changed: 176 additions & 103 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ Docs: https://docs.openclaw.ai
5252

5353
### Fixes
5454

55-
- Sessions/transcripts: replace whole-file `readFile` scans with shared streaming helpers (`streamSessionTranscriptLines` and `readSessionTranscriptTailLines`) for idempotency lookup, latest/tail assistant text reads, delivery-mirror dedupe, and compaction fork loading, so long-running sessions no longer materialize the full transcript in memory. Forward scans use `readline` over a bounded `createReadStream`; reverse scans read only the last 4 MiB (clamped to 64 MiB) and discard the leading partial line. Synthetic 200 MiB transcript: peak RSS delta drops from +252 MiB to +27 MiB while preserving malformed-line tolerance and idempotency-key return semantics. Fixes #54296. Thanks @jack-stormentswe.
55+
- Sessions/transcripts: replace whole-file `readFile` scans with shared streaming helpers (`streamSessionTranscriptLines` and `streamSessionTranscriptLinesReverse`) for idempotency lookup, latest/tail assistant text reads, delivery-mirror dedupe, and compaction fork loading, so long-running sessions no longer materialize the full transcript in memory. Forward scans use `readline` over a bounded `createReadStream`; reverse scans read bounded chunks from the file end and decode complete JSONL lines newest-first without a fixed tail cap. Synthetic 200 MiB transcript: peak RSS delta drops from +252 MiB to +27 MiB while preserving malformed-line tolerance and idempotency-key return semantics. Fixes #54296. Thanks @jack-stormentswe.
5656
- WhatsApp: apply hot-reloaded `dmPolicy` and `allowFrom` settings to the active Web listener before processing new inbound DMs. Fixes #80538. Thanks @Ampaskopi129.
5757
- Plugins: let `openclaw doctor --fix` repair managed plugin installs whose package entrypoints fail package-directory boundary validation after local state moves. Fixes #80592. Thanks @wei-wei-zhao.
5858
- Voice-call: resume voice-originated exec approval follow-ups as internal non-delivery turns instead of rejecting them as `unknown channel: voice`. Fixes #80540. Thanks @patrickmch.

‎src/config/sessions/transcript-stream.test.ts‎

Lines changed: 68 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -3,8 +3,8 @@ import os from "node:os";
33
import path from "node:path";
44
import { afterEach, beforeEach, describe, expect, it } from "vitest";
55
import {
6-
readSessionTranscriptTailLines,
76
streamSessionTranscriptLines,
7+
streamSessionTranscriptLinesReverse,
88
} from "./transcript-stream.js";
99

1010
// Regression coverage for #54296: the transcript readers must stay correct and
@@ -95,67 +95,100 @@ describe("streamSessionTranscriptLines", () => {
9595
});
9696
});
9797

98-
describe("readSessionTranscriptTailLines", () => {
99-
it("returns trimmed non-empty lines in reverse order for short files", async () => {
98+
describe("streamSessionTranscriptLinesReverse", () => {
99+
it("yields trimmed non-empty lines in reverse order for short files", async () => {
100100
fs.writeFileSync(transcriptPath, "first\nsecond\nthird\n", "utf-8");
101101

102-
const lines = await readSessionTranscriptTailLines(transcriptPath);
102+
const lines = await collect(streamSessionTranscriptLinesReverse(transcriptPath));
103103

104104
expect(lines).toEqual(["third", "second", "first"]);
105105
});
106106

107-
it("returns undefined when the file cannot be opened", async () => {
108-
const lines = await readSessionTranscriptTailLines(path.join(tempDir, "missing.jsonl"));
107+
it("returns an empty iterator when the file cannot be opened", async () => {
108+
const lines = await collect(
109+
streamSessionTranscriptLinesReverse(path.join(tempDir, "missing.jsonl")),
110+
);
109111

110-
expect(lines).toBeUndefined();
112+
expect(lines).toEqual([]);
111113
});
112114

113-
it("returns an empty array for an empty file", async () => {
115+
it("returns an empty iterator for an empty file", async () => {
114116
fs.writeFileSync(transcriptPath, "", "utf-8");
115117

116-
const lines = await readSessionTranscriptTailLines(transcriptPath);
118+
const lines = await collect(streamSessionTranscriptLinesReverse(transcriptPath));
117119

118120
expect(lines).toEqual([]);
119121
});
120122

121-
it("drops the leading partial line when the window does not start at byte zero", async () => {
122-
// Build a file longer than the requested tail window. The first line of
123-
// the slice we end up reading should be a suffix of an earlier line; the
124-
// helper must discard it so callers do not see corrupt JSON.
125-
const longPrefix = "x".repeat(2048);
126-
const content = `${longPrefix}\nbeta\ngamma\n`;
127-
fs.writeFileSync(transcriptPath, content, "utf-8");
123+
it("preserves complete lines across chunk boundaries", async () => {
124+
const longLine = "x".repeat(2048);
125+
fs.writeFileSync(transcriptPath, `${longLine}\nbeta\ngamma\n`, "utf-8");
128126

129-
const lines = await readSessionTranscriptTailLines(transcriptPath, {
130-
maxBytes: 16,
131-
});
127+
const lines = await collect(
128+
streamSessionTranscriptLinesReverse(transcriptPath, {
129+
chunkBytes: 1024,
130+
}),
131+
);
132132

133-
expect(lines).not.toContain(longPrefix);
134-
expect(lines).toEqual(["gamma", "beta"]);
133+
expect(lines).toEqual(["gamma", "beta", longLine]);
135134
});
136135

137-
it("does not drop the first line when the window covers the entire file", async () => {
138-
fs.writeFileSync(transcriptPath, "alpha\nbeta\ngamma\n", "utf-8");
136+
it("preserves multibyte UTF-8 across chunk boundaries", async () => {
137+
const firstLine = `${"a".repeat(1100)}🌊`;
138+
const secondLine = `${"b".repeat(1100)}✅`;
139+
fs.writeFileSync(transcriptPath, `${firstLine}\n${secondLine}\n`, "utf-8");
139140

140-
const lines = await readSessionTranscriptTailLines(transcriptPath, {
141-
maxBytes: 64 * 1024,
142-
});
141+
const lines = await collect(
142+
streamSessionTranscriptLinesReverse(transcriptPath, {
143+
chunkBytes: 1024,
144+
}),
145+
);
143146

144-
expect(lines).toEqual(["gamma", "beta", "alpha"]);
147+
expect(lines).toEqual([secondLine, firstLine]);
148+
});
149+
150+
it("honours an abort signal between reverse lines", async () => {
151+
fs.writeFileSync(transcriptPath, "one\ntwo\nthree\n", "utf-8");
152+
const controller = new AbortController();
153+
154+
const out: string[] = [];
155+
for await (const line of streamSessionTranscriptLinesReverse(transcriptPath, {
156+
signal: controller.signal,
157+
})) {
158+
out.push(line);
159+
if (line === "three") {
160+
controller.abort();
161+
}
162+
}
163+
164+
expect(out).toEqual(["three"]);
145165
});
146166

147-
it("clamps a sub-minimum maxBytes to the floor instead of returning nothing", async () => {
167+
it("clamps a sub-minimum chunk size without dropping older lines", async () => {
148168
fs.writeFileSync(transcriptPath, "alpha\nbeta\ngamma\n", "utf-8");
149169

150-
const lines = await readSessionTranscriptTailLines(transcriptPath, {
151-
maxBytes: 16,
152-
});
170+
const lines = await collect(
171+
streamSessionTranscriptLinesReverse(transcriptPath, {
172+
chunkBytes: 16,
173+
}),
174+
);
153175

154-
// The full file is smaller than the 1 KiB floor, so we still read the
155-
// whole file and return all three lines in reverse order.
156176
expect(lines).toEqual(["gamma", "beta", "alpha"]);
157177
});
158178

179+
it("does not emit a partial prefix until the full first line is available", async () => {
180+
const firstLine = "prefix".repeat(400);
181+
fs.writeFileSync(transcriptPath, `${firstLine}\nbeta\ngamma`, "utf-8");
182+
183+
const lines = await collect(
184+
streamSessionTranscriptLinesReverse(transcriptPath, {
185+
chunkBytes: 1024,
186+
}),
187+
);
188+
189+
expect(lines).toEqual(["gamma", "beta", firstLine]);
190+
});
191+
159192
it("preserves JSONL line ordering so reverse scans hit the newest match first", async () => {
160193
fs.writeFileSync(
161194
transcriptPath,
@@ -167,10 +200,9 @@ describe("readSessionTranscriptTailLines", () => {
167200
"utf-8",
168201
);
169202

170-
const lines = await readSessionTranscriptTailLines(transcriptPath);
203+
const lines = await collect(streamSessionTranscriptLinesReverse(transcriptPath));
204+
const parsed = lines.map((line) => JSON.parse(line) as { id: string });
171205

172-
expect(lines).toBeDefined();
173-
const parsed = lines!.map((line) => JSON.parse(line) as { id: string });
174206
expect(parsed.map((entry) => entry.id)).toEqual(["third", "second", "first"]);
175207
});
176208
});

‎src/config/sessions/transcript-stream.ts‎

Lines changed: 59 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -7,22 +7,21 @@ import readline from "node:readline";
77
// splitting on newlines. That worked fine for short sessions but produced real
88
// memory pressure on long-running ones where transcripts grow to tens or
99
// hundreds of MB (see #54296). These helpers replace the whole-file reads with
10-
// either a forward `readline` stream (bounded to one line of memory at a time)
11-
// or a tail-only read that scans the last N bytes — both preserve the
12-
// malformed-line tolerance and "first/last match wins" semantics callers rely
13-
// on.
10+
// either a forward `readline` stream or a chunked reverse scan. Both are bounded
11+
// to a small chunk plus the current line and preserve the malformed-line
12+
// tolerance and "first/last match wins" semantics callers rely on.
1413

15-
const DEFAULT_TAIL_BYTES = 4 * 1024 * 1024;
16-
const ABSOLUTE_TAIL_CAP_BYTES = 64 * 1024 * 1024;
17-
const MIN_TAIL_BYTES = 1024;
14+
const DEFAULT_REVERSE_CHUNK_BYTES = 64 * 1024;
15+
const MAX_REVERSE_CHUNK_BYTES = 1024 * 1024;
16+
const MIN_REVERSE_CHUNK_BYTES = 1024;
1817

1918
export type TranscriptStreamOptions = {
2019
signal?: AbortSignal;
2120
};
2221

23-
export type TranscriptTailOptions = {
24-
/** Maximum bytes to read from the file tail. Clamped to [1KiB, 64MiB]. */
25-
maxBytes?: number;
22+
export type TranscriptReverseStreamOptions = TranscriptStreamOptions & {
23+
/** Bytes read per reverse scan chunk. Clamped to [1KiB, 1MiB]. */
24+
chunkBytes?: number;
2625
};
2726

2827
/**
@@ -68,62 +67,75 @@ export async function* streamSessionTranscriptLines(
6867
}
6968

7069
/**
71-
* Read the last `maxBytes` of a transcript file and return its non-empty
72-
* trimmed lines in reverse (newest-first) order. The first line of the slice
73-
* is discarded when the slice does not start at the file head, because it can
74-
* be the suffix of an earlier line split mid-way.
70+
* Stream the non-empty, trimmed JSONL lines of a transcript file in reverse
71+
* (newest-first) order.
7572
*
76-
* Returns `undefined` if the file cannot be opened, `[]` if it exists but is
77-
* empty, and the trimmed reversed lines otherwise. Callers can return on the
78-
* first match without buffering the rest of the file.
73+
* Returns an empty async iterator if the file cannot be opened, is empty, or is
74+
* not a regular file. The implementation splits on newline bytes before UTF-8
75+
* decoding so multibyte characters survive arbitrary chunk boundaries.
7976
*/
80-
export async function readSessionTranscriptTailLines(
77+
export async function* streamSessionTranscriptLinesReverse(
8178
filePath: string,
82-
options: TranscriptTailOptions = {},
83-
): Promise<string[] | undefined> {
84-
const requestedMaxBytes = Number.isFinite(options.maxBytes)
85-
? Math.max(MIN_TAIL_BYTES, Math.floor(options.maxBytes as number))
86-
: DEFAULT_TAIL_BYTES;
87-
const cappedMaxBytes = Math.min(requestedMaxBytes, ABSOLUTE_TAIL_CAP_BYTES);
79+
options: TranscriptReverseStreamOptions = {},
80+
): AsyncGenerator<string> {
81+
const requestedChunkBytes = Number.isFinite(options.chunkBytes)
82+
? Math.max(MIN_REVERSE_CHUNK_BYTES, Math.floor(options.chunkBytes as number))
83+
: DEFAULT_REVERSE_CHUNK_BYTES;
84+
const chunkBytes = Math.min(requestedChunkBytes, MAX_REVERSE_CHUNK_BYTES);
8885

8986
let fileHandle: Awaited<ReturnType<typeof fs.promises.open>>;
9087
try {
9188
fileHandle = await fs.promises.open(filePath, "r");
9289
} catch {
93-
return undefined;
90+
return;
9491
}
9592
try {
9693
const stat = await fileHandle.stat();
97-
if (!stat.isFile()) {
98-
return undefined;
99-
}
100-
if (stat.size <= 0) {
101-
return [];
102-
}
103-
const readLength = Math.min(stat.size, cappedMaxBytes);
104-
const readStart = Math.max(0, stat.size - readLength);
105-
const buffer = await readFileRangeAsync(fileHandle, readStart, readLength);
106-
const text = buffer.toString("utf-8");
107-
const rawLines = text.split(/\r?\n/);
108-
if (readStart > 0 && rawLines.length > 0) {
109-
rawLines.shift();
94+
if (!stat.isFile() || stat.size <= 0 || options.signal?.aborted) {
95+
return;
11096
}
111-
const lines: string[] = [];
112-
for (let index = rawLines.length - 1; index >= 0; index -= 1) {
113-
const trimmed = rawLines[index].trim();
114-
if (!trimmed) {
115-
continue;
97+
98+
let position = stat.size;
99+
let carry: Buffer = Buffer.alloc(0);
100+
while (position > 0) {
101+
if (options.signal?.aborted) {
102+
return;
103+
}
104+
const readLength = Math.min(position, chunkBytes);
105+
position -= readLength;
106+
const chunk = await readFileRangeAsync(fileHandle, position, readLength);
107+
const combined = carry.length > 0 ? Buffer.concat([chunk, carry]) : chunk;
108+
let lineEnd = combined.length;
109+
for (let index = combined.length - 1; index >= 0; index -= 1) {
110+
if (combined[index] !== 0x0a) {
111+
continue;
112+
}
113+
const line = decodeTrimmedLine(combined.subarray(index + 1, lineEnd));
114+
if (line) {
115+
yield line;
116+
if (options.signal?.aborted) {
117+
return;
118+
}
119+
}
120+
lineEnd = index;
116121
}
117-
lines.push(trimmed);
122+
carry = combined.subarray(0, lineEnd);
123+
}
124+
125+
const firstLine = decodeTrimmedLine(carry);
126+
if (firstLine && !options.signal?.aborted) {
127+
yield firstLine;
118128
}
119-
return lines;
120-
} catch {
121-
return undefined;
122129
} finally {
123130
await fileHandle.close().catch(() => undefined);
124131
}
125132
}
126133

134+
function decodeTrimmedLine(line: Buffer): string {
135+
const trimmed = line.toString("utf-8").trim();
136+
return trimmed;
137+
}
138+
127139
async function readFileRangeAsync(
128140
fileHandle: Awaited<ReturnType<typeof fs.promises.open>>,
129141
position: number,

‎src/config/sessions/transcript.test.ts‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@ import { appendSessionTranscriptMessage } from "./transcript-append.js";
88
import {
99
appendAssistantMessageToSessionTranscript,
1010
appendExactAssistantMessageToSessionTranscript,
11+
readLatestAssistantTextFromSessionTranscript,
1112
} from "./transcript.js";
1213

1314
describe("appendAssistantMessageToSessionTranscript", () => {
@@ -182,6 +183,49 @@ describe("appendAssistantMessageToSessionTranscript", () => {
182183
}
183184
});
184185

186+
it("dedupes against the latest assistant even when a large user entry follows it", async () => {
187+
writeTranscriptStore();
188+
189+
const exactResult = await appendExactAssistantMessageToSessionTranscript({
190+
sessionKey,
191+
storePath: fixture.storePath(),
192+
message: createExactAssistantMessage({ text: "Hello before the large user entry" }),
193+
});
194+
195+
expect(exactResult.ok).toBe(true);
196+
if (!exactResult.ok) {
197+
return;
198+
}
199+
200+
const sessionFile = resolveSessionTranscriptPathInDir(sessionId, fixture.sessionsDir());
201+
await appendSessionTranscriptMessage({
202+
transcriptPath: sessionFile,
203+
message: { role: "user", content: "x".repeat(5 * 1024 * 1024) },
204+
});
205+
206+
await expect(readLatestAssistantTextFromSessionTranscript(sessionFile)).resolves.toMatchObject({
207+
id: exactResult.messageId,
208+
text: "Hello before the large user entry",
209+
});
210+
211+
const mirrorResult = await appendAssistantMessageToSessionTranscript({
212+
sessionKey,
213+
text: "Hello before the large user entry",
214+
storePath: fixture.storePath(),
215+
});
216+
217+
expect(mirrorResult.ok).toBe(true);
218+
if (mirrorResult.ok) {
219+
expect(mirrorResult.messageId).toBe(exactResult.messageId);
220+
const records = fs
221+
.readFileSync(sessionFile, "utf-8")
222+
.trim()
223+
.split("\n")
224+
.map((line) => JSON.parse(line) as { type?: string; message?: { role?: string } });
225+
expect(records.filter((record) => record.type === "message")).toHaveLength(2);
226+
}
227+
});
228+
185229
it("does not reuse an older matching assistant message across turns", async () => {
186230
writeTranscriptStore();
187231

0 commit comments

Comments
 (0)