Skip to content

Commit ec59af3

Browse files
vincentkocvincentkoc
authored
fix(gateway): bound session transcript hot paths
Bound recent transcript reads and oversized injected-message writes across gateway session paths.\n\nThanks @vincentkoc
1 parent ea4d0a3 commit ec59af3

14 files changed

Lines changed: 725 additions & 91 deletions

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,7 @@ Docs: https://docs.openclaw.ai
7878
- Plugins/update: skip ClawHub and marketplace plugin updates when the bundled version is newer than the recorded installed version, so `openclaw update` no longer overwrites working bundled plugins with older external packages. Fixes #75447. Thanks @amknight.
7979
- Gateway/sessions: use bounded tail reads for sessions-list transcript usage fallbacks and cap bulk title/last-message hydration, keeping large session stores responsive when rows request derived previews. Thanks @vincentkoc.
8080
- Gateway/sessions: yield during bulk transcript title/preview hydration and copy compaction checkpoints asynchronously, keeping the Gateway event loop responsive for large session stores and large transcripts. Refs #75330 and #75414. Thanks @amknight.
81+
- Gateway/sessions: stream bounded transcript reads for session detail, history, artifacts, compaction, and send/subscribe sequence paths so small Gateway requests no longer materialize large transcripts or OOM on oversized session logs. Thanks @vincentkoc.
8182
- Gateway/chat: bound chat-history transcript reads to the requested display window so large session logs no longer OOM the Gateway when clients ask for a small history page. Thanks @vincentkoc.
8283
- Voice Call/Twilio: honor stored pre-connect TwiML before realtime webhook shortcuts and reject DTMF sequences outside conversation mode, so Meet PIN entry cannot be skipped or silently dropped. Thanks @donkeykong91 and @PfanP.
8384
- Docs/sandboxing: clarify that sandbox setup scripts (`sandbox-setup.sh`, `sandbox-common-setup.sh`, `sandbox-browser-setup.sh`) are only available from a source checkout, and add inline `docker build` commands for npm-installed users so sandbox image setup works without cloning the repo. Fixes #75485. Thanks @amknight.

‎src/gateway/server-methods/artifacts.test.ts‎

Lines changed: 30 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@ import { artifactsHandlers, collectArtifactsFromMessages } from "./artifacts.js"
44
const hoisted = vi.hoisted(() => ({
55
getTaskSessionLookupByIdForStatus: vi.fn(),
66
loadSessionEntry: vi.fn(),
7-
readSessionMessages: vi.fn(),
7+
visitSessionMessages: vi.fn(),
88
resolveSessionKeyForRun: vi.fn(),
99
}));
1010

@@ -17,7 +17,7 @@ vi.mock("../session-utils.js", async () => {
1717
return {
1818
...actual,
1919
loadSessionEntry: hoisted.loadSessionEntry,
20-
readSessionMessages: hoisted.readSessionMessages,
20+
visitSessionMessages: hoisted.visitSessionMessages,
2121
};
2222
});
2323

@@ -49,7 +49,7 @@ describe("artifacts RPC handlers", () => {
4949
storePath: "/tmp/sessions.json",
5050
entry: { sessionId: "sess-main", sessionFile: "/tmp/sess-main.jsonl" },
5151
});
52-
hoisted.readSessionMessages.mockReturnValue([
52+
mockedMessages([
5353
{
5454
role: "assistant",
5555
content: [
@@ -66,6 +66,15 @@ describe("artifacts RPC handlers", () => {
6666
]);
6767
});
6868

69+
function mockedMessages(messages: unknown[]) {
70+
hoisted.visitSessionMessages.mockImplementation(
71+
(_sessionId, _storePath, _sessionFile, visit) => {
72+
messages.forEach((message, index) => visit(message, index + 1));
73+
return messages.length;
74+
},
75+
);
76+
}
77+
6978
it("lists stable transcript artifact summaries by sessionKey", async () => {
7079
const { calls, respond } = createResponder();
7180

@@ -99,7 +108,21 @@ describe("artifacts RPC handlers", () => {
99108
it("gets and downloads an inline artifact", async () => {
100109
const listed = collectArtifactsFromMessages({
101110
sessionKey: "agent:main:main",
102-
messages: hoisted.readSessionMessages(),
111+
messages: [
112+
{
113+
role: "assistant",
114+
content: [
115+
{ type: "text", text: "see attached" },
116+
{
117+
type: "image",
118+
data: "aGVsbG8=",
119+
mimeType: "image/png",
120+
alt: "result.png",
121+
},
122+
],
123+
__openclaw: { seq: 2 },
124+
},
125+
],
103126
});
104127
const artifactId = listed[0]?.id;
105128
expect(artifactId).toBeTruthy();
@@ -137,7 +160,7 @@ describe("artifacts RPC handlers", () => {
137160

138161
it("resolves runId queries through the gateway run-to-session lookup", async () => {
139162
hoisted.resolveSessionKeyForRun.mockReturnValue("agent:main:main");
140-
hoisted.readSessionMessages.mockReturnValue([
163+
mockedMessages([
141164
{
142165
role: "assistant",
143166
content: [{ type: "image", data: "aGVsbG8=", alt: "run-result.png" }],
@@ -166,7 +189,7 @@ describe("artifacts RPC handlers", () => {
166189
requesterSessionKey: "agent:main:main",
167190
runId: "run-for-task-1",
168191
});
169-
hoisted.readSessionMessages.mockReturnValue([
192+
mockedMessages([
170193
{
171194
role: "assistant",
172195
content: [{ type: "image", data: "dGFyZ2V0", alt: "task-result.png" }],
@@ -257,7 +280,7 @@ describe("artifacts RPC handlers", () => {
257280
});
258281

259282
it("discovers transcript image_url data blocks", async () => {
260-
hoisted.readSessionMessages.mockReturnValue([
283+
mockedMessages([
261284
{
262285
role: "user",
263286
content: [

‎src/gateway/server-methods/artifacts.ts‎

Lines changed: 73 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -10,7 +10,7 @@ import {
1010
validateArtifactsListParams,
1111
} from "../protocol/index.js";
1212
import { resolveSessionKeyForRun } from "../server-session-key.js";
13-
import { loadSessionEntry, readSessionMessages } from "../session-utils.js";
13+
import { loadSessionEntry, visitSessionMessages } from "../session-utils.js";
1414
import type { GatewayRequestHandlers, RespondFn } from "./types.js";
1515
import { assertValidParams } from "./validation.js";
1616

@@ -215,59 +215,70 @@ export function collectArtifactsFromMessages(params: {
215215
const artifacts: ArtifactRecord[] = [];
216216
let messageFallbackSeq = 0;
217217
for (const message of params.messages) {
218-
const msg = asRecord(message);
219-
if (!msg) {
220-
continue;
221-
}
222218
messageFallbackSeq += 1;
223-
const messageSeq = resolveMessageSeq(msg, messageFallbackSeq);
224-
const messageRunId = resolveMessageRunId(msg);
225-
const messageTaskId = resolveMessageTaskId(msg);
226-
if (params.runId && messageRunId !== params.runId) {
227-
continue;
228-
}
229-
if (params.taskId && messageTaskId !== params.taskId) {
219+
collectArtifactsFromMessage({ ...params, message, messageFallbackSeq, artifacts });
220+
}
221+
return artifacts;
222+
}
223+
224+
function collectArtifactsFromMessage(params: {
225+
message: unknown;
226+
messageFallbackSeq: number;
227+
artifacts: ArtifactRecord[];
228+
sessionKey: string;
229+
runId?: string;
230+
taskId?: string;
231+
}): void {
232+
const msg = asRecord(params.message);
233+
if (!msg) {
234+
return;
235+
}
236+
const messageSeq = resolveMessageSeq(msg, params.messageFallbackSeq);
237+
const messageRunId = resolveMessageRunId(msg);
238+
const messageTaskId = resolveMessageTaskId(msg);
239+
if (params.runId && messageRunId !== params.runId) {
240+
return;
241+
}
242+
if (params.taskId && messageTaskId !== params.taskId) {
243+
return;
244+
}
245+
const content = Array.isArray(msg.content) ? msg.content : [];
246+
for (let contentIndex = 0; contentIndex < content.length; contentIndex += 1) {
247+
const block = asRecord(content[contentIndex]);
248+
if (!block || !isArtifactBlock(block)) {
230249
continue;
231250
}
232-
const content = Array.isArray(msg.content) ? msg.content : [];
233-
for (let contentIndex = 0; contentIndex < content.length; contentIndex += 1) {
234-
const block = asRecord(content[contentIndex]);
235-
if (!block || !isArtifactBlock(block)) {
236-
continue;
237-
}
238-
const type = normalizeArtifactType(asNonEmptyString(block.type) ?? "file");
239-
const title =
240-
asNonEmptyString(block.title) ??
241-
asNonEmptyString(block.fileName) ??
242-
asNonEmptyString(block.filename) ??
243-
asNonEmptyString(block.alt) ??
244-
`${type} ${artifacts.length + 1}`;
245-
const download = resolveBlockDownload(block);
246-
const summary: ArtifactRecord = {
247-
id: artifactId({
248-
sessionKey: params.sessionKey,
249-
messageSeq,
250-
contentIndex,
251-
title,
252-
type,
253-
}),
254-
type,
255-
title,
256-
...(download.mimeType ? { mimeType: download.mimeType } : {}),
257-
...(download.sizeBytes !== undefined ? { sizeBytes: download.sizeBytes } : {}),
251+
const type = normalizeArtifactType(asNonEmptyString(block.type) ?? "file");
252+
const title =
253+
asNonEmptyString(block.title) ??
254+
asNonEmptyString(block.fileName) ??
255+
asNonEmptyString(block.filename) ??
256+
asNonEmptyString(block.alt) ??
257+
`${type} ${params.artifacts.length + 1}`;
258+
const download = resolveBlockDownload(block);
259+
const summary: ArtifactRecord = {
260+
id: artifactId({
258261
sessionKey: params.sessionKey,
259-
...(messageRunId ? { runId: messageRunId } : {}),
260-
...(messageTaskId ? { taskId: messageTaskId } : {}),
261262
messageSeq,
262-
source: "session-transcript",
263-
download: { mode: download.mode },
264-
...(download.data ? { data: download.data } : {}),
265-
...(download.url ? { url: download.url } : {}),
266-
};
267-
artifacts.push(summary);
268-
}
263+
contentIndex,
264+
title,
265+
type,
266+
}),
267+
type,
268+
title,
269+
...(download.mimeType ? { mimeType: download.mimeType } : {}),
270+
...(download.sizeBytes !== undefined ? { sizeBytes: download.sizeBytes } : {}),
271+
sessionKey: params.sessionKey,
272+
...(messageRunId ? { runId: messageRunId } : {}),
273+
...(messageTaskId ? { taskId: messageTaskId } : {}),
274+
messageSeq,
275+
source: "session-transcript",
276+
download: { mode: download.mode },
277+
...(download.data ? { data: download.data } : {}),
278+
...(download.url ? { url: download.url } : {}),
279+
};
280+
params.artifacts.push(summary);
269281
}
270-
return artifacts;
271282
}
272283

273284
function resolveQuerySessionKey(query: ArtifactQuery): string | undefined {
@@ -296,16 +307,23 @@ function loadArtifacts(query: ArtifactQuery): { artifacts: ArtifactRecord[]; ses
296307
}
297308
const { storePath, entry } = loadSessionEntry(sessionKey);
298309
const sessionId = entry?.sessionId;
299-
const messages =
300-
sessionId && storePath ? readSessionMessages(sessionId, storePath, entry?.sessionFile) : [];
301-
return {
302-
sessionKey,
303-
artifacts: collectArtifactsFromMessages({
304-
messages,
310+
if (!sessionId || !storePath) {
311+
return { sessionKey, artifacts: [] };
312+
}
313+
const artifacts: ArtifactRecord[] = [];
314+
visitSessionMessages(sessionId, storePath, entry?.sessionFile, (message, seq) => {
315+
collectArtifactsFromMessage({
316+
message,
317+
messageFallbackSeq: seq,
318+
artifacts,
305319
sessionKey,
306320
runId: query.runId,
307321
taskId: query.taskId,
308-
}),
322+
});
323+
});
324+
return {
325+
sessionKey,
326+
artifacts,
309327
};
310328
}
311329

‎src/gateway/server-methods/chat-transcript-inject.ts‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,14 @@
1+
import { randomUUID } from "node:crypto";
2+
import fs from "node:fs";
3+
import { StringDecoder } from "node:string_decoder";
14
import { SessionManager } from "@mariozechner/pi-coding-agent";
25
import { formatErrorMessage } from "../../infra/errors.js";
36
import { emitSessionTranscriptUpdate } from "../../sessions/transcript-events.js";
47

58
type AppendMessageArg = Parameters<SessionManager["appendMessage"]>[0];
69

10+
const SESSION_MANAGER_APPEND_MAX_BYTES = 8 * 1024 * 1024;
11+
712
export type GatewayInjectedAbortMeta = {
813
aborted: true;
914
origin: "rpc" | "stop-command";
@@ -41,6 +46,77 @@ function resolveInjectedAssistantContent(params: {
4146
return [{ type: "text", text: `${labelPrefix}${params.message}` }];
4247
}
4348

49+
function transcriptHasParentLinkedEntries(transcriptPath: string): boolean {
50+
let fd: number | null = null;
51+
try {
52+
fd = fs.openSync(transcriptPath, "r");
53+
const decoder = new StringDecoder("utf8");
54+
const buffer = Buffer.allocUnsafe(64 * 1024);
55+
let carry = "";
56+
while (true) {
57+
const bytesRead = fs.readSync(fd, buffer, 0, buffer.length, null);
58+
if (bytesRead <= 0) {
59+
break;
60+
}
61+
const text = carry + decoder.write(buffer.subarray(0, bytesRead));
62+
const lines = text.split(/\r?\n/);
63+
carry = lines.pop() ?? "";
64+
for (const line of lines) {
65+
if (lineHasParentLinkedEntry(line)) {
66+
return true;
67+
}
68+
}
69+
}
70+
return lineHasParentLinkedEntry(carry + decoder.end());
71+
} catch {
72+
return true;
73+
} finally {
74+
if (fd !== null) {
75+
fs.closeSync(fd);
76+
}
77+
}
78+
}
79+
80+
function lineHasParentLinkedEntry(line: string): boolean {
81+
if (!line.trim()) {
82+
return false;
83+
}
84+
try {
85+
const parsed = JSON.parse(line) as { type?: unknown; id?: unknown; parentId?: unknown };
86+
return parsed.type !== "session" && typeof parsed.id === "string" && "parentId" in parsed;
87+
} catch {
88+
return false;
89+
}
90+
}
91+
92+
function shouldUseRawAppend(transcriptPath: string): boolean {
93+
try {
94+
const stat = fs.statSync(transcriptPath);
95+
return (
96+
stat.size > SESSION_MANAGER_APPEND_MAX_BYTES &&
97+
!transcriptHasParentLinkedEntries(transcriptPath)
98+
);
99+
} catch {
100+
return false;
101+
}
102+
}
103+
104+
function appendRawAssistantMessageToTranscript(params: {
105+
transcriptPath: string;
106+
message: AppendMessageArg & Record<string, unknown>;
107+
now: number;
108+
}): { messageId: string } {
109+
const messageId = randomUUID();
110+
const entry = {
111+
type: "message",
112+
id: messageId,
113+
timestamp: new Date(params.now).toISOString(),
114+
message: params.message,
115+
};
116+
fs.appendFileSync(params.transcriptPath, `${JSON.stringify(entry)}\n`, "utf-8");
117+
return { messageId };
118+
}
119+
44120
export function appendInjectedAssistantMessageToTranscript(params: {
45121
transcriptPath: string;
46122
message: string;
@@ -100,6 +176,20 @@ export function appendInjectedAssistantMessageToTranscript(params: {
100176
};
101177

102178
try {
179+
if (shouldUseRawAppend(params.transcriptPath)) {
180+
const { messageId } = appendRawAssistantMessageToTranscript({
181+
transcriptPath: params.transcriptPath,
182+
message: messageBody,
183+
now,
184+
});
185+
emitSessionTranscriptUpdate({
186+
sessionFile: params.transcriptPath,
187+
message: messageBody,
188+
messageId,
189+
});
190+
return { ok: true, messageId, message: messageBody };
191+
}
192+
103193
// IMPORTANT: Use SessionManager so the entry is attached to the current leaf via parentId.
104194
// Raw jsonl appends break the parent chain and can hide compaction summaries from context.
105195
const sessionManager = SessionManager.open(params.transcriptPath);

0 commit comments

Comments
 (0)