ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-26 04:15:48 -06:00
Code Issues Packages Projects Releases Wiki Activity

refactor(gateway): unify bounded chat history across transports (#128844)

Browse Source
This commit is contained in:
Peter Steinberger
2026-08-24 13:06:06 -07:00
committed by GitHub
parent 9e98229678
commit a99317ef97
14 changed files with 455 additions and 975 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/assertion-safety-baseline.txt
+1 -1
View File
@@ -2110,7 +2110,7 @@ src/agents/tools/cron-tool-schema.ts 1
src/agents/tools/cron-tool-write.ts 2
src/agents/tools/cron-tool.ts 13
src/agents/tools/dashboard-tool.ts 3
src/agents/tools/embedded-gateway-stub.ts 13
src/agents/tools/embedded-gateway-stub.ts 8
src/agents/tools/gateway-tool.ts 5
src/agents/tools/gateway.ts 1
src/agents/tools/goal-tools.ts 4
src/agents/tools/embedded-gateway-stub.runtime.ts
+6 -13
View File
@@ -11,25 +11,18 @@ export {
resolveSessionStoreKey,
resolveStoredSessionKeyForAgentStore,
} from "../../gateway/session-store-key.js";
export {
dropPreSessionStartAnnouncePairs,
projectChatDisplayMessages,
projectRecentChatDisplayMessages,
resolveEffectiveChatHistoryMaxChars,
} from "../../gateway/chat-display-projection.js";
export { augmentChatHistoryWithCliSessionImports } from "../../gateway/cli-session-history.js";
export { resolveEffectiveChatHistoryMaxChars } from "../../gateway/chat-display-projection.js";
export { getMaxChatHistoryMessagesBytes } from "../../gateway/server-constants.js";
export {
augmentChatHistoryWithCanvasBlocks,
CHAT_HISTORY_MAX_SINGLE_MESSAGE_BYTES,
replaceOversizedChatHistoryMessages,
} from "../../gateway/server-methods/chat.js";
export {
capArrayByJsonBytes,
readRecentSessionMessagesWithStatsAsync,
readSessionMessagesPageWithStatsAsync,
readSessionMessagesAsync,
} from "../../gateway/session-transcript-readers.js";
readChatHistoryPage,
resolveChatHistoryNextOffset,
shouldReplayOldestChatHistoryRecord,
} from "../../gateway/server-methods/chat-history-pages.js";
export { capArrayByJsonBytes } from "../../gateway/session-transcript-readers.js";
export {
listSessionsFromStoreAsync,
loadCombinedSessionStoreForGatewayCore,
src/agents/tools/embedded-gateway-stub.test.ts
+108 -386
View File
@@ -24,25 +24,19 @@ const runtime = vi.hoisted(() => ({
cfg: {},
storePath: "/tmp/openclaw-sessions.json",
entry: { sessionId: "sess-main" },
canonicalKey: "agent:main:main",
})),
resolveSessionModelRef: vi.fn(() => ({ provider: "openai" })),
readSessionMessagesAsync: vi.fn(async (): Promise<unknown[]> => []),
readRecentSessionMessagesWithStatsAsync: vi.fn(async () => ({
readChatHistoryPage: vi.fn(async () => ({
messages: [] as unknown[],
totalMessages: 0,
pagination: { offset: 0, totalMessages: 0, rawPageMessages: 0 },
})),
readSessionMessagesPageWithStatsAsync: vi.fn(async () => ({
messages: [] as unknown[],
totalMessages: 0,
})),
augmentChatHistoryWithCliSessionImports: vi.fn(
({ localMessages }: { localMessages?: unknown[] }) => localMessages ?? [],
resolveChatHistoryNextOffset: vi.fn(
({ offset, rawPageMessages }: { offset: number; rawPageMessages: number }) =>
offset + rawPageMessages,
),
shouldReplayOldestChatHistoryRecord: vi.fn(() => false),
resolveEffectiveChatHistoryMaxChars: vi.fn(() => 100_000),
dropPreSessionStartAnnouncePairs: vi.fn((messages: unknown[]) => messages),
projectChatDisplayMessages: vi.fn((messages: unknown[]): unknown[] => messages),
projectRecentChatDisplayMessages: vi.fn((messages: unknown[]): unknown[] => messages),
augmentChatHistoryWithCanvasBlocks: vi.fn((messages: unknown[]) => messages),
getMaxChatHistoryMessagesBytes: vi.fn(() => 100_000),
CHAT_HISTORY_MAX_SINGLE_MESSAGE_BYTES: 100_000,
replaceOversizedChatHistoryMessages: vi.fn(({ messages }: { messages: unknown[] }) => ({
@@ -62,13 +56,9 @@ describe("embedded gateway stub", () => {
beforeEach(() => {
runtime.getRuntimeConfig.mockClear();
runtime.resolveSessionKeyFromResolveParams.mockReset();
runtime.augmentChatHistoryWithCliSessionImports.mockClear();
runtime.projectChatDisplayMessages.mockClear();
runtime.projectRecentChatDisplayMessages.mockClear();
runtime.dropPreSessionStartAnnouncePairs.mockClear();
runtime.readSessionMessagesAsync.mockClear();
runtime.readRecentSessionMessagesWithStatsAsync.mockClear();
runtime.readSessionMessagesPageWithStatsAsync.mockClear();
runtime.readChatHistoryPage.mockClear();
runtime.resolveChatHistoryNextOffset.mockClear();
runtime.shouldReplayOldestChatHistoryRecord.mockClear();
runtime.loadSessionEntry.mockClear();
runtime.resolveSessionAgentId.mockClear();
runtime.resolveSessionStoreKey.mockClear();
@@ -221,408 +211,140 @@ describe("embedded gateway stub", () => {
expect(runtime.searchSessionTranscripts).not.toHaveBeenCalled();
});
it("projects embedded chat history through the shared display projector", async () => {
// Embedded history must use the same projection path as gateway history so
// byte/message limits and display filtering stay aligned.
const rawMessages = [
{ role: "user", content: "hello" },
{ role: "assistant", content: "hi" },
];
const projectedMessages = [{ role: "assistant", content: "hi" }];
runtime.readSessionMessagesAsync.mockImplementationOnce(async () => rawMessages);
runtime.projectRecentChatDisplayMessages.mockReturnValueOnce(projectedMessages);
const callGateway = createEmbeddedCallGateway();
const result = await callGateway<{ messages: unknown[] }>({
method: "chat.history",
params: { sessionKey: "agent:main:main" },
it("reads embedded history through the canonical Gateway history owner", async () => {
const messages = [{ role: "assistant", content: "visible past a silent tail" }];
runtime.readChatHistoryPage.mockResolvedValueOnce({
messages,
pagination: { offset: 0, totalMessages: 81, rawPageMessages: 81 },
});
expect(runtime.projectRecentChatDisplayMessages).toHaveBeenCalledWith(rawMessages, {
includeCommentaryFallbacks: true,
maxChars: 100_000,
maxMessages: 200,
});
expect(runtime.readSessionMessagesAsync).toHaveBeenCalledWith(
{
agentId: "main",
sessionEntry: { sessionId: "sess-main" },
sessionId: "sess-main",
sessionKey: "agent:main:main",
storePath: "/tmp/openclaw-sessions.json",
},
{
mode: "recent",
maxMessages: 200,
maxBytes: 1024 * 1024,
allowResetArchiveFallback: true,
},
);
expect(result.messages).toEqual(projectedMessages);
});
it("scopes embedded global chat history to the requested agent", async () => {
const callGateway = createEmbeddedCallGateway();
await callGateway<{ messages: unknown[] }>({
method: "chat.history",
params: { sessionKey: "global", agentId: "work" },
});
expect(runtime.loadSessionEntry).toHaveBeenCalledWith("global", { agentId: "work" });
expect(runtime.resolveSessionAgentId).toHaveBeenCalledWith({
sessionKey: "global",
config: {},
agentId: "work",
});
});
it("infers embedded global chat history scope from agent-prefixed aliases", async () => {
// Agent-prefixed global aliases carry the target agent id even when the
// caller does not pass agentId separately.
const callGateway = createEmbeddedCallGateway();
await callGateway<{ messages: unknown[] }>({
method: "chat.history",
params: { sessionKey: "agent:work:main" },
});
expect(runtime.loadSessionEntry).toHaveBeenCalledWith("agent:work:main", { agentId: "work" });
expect(runtime.resolveSessionAgentId).toHaveBeenCalledWith({
sessionKey: "agent:work:main",
config: {},
agentId: "work",
});
});
it("passes the requested recent history window to projection", async () => {
const rawMessages = [
{ role: "user", content: "visible older" },
{ role: "assistant", content: "hidden newer" },
];
runtime.readSessionMessagesAsync.mockImplementationOnce(async () => rawMessages);
const callGateway = createEmbeddedCallGateway();
await callGateway<{ messages: unknown[] }>({
const result = await createEmbeddedCallGateway()<{ messages: unknown[] }>({
method: "chat.history",
params: { sessionKey: "agent:main:main", limit: 1 },
});
expect(runtime.projectRecentChatDisplayMessages).toHaveBeenCalledWith(rawMessages, {
includeCommentaryFallbacks: true,
maxChars: 100_000,
maxMessages: 1,
});
expect(runtime.readSessionMessagesAsync).toHaveBeenCalledWith(
{
agentId: "main",
sessionEntry: { sessionId: "sess-main" },
sessionId: "sess-main",
sessionKey: "agent:main:main",
storePath: "/tmp/openclaw-sessions.json",
},
{
mode: "recent",
maxMessages: 1,
maxBytes: 1024 * 1024,
allowResetArchiveFallback: true,
},
);
});
it("uses a bounded page read for offset chat history pages", async () => {
const rawMessages = [
{ role: "user", content: "oldest" },
{ role: "assistant", content: "older" },
{ role: "user", content: "newer" },
{ role: "assistant", content: "latest" },
];
runtime.readSessionMessagesPageWithStatsAsync.mockImplementationOnce(async () => ({
messages: rawMessages.slice(0, 2),
totalMessages: rawMessages.length,
}));
const callGateway = createEmbeddedCallGateway();
const result = await callGateway<{
messages: unknown[];
offset?: number;
nextOffset?: number;
hasMore?: boolean;
totalMessages?: number;
}>({
method: "chat.history",
params: { sessionKey: "agent:main:main", limit: 2, offset: 2 },
});
expect(runtime.readSessionMessagesAsync).not.toHaveBeenCalled();
expect(runtime.readSessionMessagesPageWithStatsAsync).toHaveBeenCalledWith(
{
agentId: "main",
sessionEntry: { sessionId: "sess-main" },
sessionId: "sess-main",
sessionKey: "agent:main:main",
storePath: "/tmp/openclaw-sessions.json",
},
{
offset: 2,
maxMessages: 3,
allowResetArchiveFallback: true,
},
);
expect(runtime.projectChatDisplayMessages).toHaveBeenCalledWith(rawMessages.slice(0, 2), {
includeCommentaryFallbacks: true,
maxChars: 100_000,
});
expect(result).toMatchObject({
messages: rawMessages.slice(0, 2),
offset: 2,
hasMore: false,
totalMessages: 4,
});
expect(result.nextOffset).toBeUndefined();
});
it("caps projected offset chat history pages to the requested limit", async () => {
const rawMessages = [
{ role: "assistant", content: "overread", __openclaw: { seq: 1 } },
{ role: "assistant", content: "page anchor", __openclaw: { seq: 2 } },
];
const projectedMessages = [
{ role: "assistant", content: "projected one", __openclaw: { seq: 2 } },
{ role: "assistant", content: "projected two", __openclaw: { seq: 3 } },
];
runtime.readSessionMessagesPageWithStatsAsync.mockImplementationOnce(async () => ({
messages: rawMessages,
totalMessages: 4,
}));
runtime.projectChatDisplayMessages.mockReturnValueOnce(projectedMessages);
const callGateway = createEmbeddedCallGateway();
const result = await callGateway<{
messages: unknown[];
nextOffset?: number;
hasMore?: boolean;
}>({
method: "chat.history",
params: { sessionKey: "agent:main:main", limit: 1, offset: 1 },
});
expect(runtime.projectChatDisplayMessages).toHaveBeenCalledWith([rawMessages[1]], {
includeCommentaryFallbacks: true,
maxChars: 100_000,
});
expect(result.messages).toEqual([projectedMessages[1]]);
expect(result.nextOffset).toBe(2);
expect(result.hasMore).toBe(true);
});
it("filters offset chat history pages at the session start boundary", async () => {
const rawMessages = [
{ role: "user", content: "stale announce", __openclaw: { seq: 1 } },
{ role: "assistant", content: "stale reply", __openclaw: { seq: 2 } },
];
const filteredMessages: unknown[] = [];
runtime.loadSessionEntry.mockReturnValueOnce({
cfg: {},
expect(runtime.readChatHistoryPage).toHaveBeenCalledWith({
entry: { sessionId: "sess-main" },
provider: "openai",
sessionId: "sess-main",
storePath: "/tmp/openclaw-sessions.json",
entry: { sessionId: "sess-main", sessionStartedAt: 1234 } as {
sessionId: string;
sessionStartedAt: number;
},
sessionAgentId: "main",
canonicalKey: "agent:main:main",
max: 1,
maxHistoryBytes: 100_000,
effectiveMaxChars: 100_000,
offset: undefined,
messageId: undefined,
});
runtime.readSessionMessagesPageWithStatsAsync.mockImplementationOnce(async () => ({
messages: rawMessages,
totalMessages: 2,
}));
runtime.dropPreSessionStartAnnouncePairs.mockReturnValueOnce(filteredMessages);
const callGateway = createEmbeddedCallGateway();
const result = await callGateway<{ messages: unknown[] }>({
method: "chat.history",
params: { sessionKey: "agent:main:main", limit: 1, offset: 1 },
});
expect(runtime.dropPreSessionStartAnnouncePairs).toHaveBeenCalledWith(rawMessages, 1234);
expect(runtime.projectChatDisplayMessages).toHaveBeenCalledWith(filteredMessages, {
includeCommentaryFallbacks: true,
maxChars: 100_000,
});
expect(result.messages).toEqual(filteredMessages);
expect(result.messages).toEqual(messages);
expect(result).not.toHaveProperty("offset");
});
it("does not merge full CLI imports into explicit offset chat history pages", async () => {
const rawMessages = [{ role: "assistant", content: "local page", __openclaw: { seq: 2 } }];
runtime.readSessionMessagesPageWithStatsAsync.mockImplementationOnce(async () => ({
messages: rawMessages,
totalMessages: 2,
}));
const callGateway = createEmbeddedCallGateway();
const result = await callGateway<{ messages: unknown[] }>({
it.each([
{ sessionKey: "global", agentId: "work" },
{ sessionKey: "agent:work:main", agentId: undefined },
])("scopes embedded chat history to its requested agent", async ({ sessionKey, agentId }) => {
await createEmbeddedCallGateway()({
method: "chat.history",
params: { sessionKey: "agent:main:main", limit: 1, offset: 1 },
params: { sessionKey, ...(agentId ? { agentId } : {}) },
});
expect(runtime.augmentChatHistoryWithCliSessionImports).not.toHaveBeenCalled();
expect(result.messages).toEqual(rawMessages);
expect(runtime.loadSessionEntry).toHaveBeenCalledWith(sessionKey, { agentId: "work" });
expect(runtime.resolveSessionAgentId).toHaveBeenCalledWith({
sessionKey,
config: {},
agentId: "work",
});
});
it("overreads bounded recent history for the first offset page", async () => {
const rawMessages = [
{ role: "user", content: "visible older", __openclaw: { seq: 6 } },
{ role: "assistant", content: "hidden control", __openclaw: { seq: 7 } },
{ role: "assistant", content: "visible latest", __openclaw: { seq: 8 } },
];
const projectedMessages = [rawMessages[0], rawMessages[2]];
runtime.readRecentSessionMessagesWithStatsAsync.mockImplementationOnce(async () => ({
messages: rawMessages,
totalMessages: 10,
}));
runtime.projectRecentChatDisplayMessages.mockReturnValueOnce(projectedMessages);
it("preserves bounded offset metadata from the shared visible-history scanner", async () => {
const messages = [{ role: "assistant", content: "older visible", __openclaw: { seq: 2 } }];
runtime.readChatHistoryPage.mockResolvedValueOnce({
messages,
pagination: { offset: 1, totalMessages: 82, rawPageMessages: 80 },
});
const callGateway = createEmbeddedCallGateway();
const result = await callGateway<{
const result = await createEmbeddedCallGateway()<{
messages: unknown[];
offset?: number;
nextOffset?: number;
hasMore?: boolean;
totalMessages?: number;
offset: number;
nextOffset: number;
hasMore: boolean;
totalMessages: number;
}>({
method: "chat.history",
params: { sessionKey: "agent:main:main", limit: 1, offset: 1 },
});
expect(runtime.readChatHistoryPage).toHaveBeenCalledWith(
expect.objectContaining({ offset: 1, max: 1 }),
);
expect(result).toMatchObject({
messages,
offset: 1,
nextOffset: 81,
hasMore: true,
totalMessages: 82,
});
});
it("computes continuation from the final byte-budgeted visible page", async () => {
const messages = [
{ role: "assistant", content: "older", __openclaw: { seq: 6 } },
{ role: "assistant", content: "latest", __openclaw: { seq: 7 } },
];
const bounded = [messages[1]];
runtime.readChatHistoryPage.mockResolvedValueOnce({
messages,
pagination: { offset: 0, totalMessages: 10, rawPageMessages: 5 },
});
runtime.capArrayByJsonBytes.mockReturnValueOnce({ items: bounded });
runtime.shouldReplayOldestChatHistoryRecord.mockReturnValueOnce(true);
runtime.resolveChatHistoryNextOffset.mockReturnValueOnce(3);
const result = await createEmbeddedCallGateway()({
method: "chat.history",
params: { sessionKey: "agent:main:main", limit: 2, offset: 0 },
});
expect(runtime.readSessionMessagesAsync).not.toHaveBeenCalled();
expect(runtime.readRecentSessionMessagesWithStatsAsync).toHaveBeenCalledWith(
{
agentId: "main",
sessionEntry: { sessionId: "sess-main" },
sessionId: "sess-main",
sessionKey: "agent:main:main",
storePath: "/tmp/openclaw-sessions.json",
},
{
maxMessages: 61,
maxBytes: 1024 * 1024,
allowResetArchiveFallback: true,
},
);
expect(runtime.projectRecentChatDisplayMessages).toHaveBeenCalledWith(rawMessages, {
includeCommentaryFallbacks: true,
maxChars: 100_000,
maxMessages: 2,
});
expect(result).toMatchObject({
messages: projectedMessages,
expect(runtime.resolveChatHistoryNextOffset).toHaveBeenCalledWith({
messages: bounded,
totalMessages: 10,
offset: 0,
nextOffset: 5,
hasMore: true,
totalMessages: 10,
rawPageMessages: 5,
replayOldestRecord: true,
});
expect(result).toMatchObject({ messages: bounded, nextOffset: 3, hasMore: true });
});
it("computes offset continuation from the final budgeted chat history page", async () => {
const rawMessages = [
{ role: "user", content: "visible older", __openclaw: { seq: 6 } },
{ role: "assistant", content: "visible newer", __openclaw: { seq: 7 } },
{ role: "assistant", content: "visible latest", __openclaw: { seq: 8 } },
];
const returnedMessages = [rawMessages[2]];
runtime.readRecentSessionMessagesWithStatsAsync.mockImplementationOnce(async () => ({
messages: rawMessages,
totalMessages: 10,
}));
runtime.capArrayByJsonBytes.mockReturnValueOnce({ items: returnedMessages });
const callGateway = createEmbeddedCallGateway();
const result = await callGateway<{
messages: unknown[];
nextOffset?: number;
hasMore?: boolean;
}>({
method: "chat.history",
params: { sessionKey: "agent:main:main", limit: 3, offset: 0 },
});
expect(result.messages).toEqual(returnedMessages);
expect(result.nextOffset).toBe(3);
expect(result.hasMore).toBe(true);
});
it("normalizes string chat history limits before projection", async () => {
const rawMessages = [
{ role: "user", content: "older" },
{ role: "assistant", content: "newer" },
];
runtime.readSessionMessagesAsync.mockResolvedValueOnce(rawMessages);
const callGateway = createEmbeddedCallGateway();
await callGateway<{ messages: unknown[] }>({
it("normalizes string history limits before calling the shared owner", async () => {
await createEmbeddedCallGateway()({
method: "chat.history",
params: { sessionKey: "agent:main:main", limit: "2" },
});
expect(runtime.projectRecentChatDisplayMessages).toHaveBeenCalledWith(rawMessages, {
includeCommentaryFallbacks: true,
maxChars: 100_000,
maxMessages: 2,
});
expect(runtime.readSessionMessagesAsync).toHaveBeenCalledWith(
{
agentId: "main",
sessionEntry: { sessionId: "sess-main" },
sessionId: "sess-main",
sessionKey: "agent:main:main",
storePath: "/tmp/openclaw-sessions.json",
},
{
mode: "recent",
maxMessages: 2,
maxBytes: 1024 * 1024,
allowResetArchiveFallback: true,
},
);
expect(runtime.readChatHistoryPage).toHaveBeenCalledWith(expect.objectContaining({ max: 2 }));
});
it("rejects malformed chat history limits before reading session files", async () => {
const callGateway = createEmbeddedCallGateway();
it.each(["2.5", -1])("rejects malformed history limit %j before reading", async (limit) => {
await expect(
callGateway({
createEmbeddedCallGateway()({
method: "chat.history",
params: { sessionKey: "agent:main:main", limit: "2.5" },
params: { sessionKey: "agent:main:main", limit },
}),
).rejects.toThrow("limit must be a positive integer");
await expect(
callGateway({
method: "chat.history",
params: { sessionKey: "agent:main:main", limit: -1 },
}),
).rejects.toThrow("limit must be a positive integer");
expect(runtime.readSessionMessagesAsync).not.toHaveBeenCalled();
expect(runtime.readChatHistoryPage).not.toHaveBeenCalled();
});
it("rejects malformed chat history offsets before reading session files", async () => {
const callGateway = createEmbeddedCallGateway();
await expect(
callGateway({
method: "chat.history",
params: { sessionKey: "agent:main:main", offset: -1 },
}),
).rejects.toThrow("offset must be a non-negative integer");
await expect(
callGateway({
method: "chat.history",
params: { sessionKey: "agent:main:main", offset: 1.5 },
}),
).rejects.toThrow("offset must be a non-negative integer");
await expect(
callGateway({
method: "chat.history",
params: { sessionKey: "agent:main:main", offset: "1abc" },
}),
).rejects.toThrow("offset must be a non-negative integer");
expect(runtime.readSessionMessagesAsync).not.toHaveBeenCalled();
expect(runtime.readRecentSessionMessagesWithStatsAsync).not.toHaveBeenCalled();
expect(runtime.readSessionMessagesPageWithStatsAsync).not.toHaveBeenCalled();
});
it.each([-1, 1.5, "1abc"])(
"rejects malformed history offset %j before reading",
async (offset) => {
await expect(
createEmbeddedCallGateway()({
method: "chat.history",
params: { sessionKey: "agent:main:main", offset },
}),
).rejects.toThrow("offset must be a non-negative integer");
expect(runtime.readChatHistoryPage).not.toHaveBeenCalled();
},
);
});
src/agents/tools/embedded-gateway-stub.ts
+47 -232
View File
@@ -3,7 +3,6 @@
*
* Implements only the Gateway calls needed by session tools and rejects unsupported methods.
*/
import { asPositiveSafeInteger } from "@openclaw/normalization-core/number-coercion";
import { normalizeFastMode, type FastMode } from "@openclaw/normalization-core/string-coerce";
import type {
SessionsListParams,
@@ -12,9 +11,10 @@ import type {
import type { OpenClawConfig } from "../../config/types.openclaw.js";
import type { CallGatewayOptions } from "../../gateway/call.js";
import type {
ReadSessionMessagesAsyncOptions,
SessionTranscriptReadScope,
} from "../../gateway/session-transcript-readers.js";
readChatHistoryPage,
resolveChatHistoryNextOffset,
shouldReplayOldestChatHistoryRecord,
} from "../../gateway/server-methods/chat-history-pages.js";
import type { SessionsListResult } from "../../gateway/session-utils.types.js";
import type { SessionsResolveResult } from "../../gateway/sessions-resolve.js";
import { parseAgentSessionKey } from "../../routing/session-key.js";
@@ -48,31 +48,13 @@ interface EmbeddedGatewayRuntime {
indexing: boolean;
truncated: boolean;
};
augmentChatHistoryWithCliSessionImports: (opts: {
entry: unknown;
provider: string | undefined;
localMessages: unknown[];
}) => unknown[];
getMaxChatHistoryMessagesBytes: () => number;
augmentChatHistoryWithCanvasBlocks: (msgs: unknown[]) => unknown[];
CHAT_HISTORY_MAX_SINGLE_MESSAGE_BYTES: number;
replaceOversizedChatHistoryMessages: (opts: {
messages: unknown[];
maxSingleMessageBytes: number;
}) => { messages: unknown[] };
resolveEffectiveChatHistoryMaxChars: (cfg: OpenClawConfig) => number;
dropPreSessionStartAnnouncePairs: (
messages: unknown[],
sessionStartedAt: number | undefined,
) => unknown[];
projectChatDisplayMessages: (
msgs: unknown[],
opts?: { includeCommentaryFallbacks?: boolean; maxChars?: number },
) => unknown[];
projectRecentChatDisplayMessages: (
msgs: unknown[],
opts?: { includeCommentaryFallbacks?: boolean; maxChars?: number; maxMessages?: number },
) => unknown[];
capArrayByJsonBytes: (items: unknown[], maxBytes: number) => { items: unknown[] };
listSessionsFromStoreAsync: (opts: {
cfg: OpenClawConfig;
@@ -98,20 +80,12 @@ interface EmbeddedGatewayRuntime {
) => {
cfg: OpenClawConfig;
storePath: string | undefined;
entry: Record<string, unknown> | undefined;
entry: Parameters<typeof readChatHistoryPage>[0]["entry"];
canonicalKey: string;
};
readSessionMessagesAsync: (
scope: SessionTranscriptReadScope,
opts: ReadSessionMessagesAsyncOptions,
) => Promise<unknown[]>;
readRecentSessionMessagesWithStatsAsync: (
scope: SessionTranscriptReadScope,
opts: { maxMessages: number; maxBytes?: number; allowResetArchiveFallback?: boolean },
) => Promise<{ messages: unknown[]; totalMessages: number }>;
readSessionMessagesPageWithStatsAsync: (
scope: SessionTranscriptReadScope,
opts: { offset: number; maxMessages: number; allowResetArchiveFallback?: boolean },
) => Promise<{ messages: unknown[]; totalMessages: number }>;
readChatHistoryPage: typeof readChatHistoryPage;
resolveChatHistoryNextOffset: typeof resolveChatHistoryNextOffset;
shouldReplayOldestChatHistoryRecord: typeof shouldReplayOldestChatHistoryRecord;
resolveSessionModelRef: (
cfg: OpenClawConfig,
entry: unknown,
@@ -137,65 +111,6 @@ function readOffsetParam(params: Record<string, unknown>): number | undefined {
return offset;
}
function readChatHistoryMessageSeq(message: unknown): number | undefined {
if (!message || typeof message !== "object" || Array.isArray(message)) {
return undefined;
}
const metadata = (message as Record<string, unknown>)["__openclaw"];
if (!metadata || typeof metadata !== "object" || Array.isArray(metadata)) {
return undefined;
}
const seq = (metadata as Record<string, unknown>).seq;
return asPositiveSafeInteger(seq);
}
function resolveChatHistoryNextOffset(params: {
messages: unknown[];
totalMessages: number;
offset: number;
rawPageMessages: number;
}): number {
const oldestSeq = params.messages
.map((message) => readChatHistoryMessageSeq(message))
.find((seq): seq is number => typeof seq === "number");
if (oldestSeq !== undefined) {
return Math.max(params.offset, params.totalMessages - oldestSeq + 1);
}
return params.offset + params.rawPageMessages;
}
function capOffsetChatHistoryProjectedMessages(messages: unknown[], max: number): unknown[] {
if (messages.length <= max) {
return messages;
}
const start = Math.max(0, messages.length - max);
const boundarySeq = readChatHistoryMessageSeq(messages[start]);
if (boundarySeq === undefined) {
return messages.slice(start);
}
// Offset cursors can only resume at transcript-record boundaries.
// Keep boundary rows with the same seq together so projection mirrors are not stranded.
let safeStart = start;
while (safeStart > 0 && readChatHistoryMessageSeq(messages[safeStart - 1]) === boundarySeq) {
safeStart--;
}
return messages.slice(safeStart);
}
function dropChatHistoryOverreadContextMessage(
messages: unknown[],
contextMessage: unknown,
): unknown[] {
if (contextMessage === undefined) {
return messages;
}
const index = messages.indexOf(contextMessage);
if (index < 0) {
return messages;
}
return [...messages.slice(0, index), ...messages.slice(index + 1)];
}
async function handleSessionsList(params: Record<string, unknown>) {
const rt = await getRuntime();
const cfg = rt.getRuntimeConfig();
@@ -314,8 +229,11 @@ async function handleChatHistory(params: Record<string, unknown>): Promise<{
const offset = readOffsetParam(params) ?? 0;
const sessionLoadOptions = requestedAgentId ? { agentId: requestedAgentId } : undefined;
const { cfg, storePath, entry } = rt.loadSessionEntry(sessionKey, sessionLoadOptions);
const sessionId = entry?.sessionId as string | undefined;
const { cfg, storePath, entry, canonicalKey } = rt.loadSessionEntry(
sessionKey,
sessionLoadOptions,
);
const sessionId = entry?.sessionId;
const sessionAgentId = rt.resolveSessionAgentId({
sessionKey,
config: cfg,
@@ -326,162 +244,59 @@ async function handleChatHistory(params: Record<string, unknown>): Promise<{
const defaultLimit = 200;
const requested = typeof limit === "number" ? limit : defaultLimit;
const max = Math.min(hardMax, requested);
const rawHistoryWindowMessages = max * 20 + 20;
const maxHistoryBytes = rt.getMaxChatHistoryMessagesBytes();
const sessionEntry =
typeof entry?.sessionId === "string"
? {
sessionId: entry.sessionId,
...(typeof entry.sessionFile === "string" ? { sessionFile: entry.sessionFile } : {}),
}
: undefined;
const localMessages =
params.offset === undefined && sessionId && storePath
? await rt.readSessionMessagesAsync(
{
agentId: sessionAgentId,
sessionEntry,
sessionId,
sessionKey,
storePath,
},
params.offset === undefined
? {
mode: "recent",
maxMessages: max,
maxBytes: Math.max(maxHistoryBytes * 2, 1024 * 1024),
allowResetArchiveFallback: true,
}
: {
mode: "full",
reason: "chat.history offset pagination",
allowResetArchiveFallback: true,
},
)
: [];
const offsetPage =
params.offset !== undefined && sessionId && storePath
? offset === 0
? await rt.readRecentSessionMessagesWithStatsAsync(
{
agentId: sessionAgentId,
sessionEntry,
sessionId,
sessionKey,
storePath,
},
{
maxMessages: rawHistoryWindowMessages + 1,
maxBytes: Math.max(maxHistoryBytes * 2, 1024 * 1024),
allowResetArchiveFallback: true,
},
)
: await rt.readSessionMessagesPageWithStatsAsync(
{
agentId: sessionAgentId,
sessionEntry,
sessionId,
sessionKey,
storePath,
},
{
offset,
maxMessages: max + 1,
allowResetArchiveFallback: true,
},
)
: undefined;
const sessionStartedAt =
typeof entry?.sessionStartedAt === "number" ? entry.sessionStartedAt : undefined;
const offsetPageOverreadContextMessage =
offsetPage !== undefined
? offset === 0
? offsetPage.messages.length > rawHistoryWindowMessages
? offsetPage.messages[0]
: undefined
: offsetPage.messages.length > max
? offsetPage.messages[0]
: undefined
: undefined;
const localMessagesForHistory =
offsetPage !== undefined
? dropChatHistoryOverreadContextMessage(
rt.dropPreSessionStartAnnouncePairs(offsetPage.messages, sessionStartedAt),
offsetPageOverreadContextMessage,
)
: localMessages;
const rawMessages =
params.offset === undefined
? rt.augmentChatHistoryWithCliSessionImports({
entry,
provider: resolvedSessionModel.provider,
localMessages: localMessagesForHistory,
})
: localMessagesForHistory;
const recencyFilteredMessages = rt.dropPreSessionStartAnnouncePairs(
rawMessages,
sessionStartedAt,
);
const effectiveMaxChars = rt.resolveEffectiveChatHistoryMaxChars(cfg);
const page = await rt.readChatHistoryPage({
entry,
provider: resolvedSessionModel.provider,
sessionId,
storePath,
sessionAgentId,
canonicalKey,
max,
maxHistoryBytes,
effectiveMaxChars,
offset: params.offset === undefined ? undefined : offset,
messageId: undefined,
});
// Mirror Gateway chat.history trimming so embedded mode has the same byte ceilings.
const projected =
params.offset === undefined
? rt.projectRecentChatDisplayMessages(recencyFilteredMessages, {
includeCommentaryFallbacks: true,
maxChars: effectiveMaxChars,
maxMessages: max,
})
: offset === 0
? rt.projectRecentChatDisplayMessages(recencyFilteredMessages, {
includeCommentaryFallbacks: true,
maxChars: effectiveMaxChars,
maxMessages: max,
})
: rt.projectChatDisplayMessages(recencyFilteredMessages, {
includeCommentaryFallbacks: true,
maxChars: effectiveMaxChars,
});
const windowed =
params.offset === undefined || offset === 0
? projected
: capOffsetChatHistoryProjectedMessages(projected, max);
const normalized = rt.augmentChatHistoryWithCanvasBlocks(windowed);
// Keep transport-level byte limits identical after the shared reader projects the page.
const perMessageHardCap = Math.min(rt.CHAT_HISTORY_MAX_SINGLE_MESSAGE_BYTES, maxHistoryBytes);
const replaced = rt.replaceOversizedChatHistoryMessages({
messages: normalized,
messages: page.messages,
maxSingleMessageBytes: perMessageHardCap,
});
const capped = rt.capArrayByJsonBytes(replaced.messages, maxHistoryBytes).items;
const pagination = params.offset === undefined ? undefined : page.pagination;
const nextOffset =
offsetPage !== undefined
? resolveChatHistoryNextOffset({
pagination !== undefined
? rt.resolveChatHistoryNextOffset({
messages: capped,
totalMessages: offsetPage.totalMessages,
offset,
rawPageMessages:
offset === 0
? offsetPage.messages.length
: Math.min(max, Math.max(0, offsetPage.totalMessages - offset)),
totalMessages: pagination.totalMessages,
offset: pagination.offset,
rawPageMessages: pagination.rawPageMessages,
replayOldestRecord: rt.shouldReplayOldestChatHistoryRecord({
projected: page.messages,
bounded: capped,
}),
})
: 0;
const hasMore = offsetPage !== undefined ? nextOffset < offsetPage.totalMessages : false;
const hasMore =
pagination !== undefined &&
pagination.exhausted !== true &&
nextOffset < pagination.totalMessages;
return {
sessionKey,
sessionId,
messages: capped,
...(params.offset !== undefined
? { offset, hasMore, totalMessages: offsetPage?.totalMessages ?? projected.length }
? { offset, hasMore, totalMessages: pagination?.totalMessages ?? page.messages.length }
: {}),
...(hasMore && offsetPage !== undefined ? { nextOffset } : {}),
thinkingLevel: entry?.thinkingLevel as string | undefined,
...(hasMore ? { nextOffset } : {}),
thinkingLevel: entry?.thinkingLevel,
fastMode: normalizeFastMode(entry?.fastMode),
verboseLevel: entry?.verboseLevel as string | undefined,
verboseLevel: entry?.verboseLevel,
};
}
src/gateway/cli-session-history.test.ts
+9 -10
View File
@@ -11,7 +11,6 @@ import { redactTranscriptMessage } from "../agents/transcript-redact.js";
import { withEnvAsync } from "../test-utils/env.js";
import { readClaudeCliSessionMessages } from "./cli-session-history.claude.js";
import {
augmentChatHistoryWithCliSessionImports,
readClaudeCliFallbackSeed,
readChatHistoryCliSessionImportSnapshot,
resolveChatHistoryWithCliSessionImports,
@@ -20,7 +19,7 @@ import { mergeImportedChatHistoryMessages } from "./cli-session-history.merge.js
import { expectRecordFields, requireGatewayRecord } from "./test-helpers.assertions.js";
type ClaudeCliFallbackSeed = NonNullable<ReturnType<typeof readClaudeCliFallbackSeed>>;
type AugmentCliHistoryParams = Parameters<typeof augmentChatHistoryWithCliSessionImports>[0];
type AugmentCliHistoryParams = Parameters<typeof resolveChatHistoryWithCliSessionImports>[0];
function requireFallbackSeed(
seed: ReturnType<typeof readClaudeCliFallbackSeed>,
@@ -50,7 +49,7 @@ function augmentBoundClaudeHistory(params: {
provider: AugmentCliHistoryParams["provider"];
localMessages?: AugmentCliHistoryParams["localMessages"];
}) {
return augmentChatHistoryWithCliSessionImports({
return resolveChatHistoryWithCliSessionImports({
entry: {
sessionId: "openclaw-session",
updatedAt: Date.now(),
@@ -63,7 +62,7 @@ function augmentBoundClaudeHistory(params: {
provider: params.provider,
localMessages: params.localMessages ?? [],
homeDir: params.homeDir,
});
}).messages;
}
function buildLegacyReseedPrompt(current = "current"): string {
@@ -1265,7 +1264,7 @@ describe("cli session history", () => {
"utf-8",
);
const messages = augmentChatHistoryWithCliSessionImports({
const messages = resolveChatHistoryWithCliSessionImports({
entry: {
sessionId: "openclaw-session",
updatedAt: Date.now(),
@@ -1290,7 +1289,7 @@ describe("cli session history", () => {
},
],
homeDir,
});
}).messages;
expect(messages).toHaveLength(2);
expectFields(messages[0], { role: "user", content: "current recovered ask" });
@@ -1375,7 +1374,7 @@ describe("cli session history", () => {
it("falls back to legacy cliSessionIds when bindings are absent", async () => {
await withClaudeProjectsDir(async ({ homeDir, sessionId }) => {
const messages = augmentChatHistoryWithCliSessionImports({
const messages = resolveChatHistoryWithCliSessionImports({
entry: {
sessionId: "openclaw-session",
updatedAt: Date.now(),
@@ -1386,7 +1385,7 @@ describe("cli session history", () => {
provider: "claude-cli",
localMessages: [],
homeDir,
});
}).messages;
expect(messages).toHaveLength(3);
expectFields(messages[1], {
role: "assistant",
@@ -1397,7 +1396,7 @@ describe("cli session history", () => {
it("falls back to legacy claudeCliSessionId when newer fields are absent", async () => {
await withClaudeProjectsDir(async ({ homeDir, sessionId }) => {
const messages = augmentChatHistoryWithCliSessionImports({
const messages = resolveChatHistoryWithCliSessionImports({
entry: {
sessionId: "openclaw-session",
updatedAt: Date.now(),
@@ -1406,7 +1405,7 @@ describe("cli session history", () => {
provider: "claude-cli",
localMessages: [],
homeDir,
});
}).messages;
expect(messages).toHaveLength(3);
expectFields(messages[0], {
role: "user",
src/gateway/cli-session-history.ts
-7
View File
@@ -80,10 +80,3 @@ export async function readChatHistoryCliSessionImportSnapshot(
})
: [];
}
/** Augments local chat history with bound Claude CLI session messages when applicable. */
export function augmentChatHistoryWithCliSessionImports(
params: Parameters<typeof resolveChatHistoryWithCliSessionImports>[0],
): unknown[] {
return resolveChatHistoryWithCliSessionImports(params).messages;
}
src/gateway/server-methods/chat-history-handler.ts
+2 -42
View File
@@ -48,7 +48,8 @@ import {
capChatHistoryAroundMessage,
enrichChatHistoryCompactionMarkers,
readChatHistoryPage,
readChatHistoryMessageSeq,
resolveChatHistoryNextOffset,
shouldReplayOldestChatHistoryRecord,
} from "./chat-history-pages.js";
import { resolveRequestedChatAgentId, validateChatSelectedAgent } from "./chat-origin-routing.js";
import { normalizeOptionalChatText as normalizeOptionalText } from "./chat-text-normalization.js";
@@ -111,47 +112,6 @@ async function handleChatMetadataRequest({
// The UI fills metadata gaps as soon as chat.startup returns, so history never waits
// beyond this budget for a catalog snapshot that requires slower discovery.
const CHAT_OPTIONAL_MODEL_CATALOG_TIMEOUT_MS = 25;
function resolveChatHistoryNextOffset(params: {
messages: unknown[];
totalMessages: number;
offset: number;
rawPageMessages: number;
replayOldestRecord?: boolean;
}): number {
const oldestSeq = params.messages
.map((message) => readChatHistoryMessageSeq(message))
.find((seq): seq is number => typeof seq === "number");
if (oldestSeq !== undefined) {
const recordOffset = params.totalMessages - oldestSeq + 1;
const replayOffset = recordOffset - 1;
if (params.replayOldestRecord && replayOffset > params.offset) {
return replayOffset;
}
// A replay cursor that does not advance strands every older record. Skip
// the pathological projected siblings and continue with the next record.
return Math.max(params.offset + 1, recordOffset);
}
return params.offset + params.rawPageMessages;
}
function shouldReplayOldestChatHistoryRecord(params: {
projected: unknown[];
bounded: unknown[];
}): boolean {
const oldestSeq = params.bounded
.map((message) => readChatHistoryMessageSeq(message))
.find((seq): seq is number => typeof seq === "number");
if (oldestSeq === undefined) {
return false;
}
const projectedCount = params.projected.filter(
(message) => readChatHistoryMessageSeq(message) === oldestSeq,
).length;
const boundedCount = params.bounded.filter(
(message) => readChatHistoryMessageSeq(message) === oldestSeq,
).length;
return boundedCount < projectedCount;
}
async function handleChatHistoryRequest({
params,
src/gateway/server-methods/chat-history-pages.ts
+48 -224
View File
@@ -1,11 +1,9 @@
import { asPositiveSafeInteger } from "@openclaw/normalization-core/number-coercion";
import { asOptionalRecord } from "@openclaw/normalization-core/record-coerce";
import { resolveSessionTranscriptActiveLeafEntryId } from "../../config/sessions/session-accessor.js";
import {
dropPreSessionStartAnnouncePairs,
isHeartbeatHistoryTurnBoundaryMessage,
projectChatDisplayMessages,
projectRecentChatDisplayMessages,
augmentChatHistoryWithCanvasBlocks,
} from "../chat-display-projection.js";
import {
@@ -14,14 +12,17 @@ import {
resolveClaudeCliBindingSessionId,
} from "../cli-session-history.js";
import { resolveCurrentUserProfileDisplay } from "../current-user-profile-display.js";
import { resolveSessionHistoryTailReadOptions } from "../session-history-state.js";
import {
capOffsetChatHistoryProjectedMessages,
dropChatHistoryOverreadContextMessage,
readChatHistoryMessageSeq,
readIncrementalChatHistoryTail,
type IncrementalChatHistoryTail,
} from "../session-history-tail.js";
import { readSessionMessagesAroundIdWithStatsAsync } from "../session-transcript-anchor-reader.js";
import {
readRecentSessionMessagesWithStatsAsync,
readSessionMessagesAsync,
readSessionMessagesPageWithStatsAsync,
type ReadRecentSessionMessagesResult,
type SessionTranscriptReadScope,
} from "../session-transcript-readers.js";
import type { loadSessionEntry } from "../session-utils.js";
@@ -30,11 +31,6 @@ export function readChatHistoryMessageId(message: unknown): string | undefined {
return typeof metadata?.id === "string" ? metadata.id : undefined;
}
export function readChatHistoryMessageSeq(message: unknown): number | undefined {
const metadata = asOptionalRecord(asOptionalRecord(message)?.["__openclaw"]);
return asPositiveSafeInteger(metadata?.seq);
}
type ChatHistoryPage = {
activeLeafEntryId?: string | null;
deltaCursor?: string;
@@ -52,6 +48,42 @@ type ChatHistoryPage = {
};
};
export function resolveChatHistoryNextOffset(params: {
messages: unknown[];
totalMessages: number;
offset: number;
rawPageMessages: number;
replayOldestRecord?: boolean;
}): number {
const oldestSeq = params.messages
.map((message) => readChatHistoryMessageSeq(message))
.find((seq): seq is number => typeof seq === "number");
if (oldestSeq === undefined) {
return params.offset + params.rawPageMessages;
}
const recordOffset = params.totalMessages - oldestSeq + 1;
const replayOffset = recordOffset - 1;
if (params.replayOldestRecord && replayOffset > params.offset) {
return replayOffset;
}
// A replay cursor that does not advance strands every older transcript record.
return Math.max(params.offset + 1, recordOffset);
}
export function shouldReplayOldestChatHistoryRecord(params: {
projected: unknown[];
bounded: unknown[];
}): boolean {
const oldestSeq = params.bounded
.map((message) => readChatHistoryMessageSeq(message))
.find((seq): seq is number => typeof seq === "number");
return (
oldestSeq !== undefined &&
params.bounded.filter((message) => readChatHistoryMessageSeq(message) === oldestSeq).length <
params.projected.filter((message) => readChatHistoryMessageSeq(message) === oldestSeq).length
);
}
function resolveChatHistoryActiveLeafEntryId(
readPage: ReadRecentSessionMessagesResult,
): string | null {
@@ -113,24 +145,6 @@ export function enrichChatHistoryCompactionMarkers(
return changed ? enriched : messages;
}
function capOffsetChatHistoryProjectedMessages(messages: unknown[], max: number): unknown[] {
if (messages.length <= max) {
return messages;
}
const start = Math.max(0, messages.length - max);
const boundarySeq = readChatHistoryMessageSeq(messages[start]);
if (boundarySeq === undefined) {
return messages.slice(start);
}
// Offset cursors can only resume at transcript-record boundaries.
// Keep boundary rows with the same seq together so projection mirrors are not stranded.
let safeStart = start;
while (safeStart > 0 && readChatHistoryMessageSeq(messages[safeStart - 1]) === boundarySeq) {
safeStart--;
}
return messages.slice(safeStart);
}
function resolveChatHistoryMessageGroup(
messages: unknown[],
index: number,
@@ -193,180 +207,6 @@ export function capChatHistoryAroundMessage(params: {
return params.messages.slice(start, end);
}
function dropLocalHistoryOverreadContextMessage(
messages: unknown[],
contextMessage: unknown,
): unknown[] {
if (contextMessage === undefined) {
return messages;
}
const index = messages.indexOf(contextMessage);
if (index < 0) {
return messages;
}
return [...messages.slice(0, index), ...messages.slice(index + 1)];
}
// A silent tail can outrun the bounded raw window: tool traffic, hidden
// memory-flush prompts, and dropped silent turns all consume raw records
// without producing a display row, so the newest window can project to nothing
// while visible history is still on the branch. Snapshot clients rebuild
// destructively from a first page, so returning that empty page erases a
// rendered conversation the transcript still holds. Scan older pages until a
// display row appears, bounded so a pathological transcript cannot turn the
// tail read into a full scan.
const SILENT_CHAT_HISTORY_TAIL_SCAN_MAX_MESSAGES = 8_000;
// Chunk size stays independent of the requested display limit: a `limit: 1`
// request must not turn the scan into one transcript read per raw record. Pages
// are contiguous and released between iterations, and the chunk stays below the
// record count an explicit offset page already materializes, so peak scan memory
// never exceeds what one ordinary history page costs. A byte-budgeted read is
// deliberately not used here: it drops oversized records from the middle of a
// window, which would punch holes in the cursor and hide the oversized-message
// placeholder the handler would otherwise render. Accepted tradeoff: a chunk is
// materialized before its bytes can be counted, so the walk can overshoot its
// budget by one page. That page is smaller than the one every explicit offset
// request already reads, and closing the gap would need a stop-at-budget
// contiguous reader that does not exist yet.
const SILENT_CHAT_HISTORY_TAIL_SCAN_CHUNK_MESSAGES = 100;
type IncrementalChatHistoryTail = {
overreadContextMessage: unknown;
projected: unknown[];
rawMessages: unknown[];
rawPageMessages: number;
readPage: ReadRecentSessionMessagesResult;
};
/** Reads only enough raw tail records to fill one projected history page. */
async function readIncrementalChatHistoryTail(params: {
entry: ReturnType<typeof loadSessionEntry>["entry"];
readScope: SessionTranscriptReadScope;
effectiveMaxChars: number;
max: number;
maxBytes: number;
offset?: number;
}): Promise<IncrementalChatHistoryTail> {
const offset = params.offset ?? 0;
const rawHistoryWindow = resolveSessionHistoryTailReadOptions(params.max);
// Three raw rows per requested display row covers common tool/silent pairs
// while keeping the first read far below the legacy 20x safety ceiling.
const initialMessages = Math.min(
rawHistoryWindow.maxMessages,
Math.max(1, offset === 0 ? params.max * 3 : params.max),
);
const readPage =
offset === 0
? await readRecentSessionMessagesWithStatsAsync(params.readScope, {
maxMessages: initialMessages + 1,
maxLines: initialMessages + 1,
maxBytes: Math.max(params.maxBytes * 2, 1024 * 1024),
allowResetArchiveFallback: true,
})
: await readSessionMessagesPageWithStatsAsync(params.readScope, {
offset,
maxMessages: initialMessages + 1,
allowResetArchiveFallback: true,
});
const sessionStartedAt =
typeof params.entry?.sessionStartedAt === "number" ? params.entry.sessionStartedAt : undefined;
let rawPageMessages = Math.min(
initialMessages,
Math.max(readPage.messages.length, readPage.totalMessages > offset ? 1 : 0),
);
let overreadContextMessage =
readPage.messages.length > initialMessages ? readPage.messages[0] : undefined;
let rawMessages = dropLocalHistoryOverreadContextMessage(
readPage.messages,
overreadContextMessage,
);
const filteredRawMessages = () =>
dropLocalHistoryOverreadContextMessage(
dropPreSessionStartAnnouncePairs(
overreadContextMessage === undefined
? rawMessages
: [overreadContextMessage, ...rawMessages],
sessionStartedAt,
),
overreadContextMessage,
);
const project = () => {
const options = {
maxChars: params.effectiveMaxChars,
resolveCurrentUserProfileDisplay,
turnBoundaryPending: isHeartbeatHistoryTurnBoundaryMessage(overreadContextMessage),
};
return offset === 0
? projectRecentChatDisplayMessages(filteredRawMessages(), {
...options,
maxMessages: params.max,
})
: capOffsetChatHistoryProjectedMessages(
projectChatDisplayMessages(filteredRawMessages(), options),
params.max,
);
};
let projected = project();
let scanLimit = rawHistoryWindow.maxMessages;
// Record count alone does not bound a walk over large tool results, and the
// reader cannot bound it either: a byte-budgeted page skips an oversized
// record mid-window, which would strand it and its placeholder. Budget the
// whole walk instead, at the payload one history response may already return.
let scannedBytes = 0;
// Projection pairs records across turns, so keep the accumulated window in
// chronological order and retain exactly one older context record.
while (offset + rawPageMessages < readPage.totalMessages) {
if (projected.length >= params.max) {
break;
}
if (rawPageMessages >= rawHistoryWindow.maxMessages) {
if (projected.length > 0) {
break;
}
scanLimit = rawHistoryWindow.maxMessages + SILENT_CHAT_HISTORY_TAIL_SCAN_MAX_MESSAGES;
}
if (rawPageMessages >= scanLimit) {
break;
}
const chunkMessages = Math.min(
SILENT_CHAT_HISTORY_TAIL_SCAN_CHUNK_MESSAGES,
scanLimit - rawPageMessages,
);
const page = await readSessionMessagesPageWithStatsAsync(params.readScope, {
offset: offset + rawPageMessages,
maxMessages: chunkMessages + 1,
allowResetArchiveFallback: true,
});
if (page.messages.length === 0) {
break;
}
// The extra oldest record only supplies pair-filter and turn-boundary
// context. Without it a chunk boundary between a stale announce and its
// reply would leak the reply that the tail read hides. Each chunk's context
// record is the newest record of the next chunk, so this one overread also
// covers every junction the walk creates, including tail-to-first-chunk.
const contextMessage = page.messages.length > chunkMessages ? page.messages[0] : undefined;
rawPageMessages += page.messages.length - (contextMessage === undefined ? 0 : 1);
rawMessages = dropLocalHistoryOverreadContextMessage(
[...page.messages, ...rawMessages],
contextMessage,
);
overreadContextMessage = contextMessage;
projected = project();
scannedBytes += Buffer.byteLength(JSON.stringify(page.messages), "utf8");
if (rawPageMessages > rawHistoryWindow.maxMessages && scannedBytes >= params.maxBytes) {
break;
}
}
return {
overreadContextMessage,
projected,
rawMessages: filteredRawMessages(),
rawPageMessages,
readPage,
};
}
export async function readChatHistoryPage(params: {
entry: ReturnType<typeof loadSessionEntry>["entry"];
provider: string | undefined;
@@ -456,7 +296,7 @@ export async function readChatHistoryPage(params: {
: undefined;
const localMessages = incrementalTail
? incrementalTail.rawMessages
: dropLocalHistoryOverreadContextMessage(
: dropChatHistoryOverreadContextMessage(
dropPreSessionStartAnnouncePairs(
readPage.messages,
typeof entry?.sessionStartedAt === "number" ? entry.sessionStartedAt : undefined,
@@ -469,12 +309,9 @@ export async function readChatHistoryPage(params: {
max,
Math.max(readPage.messages.length, readPage.totalMessages > pageOffset ? 1 : 0),
);
// localMessages is already announce-filtered above; the filter is
// single-pass complete, so no second pass is needed.
const recencyFilteredMessages = localMessages;
const projected = incrementalTail
? incrementalTail.projected
: projectChatDisplayMessages(recencyFilteredMessages, {
: projectChatDisplayMessages(localMessages, {
includeCommentaryFallbacks: true,
maxChars: effectiveMaxChars,
resolveCurrentUserProfileDisplay,
@@ -486,9 +323,7 @@ export async function readChatHistoryPage(params: {
messageId,
fits: (messages) => messages.length <= max,
}) ?? capOffsetChatHistoryProjectedMessages(projected, max))
: isTailPage
? projected
: capOffsetChatHistoryProjectedMessages(projected, max);
: projected;
if (messageId) {
// Numeric offsets do not encode the selected historical transcript source.
return { messages: augmentChatHistoryWithCanvasBlocks(windowed) };
@@ -519,8 +354,7 @@ export async function readChatHistoryPage(params: {
max,
maxBytes: maxHistoryBytes,
});
const { overreadContextMessage, readPage } = incrementalTail;
const turnBoundaryPending = isHeartbeatHistoryTurnBoundaryMessage(overreadContextMessage);
const { readPage } = incrementalTail;
const activeLeafEntryId = resolveChatHistoryActiveLeafEntryId(readPage);
const localMessagesWithBoundaryFilter = incrementalTail.rawMessages;
// The ignore flag must gate this resolver too: the tail-window merge can report
@@ -585,22 +419,12 @@ export async function readChatHistoryPage(params: {
},
};
}
// The imported case returned above, so these are the already announce-filtered
// local messages; the filter is single-pass complete, so no second pass is needed.
const recencyFilteredMessages = cliHistory.messages;
const displayMessages = projectRecentChatDisplayMessages(recencyFilteredMessages, {
includeCommentaryFallbacks: true,
maxChars: effectiveMaxChars,
maxMessages: max,
resolveCurrentUserProfileDisplay,
turnBoundaryPending,
});
return {
activeLeafEntryId,
...(readPage.transcriptSource === "active" && readPage.deltaCursor
? { deltaCursor: readPage.deltaCursor }
: {}),
messages: augmentChatHistoryWithCanvasBlocks(displayMessages),
messages: augmentChatHistoryWithCanvasBlocks(incrementalTail.projected),
pagination: {
offset: 0,
totalMessages: readPage.totalMessages,
src/gateway/session-history-state.ts
+25 -26
View File
@@ -2,17 +2,20 @@
// Tracks transcript sequence windows for paginated chat-history SSE updates.
import { isDeepStrictEqual } from "node:util";
import { asPositiveSafeInteger } from "@openclaw/normalization-core/number-coercion";
import type { SessionEntry } from "../config/sessions.js";
import {
DEFAULT_CHAT_HISTORY_TEXT_MAX_CHARS,
projectChatDisplayMessages,
projectChatDisplayMessagesWithState,
} from "./chat-display-projection.js";
import { resolveCurrentUserProfileDisplay } from "./current-user-profile-display.js";
import { getMaxChatHistoryMessagesBytes } from "./server-constants.js";
import { readIncrementalChatHistoryTail } from "./session-history-tail.js";
import { resolveTranscriptPathForComparison } from "./session-transcript-path.js";
import {
attachOpenClawTranscriptMeta,
readRecentSessionMessagesWithStatsAsync,
readSessionMessagesWithSourceAsync,
type ReadRecentSessionMessagesResult,
} from "./session-transcript-readers.js";
// Session history state owns the SSE-friendly projection of transcript JSONL:
@@ -50,7 +53,7 @@ type InlineSessionHistoryAppend = {
type SessionHistoryTranscriptTarget = {
agentId?: string;
sessionEntry?: { sessionFile?: string; sessionId?: string };
sessionEntry?: SessionEntry;
sessionId: string;
sessionKey: string;
storePath?: string;
@@ -71,17 +74,21 @@ function readMessageIdempotencyKey(message: unknown): string | undefined {
return typeof value === "string" && value.trim() ? value : undefined;
}
/** Computes an oversized raw transcript tail window for projected chat history. */
export function resolveSessionHistoryTailReadOptions(limit: number): {
maxMessages: number;
maxLines: number;
} {
const requested = Math.max(1, Math.floor(limit));
const rawWindow = requested * 20 + 20;
return {
maxMessages: rawWindow,
maxLines: rawWindow,
};
/** Shares the bounded visible-message scanner across HTTP snapshots and SSE refreshes. */
export async function readBoundedSessionHistorySnapshotAsync(params: {
target: SessionHistoryTranscriptTarget;
limit: number;
maxChars: number;
}): Promise<ReadRecentSessionMessagesResult> {
const tail = await readIncrementalChatHistoryTail({
entry: params.target.sessionEntry,
readScope: params.target,
effectiveMaxChars: params.maxChars,
max: params.limit,
maxBytes: getMaxChatHistoryMessagesBytes(),
preserveProjectionContext: true,
});
return { ...tail.readPage, messages: tail.rawMessages };
}
export function resolveCursorSeq(cursor: string | undefined): number | undefined {
@@ -457,19 +464,11 @@ export class SessionHistorySseState {
private async readRawSnapshotAsync(): Promise<SessionHistoryRawSnapshot> {
if (this.cursor === undefined && typeof this.limit === "number") {
const snapshot = await readRecentSessionMessagesWithStatsAsync(
{
agentId: this.target.agentId,
sessionEntry: this.target.sessionEntry,
sessionId: this.target.sessionId,
sessionKey: this.target.sessionKey,
storePath: this.target.storePath,
},
{
...resolveSessionHistoryTailReadOptions(this.limit),
allowResetArchiveFallback: true,
},
);
const snapshot = await readBoundedSessionHistorySnapshotAsync({
target: this.target,
limit: this.limit,
maxChars: this.maxChars,
});
return {
rawMessages: snapshot.messages,
rawTranscriptSeq: snapshot.totalMessages,
src/gateway/session-history-tail.ts
+180
View File
@@ -0,0 +1,180 @@
import { asPositiveSafeInteger } from "@openclaw/normalization-core/number-coercion";
import { asOptionalRecord } from "@openclaw/normalization-core/record-coerce";
import type { SessionEntry } from "../config/sessions.js";
import {
dropPreSessionStartAnnouncePairs,
isHeartbeatHistoryTurnBoundaryMessage,
projectChatDisplayMessages,
projectRecentChatDisplayMessages,
} from "./chat-display-projection.js";
import { resolveCurrentUserProfileDisplay } from "./current-user-profile-display.js";
import {
readRecentSessionMessagesWithStatsAsync,
readSessionMessagesPageWithStatsAsync,
type ReadRecentSessionMessagesResult,
type SessionTranscriptReadScope,
} from "./session-transcript-readers.js";
const SILENT_CHAT_HISTORY_TAIL_SCAN_MAX_MESSAGES = 8_000;
const SILENT_CHAT_HISTORY_TAIL_SCAN_CHUNK_MESSAGES = 100;
export function readChatHistoryMessageSeq(message: unknown): number | undefined {
const metadata = asOptionalRecord(asOptionalRecord(message)?.["__openclaw"]);
return asPositiveSafeInteger(metadata?.seq);
}
export function capOffsetChatHistoryProjectedMessages(messages: unknown[], max: number): unknown[] {
if (messages.length <= max) {
return messages;
}
const start = Math.max(0, messages.length - max);
const boundarySeq = readChatHistoryMessageSeq(messages[start]);
if (boundarySeq === undefined) {
return messages.slice(start);
}
// Numeric cursors resume at transcript records, so projected siblings stay together.
let safeStart = start;
while (safeStart > 0 && readChatHistoryMessageSeq(messages[safeStart - 1]) === boundarySeq) {
safeStart--;
}
return messages.slice(safeStart);
}
export function dropChatHistoryOverreadContextMessage(
messages: unknown[],
contextMessage: unknown,
): unknown[] {
if (contextMessage === undefined) {
return messages;
}
const index = messages.indexOf(contextMessage);
return index < 0 ? messages : [...messages.slice(0, index), ...messages.slice(index + 1)];
}
export type IncrementalChatHistoryTail = {
overreadContextMessage: unknown;
projected: unknown[];
rawMessages: unknown[];
rawPageMessages: number;
readPage: ReadRecentSessionMessagesResult;
};
/** Scans indexed transcript records until one bounded visible history page is filled. */
export async function readIncrementalChatHistoryTail(params: {
entry: SessionEntry | undefined;
readScope: SessionTranscriptReadScope;
effectiveMaxChars: number;
max: number;
maxBytes: number;
offset?: number;
preserveProjectionContext?: boolean;
}): Promise<IncrementalChatHistoryTail> {
const offset = params.offset ?? 0;
const rawHistoryWindowMessages = Math.max(1, Math.floor(params.max)) * 20 + 20;
// Sequence-cursor transports group tool results and derived mirrors together,
// so their initial read keeps the established wider projection context.
const initialMessages = params.preserveProjectionContext
? rawHistoryWindowMessages
: Math.min(rawHistoryWindowMessages, Math.max(1, offset === 0 ? params.max * 3 : params.max));
const readPage =
offset === 0
? await readRecentSessionMessagesWithStatsAsync(params.readScope, {
maxMessages: initialMessages + 1,
maxLines: initialMessages + 1,
maxBytes: Math.max(params.maxBytes * 2, 1024 * 1024),
allowResetArchiveFallback: true,
})
: await readSessionMessagesPageWithStatsAsync(params.readScope, {
offset,
maxMessages: initialMessages + 1,
allowResetArchiveFallback: true,
});
const sessionStartedAt =
typeof params.entry?.sessionStartedAt === "number" ? params.entry.sessionStartedAt : undefined;
let rawPageMessages = Math.min(
initialMessages,
Math.max(readPage.messages.length, readPage.totalMessages > offset ? 1 : 0),
);
let overreadContextMessage =
readPage.messages.length > initialMessages ? readPage.messages[0] : undefined;
let rawMessages = dropChatHistoryOverreadContextMessage(
readPage.messages,
overreadContextMessage,
);
const filteredRawMessages = () =>
dropChatHistoryOverreadContextMessage(
dropPreSessionStartAnnouncePairs(
overreadContextMessage === undefined
? rawMessages
: [overreadContextMessage, ...rawMessages],
sessionStartedAt,
),
overreadContextMessage,
);
const project = () => {
const options = {
includeCommentaryFallbacks: true,
maxChars: params.effectiveMaxChars,
resolveCurrentUserProfileDisplay,
turnBoundaryPending: isHeartbeatHistoryTurnBoundaryMessage(overreadContextMessage),
};
return offset === 0
? projectRecentChatDisplayMessages(filteredRawMessages(), {
...options,
maxMessages: params.max,
})
: capOffsetChatHistoryProjectedMessages(
projectChatDisplayMessages(filteredRawMessages(), options),
params.max,
);
};
let projected = project();
let scanLimit = rawHistoryWindowMessages;
let scannedBytes = 0;
while (offset + rawPageMessages < readPage.totalMessages) {
if (projected.length >= params.max) {
break;
}
if (rawPageMessages >= rawHistoryWindowMessages) {
if (projected.length > 0) {
break;
}
scanLimit = rawHistoryWindowMessages + SILENT_CHAT_HISTORY_TAIL_SCAN_MAX_MESSAGES;
}
if (rawPageMessages >= scanLimit) {
break;
}
const chunkMessages = Math.min(
SILENT_CHAT_HISTORY_TAIL_SCAN_CHUNK_MESSAGES,
scanLimit - rawPageMessages,
);
const page = await readSessionMessagesPageWithStatsAsync(params.readScope, {
offset: offset + rawPageMessages,
maxMessages: chunkMessages + 1,
allowResetArchiveFallback: true,
});
if (page.messages.length === 0) {
break;
}
// One older context row preserves stale-pair and heartbeat boundaries across chunks.
const contextMessage = page.messages.length > chunkMessages ? page.messages[0] : undefined;
rawPageMessages += page.messages.length - (contextMessage === undefined ? 0 : 1);
rawMessages = dropChatHistoryOverreadContextMessage(
[...page.messages, ...rawMessages],
contextMessage,
);
overreadContextMessage = contextMessage;
projected = project();
scannedBytes += Buffer.byteLength(JSON.stringify(page.messages), "utf8");
if (rawPageMessages > rawHistoryWindowMessages && scannedBytes >= params.maxBytes) {
break;
}
}
return {
overreadContextMessage,
projected,
rawMessages: filteredRawMessages(),
rawPageMessages,
readPage,
};
}
src/gateway/sessions-history-http.revocation.test.ts
+6 -4
View File
@@ -141,10 +141,12 @@ vi.mock("./session-history-state.js", () => ({
history: { items: [], nextCursor: null, messages: [] },
}),
resolveCursorSeq: (_cursor: string | undefined) => undefined,
resolveSessionHistoryTailReadOptions: (limit: number) => ({
maxMessages: limit * 20 + 20,
maxLines: limit * 20 + 20,
}),
readBoundedSessionHistorySnapshotAsync: async () => {
if (transcriptReadError) {
throw transcriptReadError;
}
return { messages: [], totalMessages: 0 };
},
SessionHistorySseState: {
fromRawSnapshot: (_params: unknown) => ({
snapshot: () => ({ items: [], nextCursor: null, messages: [] }),
src/gateway/sessions-history-http.test.ts
+12 -14
View File
@@ -1215,7 +1215,7 @@ describe("session history HTTP endpoints", () => {
});
});
test("keeps older SQLite history reachable past an all-silent bounded tail", async () => {
test("backfills REST and SSE history past an all-silent bounded tail", async () => {
const storePath = await createSessionStoreFile();
const sessionId = "sess-silent-tail";
const sessionKey = "agent:main:main";
@@ -1235,9 +1235,11 @@ describe("session history HTTP endpoints", () => {
const firstPage = await readSessionHistoryBody(harness.port, sessionKey, {
query: "?limit=1",
});
expect(firstPage.messages).toEqual([]);
expect(firstPage.hasMore).toBe(true);
expect(firstPage.nextCursor).toBe("2");
expect(firstPage.messages?.map((message) => message.content)).toEqual([
"reachable older history",
]);
expect(firstPage.hasMore).toBe(false);
expect(firstPage.nextCursor).toBeUndefined();
const stream = await openSessionHistorySse(harness.port, sessionKey, {
query: "?limit=1",
@@ -1245,19 +1247,15 @@ describe("session history HTTP endpoints", () => {
try {
const event = await readSseEvent(stream.reader, stream.streamState);
expect(event.event).toBe("history");
expect(event.data).toMatchObject({ messages: [], hasMore: true, nextCursor: "2" });
const history = event.data as SessionHistoryBody;
expect(history.messages?.map((message) => message.content)).toEqual([
"reachable older history",
]);
expect(history.hasMore).toBe(false);
expect(history.nextCursor).toBeUndefined();
} finally {
await stream.reader.cancel();
}
const olderPage = await readSessionHistoryBody(harness.port, sessionKey, {
query: "?limit=1&cursor=2",
});
expect(olderPage.messages?.map((message) => message.content)).toEqual([
"reachable older history",
]);
expect(olderPage.hasMore).toBe(false);
expect(olderPage.nextCursor).toBeUndefined();
});
});
src/gateway/sessions-history-http.ts
+8 -13
View File
@@ -36,16 +36,13 @@ import { authorizeOperatorScopesForMethod } from "./method-scopes.js";
import type { GatewayClient } from "./server-methods/shared-types.js";
import {
buildSessionHistorySnapshot,
readBoundedSessionHistorySnapshotAsync,
resolveCursorSeq,
resolveSessionHistoryTailReadOptions,
SessionHistorySseState,
} from "./session-history-state.js";
import { createSessionListEntryFilter, resolveSessionSharingTarget } from "./session-sharing.js";
import { resolveTranscriptPathForComparison } from "./session-transcript-path.js";
import {
readRecentSessionMessagesWithStatsAsync,
readSessionMessagesWithSourceAsync,
} from "./session-transcript-readers.js";
import { readSessionMessagesWithSourceAsync } from "./session-transcript-readers.js";
import {
resolveCanonicalSessionEntryFromStoreKeys,
resolveGatewaySessionStoreTargetWithStore,
@@ -222,25 +219,23 @@ export async function handleSessionHistoryHttpRequest(
}
const effectiveMaxChars = DEFAULT_CHAT_HISTORY_TEXT_MAX_CHARS;
let boundedSnapshot:
| Awaited<ReturnType<typeof readRecentSessionMessagesWithStatsAsync>>
| Awaited<ReturnType<typeof readBoundedSessionHistorySnapshotAsync>>
| undefined;
let fullSnapshot: Awaited<ReturnType<typeof readSessionMessagesWithSourceAsync>> | undefined;
try {
boundedSnapshot =
cursor === undefined && typeof limit === "number"
? await readRecentSessionMessagesWithStatsAsync(
{
? await readBoundedSessionHistorySnapshotAsync({
target: {
agentId: target.agentId,
sessionEntry: entry,
sessionId: entry.sessionId,
sessionKey: target.canonicalKey,
storePath: target.storePath,
},
{
...resolveSessionHistoryTailReadOptions(limit),
allowResetArchiveFallback: true,
},
)
limit,
maxChars: effectiveMaxChars,
})
: undefined;
// Cursor reads still need an arbitrary historical window. The common first
// page path is bounded above so `limit=1` cannot materialize huge transcripts.
ui/src/e2e/chat-cli-history-redaction.e2e.test.ts
+3 -3
View File
@@ -3,7 +3,7 @@ import path from "node:path";
import { afterEach, expect, it } from "vitest";
import type { AgentMessage } from "../../../src/agents/runtime/index.js";
import { redactTranscriptMessage } from "../../../src/agents/transcript-redact.js";
import { augmentChatHistoryWithCliSessionImports } from "../../../src/gateway/cli-session-history.js";
import { resolveChatHistoryWithCliSessionImports } from "../../../src/gateway/cli-session-history.js";
import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js";
import {
chatSessionListResponse,
@@ -61,7 +61,7 @@ suite.define(() => {
content: userText,
timestamp: Date.parse("2026-03-26T16:29:54.800Z"),
} as AgentMessage);
const mergedMessages = augmentChatHistoryWithCliSessionImports({
const mergedMessages = resolveChatHistoryWithCliSessionImports({
entry: {
sessionId: "control-ui-local-claude-history",
updatedAt: Date.now(),
@@ -70,7 +70,7 @@ suite.define(() => {
provider: "claude-cli",
localMessages: [localUserMessage],
homeDir,
});
}).messages;
expect(mergedMessages).toHaveLength(2);
expect(mergedMessages[0]).toBe(localUserMessage);