From a19da06a2d09d364d79fd7aac82db73dcbd2aec6 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Tue, 25 Aug 2026 01:18:14 -0700 Subject: [PATCH] improve(gateway): load session history faster without duplicate readers (#129114) * perf(gateway): consolidate session history readers * chore: leave changelog to release generation --- ...sion-accessor.sqlite-active-events.test.ts | 11 +- .../session-accessor.sqlite-active-events.ts | 108 ---------- ...ion-accessor.sqlite-history-events.test.ts | 185 ++++++------------ .../session-accessor.sqlite-history-events.ts | 3 + ...ession-accessor.sqlite-message-cut.test.ts | 7 +- src/config/sessions/session-accessor.ts | 3 - src/gateway/session-history-state.ts | 155 +++++---------- src/gateway/session-transcript-readers.ts | 26 +-- .../sessions-history-http.revocation.test.ts | 20 +- src/gateway/sessions-history-http.test.ts | 7 +- src/gateway/sessions-history-http.ts | 53 ++--- 11 files changed, 145 insertions(+), 433 deletions(-) diff --git a/src/config/sessions/session-accessor.sqlite-active-events.test.ts b/src/config/sessions/session-accessor.sqlite-active-events.test.ts index 03383ce59b24..2f4645e1c0da 100644 --- a/src/config/sessions/session-accessor.sqlite-active-events.test.ts +++ b/src/config/sessions/session-accessor.sqlite-active-events.test.ts @@ -14,12 +14,13 @@ import { readSessionTranscriptActivePathEntryRelation, readSessionTranscriptActiveStats, readSessionTranscriptBoundedMessageTailPage, - readSessionTranscriptMessageAnchorPage, - readSessionTranscriptMessageEventById, - readSessionTranscriptMessageEventCount, readSessionTranscriptMessageEventPage, SessionTranscriptProjectionUnavailableError, } from "./session-accessor.sqlite-active-events.js"; +import { + readSessionTranscriptHistoryAnchorPage as readSessionTranscriptMessageAnchorPage, + readSessionTranscriptHistoryEventById as readSessionTranscriptMessageEventById, +} from "./session-accessor.sqlite-history-events.js"; import { runExclusiveSqliteSessionWrite } from "./session-accessor.sqlite-scope.js"; import { appendTranscriptEventsInTransaction } from "./session-accessor.sqlite-transcript-store.js"; import { @@ -45,6 +46,10 @@ vi.mock("../../shared/store-writer-queue.js", async (importOriginal) => { }); const tempDirs = useAutoCleanupTempDirTracker(afterEach); +const readSessionTranscriptMessageEventCount = ( + scope: Parameters[0], +): number => + readSessionTranscriptMessageEventPage(scope, { maxMessages: 0, offset: 0 }).totalMessages; describe("SQLite active transcript event projection", () => { let stateDir: string; diff --git a/src/config/sessions/session-accessor.sqlite-active-events.ts b/src/config/sessions/session-accessor.sqlite-active-events.ts index ad2e49928869..bf792125aaea 100644 --- a/src/config/sessions/session-accessor.sqlite-active-events.ts +++ b/src/config/sessions/session-accessor.sqlite-active-events.ts @@ -512,111 +512,3 @@ export function readSessionTranscriptBoundedMessageTailPage( }; }); } - -export function readSessionTranscriptMessageEventCount(scope: SessionTranscriptReadScope): number { - return withCurrentProjectionSnapshot( - scope, - (projection) => resolveVisibleMessagePositions(projection).total, - ); -} - -/** Reads one active message by event id without materializing sibling rows. */ -export function readSessionTranscriptMessageEventById( - scope: SessionTranscriptReadScope, - messageId: string, -): SessionTranscriptMessageEvent | undefined { - return withCurrentProjectionSnapshot(scope, (projection) => { - const db = getActiveTranscriptKysely(projection.database); - const row = executeSqliteQueryTakeFirstSync( - projection.database.db, - db - .selectFrom("transcript_event_identities as identity") - .innerJoin("session_transcript_active_events as active", (join) => - join - .onRef("active.session_id", "=", "identity.session_id") - .onRef("active.event_seq", "=", "identity.seq"), - ) - .innerJoin("transcript_events as event", (join) => - join - .onRef("event.session_id", "=", "active.session_id") - .onRef("event.seq", "=", "active.event_seq"), - ) - .select(["active.message_position", "event.event_json"]) - .where("identity.session_id", "=", projection.resolved.sessionId) - .where("identity.event_id", "=", messageId) - .where("active.message_position", "is not", null), - ); - if (!row || row.message_position === null) { - return undefined; - } - const visible = resolveVisibleMessagePositions(projection); - return row.message_position >= visible.postStart || visible.kept.includes(row.message_position) - ? parseMessageEventRow(row) - : undefined; - }); -} - -/** Reads a centered active-message page plus one older context row for split rendering. */ -export function readSessionTranscriptMessageAnchorPage( - scope: SessionTranscriptReadScope, - options: { maxMessages: number; messageId: string }, -): SessionTranscriptMessageAnchorPage { - return withCurrentProjectionSnapshot(scope, (projection) => { - const db = getActiveTranscriptKysely(projection.database); - const anchor = executeSqliteQueryTakeFirstSync( - projection.database.db, - db - .selectFrom("transcript_event_identities as identity") - .innerJoin("session_transcript_active_events as active", (join) => - join - .onRef("active.session_id", "=", "identity.session_id") - .onRef("active.event_seq", "=", "identity.seq"), - ) - .select("active.message_position") - .where("identity.session_id", "=", projection.resolved.sessionId) - .where("identity.event_id", "=", options.messageId) - .where("active.message_position", "is not", null), - ); - const visible = resolveVisibleMessagePositions(projection); - const totalMessages = visible.total; - if (anchor?.message_position === null || anchor?.message_position === undefined) { - return { - events: [], - found: false, - hasOverreadContext: false, - offset: 0, - totalMessages, - }; - } - const anchorVisiblePosition = - anchor.message_position >= visible.postStart - ? visible.kept.length + anchor.message_position - visible.postStart - : visible.kept.indexOf(anchor.message_position); - if (anchorVisiblePosition < 0) { - return { - events: [], - found: false, - hasOverreadContext: false, - offset: 0, - totalMessages, - }; - } - const pageSize = Math.max( - 1, - Math.floor(Number.isFinite(options.maxMessages) ? options.maxMessages : 1), - ); - const newerMessages = Math.floor(pageSize / 2); - const olderMessages = pageSize - newerMessages - 1; - const latestStart = Math.max(0, totalMessages - pageSize); - const start = Math.min(Math.max(0, anchorVisiblePosition - olderMessages), latestStart); - const endExclusive = Math.min(totalMessages, start + pageSize); - const readStart = Math.max(0, start - 1); - return { - events: readVisibleMessageRange(projection, readStart, endExclusive), - found: true, - hasOverreadContext: readStart < start, - offset: totalMessages - endExclusive, - totalMessages, - }; - }); -} diff --git a/src/config/sessions/session-accessor.sqlite-history-events.test.ts b/src/config/sessions/session-accessor.sqlite-history-events.test.ts index 18d4b2571003..3651bcc9d474 100644 --- a/src/config/sessions/session-accessor.sqlite-history-events.test.ts +++ b/src/config/sessions/session-accessor.sqlite-history-events.test.ts @@ -1,5 +1,6 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; +import { runSqliteImmediateTransactionSync } from "../../infra/sqlite-transaction.js"; import { closeOpenClawAgentDatabasesForTest, openOpenClawAgentDatabase, @@ -32,135 +33,63 @@ function enforceSqliteVariableLimit(database: OpenClawAgentDatabase): void { }); } -function insertSyntheticMessages( +function insertSyntheticHistory( database: OpenClawAgentDatabase, sessionId: string, - additionalCount: number, + count: number, + boundaries = false, ): void { - const lastSeq = additionalCount + 1; - database.db - .prepare( - `WITH RECURSIVE synthetic(seq) AS ( - SELECT 2 - UNION ALL - SELECT seq + 1 FROM synthetic WHERE seq < ? - ) - INSERT INTO transcript_events (session_id, seq, event_json, created_at) - SELECT ?, seq, - printf('{"type":"message","id":"synthetic-message-%d","parentId":null,"timestamp":"2026-08-15T00:00:00.000Z","message":{"role":"user","content":"synthetic"}}', seq), - seq - FROM synthetic`, - ) - .run(lastSeq, sessionId); - database.db - .prepare( - `WITH RECURSIVE synthetic(seq) AS ( - SELECT 2 - UNION ALL - SELECT seq + 1 FROM synthetic WHERE seq < ? - ) - INSERT INTO transcript_event_identities - (session_id, event_id, seq, event_type, parent_id, message_idempotency_key, created_at) - SELECT ?, printf('synthetic-message-%d', seq), seq, 'message', NULL, NULL, seq - FROM synthetic`, - ) - .run(lastSeq, sessionId); - database.db - .prepare( - `WITH RECURSIVE synthetic(seq) AS ( - SELECT 2 - UNION ALL - SELECT seq + 1 FROM synthetic WHERE seq < ? - ) - INSERT INTO session_transcript_active_events - (session_id, active_position, event_seq, message_position) - SELECT ?, seq - 1, seq, seq - 1 - FROM synthetic`, - ) - .run(lastSeq, sessionId); - database.db - .prepare( - `UPDATE session_transcript_index_state - SET indexed_seq = ?, leaf_event_id = ?, active_event_count = ?, active_message_count = ? - WHERE session_id = ?`, - ) - .run(lastSeq, `synthetic-message-${String(lastSeq)}`, lastSeq, lastSeq, sessionId); -} - -function insertSyntheticBoundaryPairs( - database: OpenClawAgentDatabase, - sessionId: string, - pairCount: number, -): void { - const lastSeq = pairCount * 2 + 1; - database.db - .prepare( - `WITH RECURSIVE synthetic(pair_index) AS ( - SELECT 1 - UNION ALL - SELECT pair_index + 1 FROM synthetic WHERE pair_index < ? - ) - INSERT INTO transcript_events (session_id, seq, event_json, created_at) - SELECT ?, pair_index * 2, - printf('{"type":"compaction","id":"synthetic-boundary-%d","parentId":null,"timestamp":"2026-08-15T00:00:00.000Z","summary":"synthetic"}', pair_index * 2), - pair_index * 2 - FROM synthetic - UNION ALL - SELECT ?, pair_index * 2 + 1, - printf('{"type":"message","id":"synthetic-message-%d","parentId":null,"timestamp":"2026-08-15T00:00:00.000Z","message":{"role":"user","content":"synthetic"}}', pair_index * 2 + 1), - pair_index * 2 + 1 - FROM synthetic`, - ) - .run(pairCount, sessionId, sessionId); - database.db - .prepare( - `WITH RECURSIVE synthetic(pair_index) AS ( - SELECT 1 - UNION ALL - SELECT pair_index + 1 FROM synthetic WHERE pair_index < ? - ) - INSERT INTO transcript_event_identities - (session_id, event_id, seq, event_type, parent_id, message_idempotency_key, created_at) - SELECT ?, printf('synthetic-boundary-%d', pair_index * 2), pair_index * 2, - 'compaction', NULL, NULL, pair_index * 2 - FROM synthetic - UNION ALL - SELECT ?, printf('synthetic-message-%d', pair_index * 2 + 1), pair_index * 2 + 1, - 'message', NULL, NULL, pair_index * 2 + 1 - FROM synthetic`, - ) - .run(pairCount, sessionId, sessionId); - database.db - .prepare( - `WITH RECURSIVE synthetic(pair_index) AS ( - SELECT 1 - UNION ALL - SELECT pair_index + 1 FROM synthetic WHERE pair_index < ? - ) - INSERT INTO session_transcript_active_events - (session_id, active_position, event_seq, message_position) - SELECT ?, pair_index * 2 - 1, pair_index * 2, NULL - FROM synthetic - UNION ALL - SELECT ?, pair_index * 2, pair_index * 2 + 1, pair_index - FROM synthetic`, - ) - .run(pairCount, sessionId, sessionId); - const activeEventCount = pairCount * 2 + 1; - const activeMessageCount = pairCount + 1; - database.db - .prepare( - `UPDATE session_transcript_index_state - SET indexed_seq = ?, leaf_event_id = ?, active_event_count = ?, active_message_count = ? - WHERE session_id = ?`, - ) - .run( - lastSeq, - `synthetic-message-${String(lastSeq)}`, - activeEventCount, - activeMessageCount, - sessionId, - ); + const lastSeq = count * (boundaries ? 2 : 1) + 1; + const insertEvent = database.db.prepare( + "INSERT INTO transcript_events (session_id, seq, event_json, created_at) VALUES (?, ?, ?, ?)", + ); + const insertIdentity = database.db.prepare( + `INSERT INTO transcript_event_identities + (session_id, event_id, seq, event_type, parent_id, message_idempotency_key, created_at) + VALUES (?, ?, ?, ?, NULL, NULL, ?)`, + ); + const insertActive = database.db.prepare( + `INSERT INTO session_transcript_active_events + (session_id, active_position, event_seq, message_position) + VALUES (?, ?, ?, ?)`, + ); + runSqliteImmediateTransactionSync(database.db, () => { + for (let seq = 2; seq <= lastSeq; seq += 1) { + const isBoundary = boundaries && seq % 2 === 0; + const id = `synthetic-${isBoundary ? "boundary" : "message"}-${String(seq)}`; + const type = isBoundary ? "compaction" : "message"; + const event = { + type, + id, + parentId: null, + timestamp: "2026-08-15T00:00:00.000Z", + ...(isBoundary + ? { summary: "synthetic" } + : { message: { role: "user", content: "synthetic" } }), + }; + insertEvent.run(sessionId, seq, JSON.stringify(event), seq); + insertIdentity.run(sessionId, id, seq, type, seq); + insertActive.run( + sessionId, + seq - 1, + seq, + isBoundary ? null : boundaries ? Math.floor(seq / 2) : seq - 1, + ); + } + database.db + .prepare( + `UPDATE session_transcript_index_state + SET indexed_seq = ?, leaf_event_id = ?, active_event_count = ?, active_message_count = ? + WHERE session_id = ?`, + ) + .run( + lastSeq, + `synthetic-message-${String(lastSeq)}`, + lastSeq, + boundaries ? count + 1 : lastSeq, + sessionId, + ); + }); } describe("SQLite transcript history events", () => { @@ -306,7 +235,7 @@ describe("SQLite transcript history events", () => { }); const database = openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env }); const bindingCount = REGRESSION_SQLITE_VARIABLE_LIMIT; - insertSyntheticMessages(database, scope.sessionId, bindingCount); + insertSyntheticHistory(database, scope.sessionId, bindingCount); enforceSqliteVariableLimit(database); const page = readRecentSessionTranscriptHistoryEvents(scope, { @@ -332,7 +261,7 @@ describe("SQLite transcript history events", () => { }); const database = openOpenClawAgentDatabase({ agentId: scope.agentId, env: scope.env }); const bindingCount = REGRESSION_SQLITE_VARIABLE_LIMIT; - insertSyntheticBoundaryPairs(database, scope.sessionId, bindingCount); + insertSyntheticHistory(database, scope.sessionId, bindingCount, true); enforceSqliteVariableLimit(database); const events = readSessionTranscriptHistoryEvents(scope); diff --git a/src/config/sessions/session-accessor.sqlite-history-events.ts b/src/config/sessions/session-accessor.sqlite-history-events.ts index 4cacecfbd6d1..773f3ca58ec2 100644 --- a/src/config/sessions/session-accessor.sqlite-history-events.ts +++ b/src/config/sessions/session-accessor.sqlite-history-events.ts @@ -42,6 +42,9 @@ type VisibleHistoryProjection = { function resolveVisibleHistoryProjection( projection: CurrentTranscriptProjection, ): VisibleHistoryProjection { + if (projection.state.activeEventCount === projection.state.activeMessageCount) { + return { boundaries: [], total: projection.state.activeMessageCount }; + } const visibleMessages = resolveVisibleMessagePositions(projection); const db = getActiveTranscriptKysely(projection.database); const rows = executeSqliteQuerySync( diff --git a/src/config/sessions/session-accessor.sqlite-message-cut.test.ts b/src/config/sessions/session-accessor.sqlite-message-cut.test.ts index 251ddb72364a..afffb8e5f028 100644 --- a/src/config/sessions/session-accessor.sqlite-message-cut.test.ts +++ b/src/config/sessions/session-accessor.sqlite-message-cut.test.ts @@ -17,7 +17,7 @@ import { listSessionBranches, loadSessionEntry, loadTranscriptEvents, - readSessionTranscriptMessageEventCount, + readSessionTranscriptMessageEventPage, readSessionTranscriptMessageEvents, rewindSessionToMessage, switchSessionBranch, @@ -480,7 +480,10 @@ describe("SQLite session message cuts", () => { throw new Error("expected rewind result"); } expect( - readSessionTranscriptMessageEventCount({ agentId, env, sessionId: result.entry.sessionId }), + readSessionTranscriptMessageEventPage( + { agentId, env, sessionId: result.entry.sessionId }, + { maxMessages: 0, offset: 0 }, + ).totalMessages, ).toBe(2); expect(loadSessionEntry({ agentId, env, sessionKey })?.sessionId).toBe(result.entry.sessionId); expect(result.entry).toMatchObject({ diff --git a/src/config/sessions/session-accessor.ts b/src/config/sessions/session-accessor.ts index f813f0ed31bf..10d3c2495cc3 100644 --- a/src/config/sessions/session-accessor.ts +++ b/src/config/sessions/session-accessor.ts @@ -269,9 +269,6 @@ export { readSessionTranscriptBoundedMessageTailPage, readRecentSessionTranscriptMessageEvents, readSessionTranscriptActivePathEntryRelation, - readSessionTranscriptMessageAnchorPage, - readSessionTranscriptMessageEventById, - readSessionTranscriptMessageEventCount, readSessionTranscriptMessageEventPage, readSessionTranscriptMessageEvents, readSessionTranscriptVisibleMessageDeltaCore, diff --git a/src/gateway/session-history-state.ts b/src/gateway/session-history-state.ts index 023a94c1eae3..ead31986809a 100644 --- a/src/gateway/session-history-state.ts +++ b/src/gateway/session-history-state.ts @@ -1,6 +1,7 @@ // Gateway session-history projection state. // Tracks transcript sequence windows for paginated chat-history SSE updates. import { isDeepStrictEqual } from "node:util"; +import { expectDefined } from "@openclaw/normalization-core"; import { asPositiveSafeInteger } from "@openclaw/normalization-core/number-coercion"; import type { SessionEntry } from "../config/sessions.js"; import { @@ -10,13 +11,15 @@ import { } 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 { + readChatHistoryMessageSeq as resolveMessageSeq, + readIncrementalChatHistoryTail, +} from "./session-history-tail.js"; import { resolveTranscriptPathForComparison } from "./session-transcript-path.js"; import { attachOpenClawTranscriptMeta, readSessionMessagesPageWithStatsAsync, readSessionMessagesWithSourceAsync, - type ReadRecentSessionMessagesResult, } from "./session-transcript-readers.js"; // Session history state owns the SSE-friendly projection of transcript JSONL: @@ -83,17 +86,18 @@ function readMessageIdempotencyKey(message: unknown): string | undefined { return typeof value === "string" && value.trim() ? value : undefined; } -/** Shares the bounded visible-message scanner across HTTP snapshots and SSE refreshes. */ -export async function readBoundedSessionHistorySnapshotAsync(params: { - cursor?: string; - target: SessionHistoryTranscriptTarget; - limit: number; - maxChars: number; -}): Promise< - ReadRecentSessionMessagesResult & { - projection: ReturnType; +/** Owns both complete history snapshots and bounded visible-message pages. */ +export async function readSessionHistoryRawSnapshotAsync( + params: Pick, +): Promise { + if (typeof params.limit !== "number") { + const snapshot = await readSessionMessagesWithSourceAsync(params.target, { + mode: "full", + reason: "session history cursor pagination", + allowResetArchiveFallback: true, + }); + return { rawMessages: snapshot.messages, transcriptPath: snapshot.transcriptPath }; } -> { const cursorSeq = resolveCursorSeq(params.cursor); const offset = cursorSeq === undefined @@ -113,13 +117,19 @@ export async function readBoundedSessionHistorySnapshotAsync(params: { const tail = await readIncrementalChatHistoryTail({ entry: params.target.sessionEntry, readScope: params.target, - effectiveMaxChars: params.maxChars, + effectiveMaxChars: params.maxChars ?? DEFAULT_CHAT_HISTORY_TEXT_MAX_CHARS, max: params.limit, maxBytes: getMaxChatHistoryMessagesBytes(), ...(offset === undefined ? {} : { offset }), preserveProjectionContext: true, }); - return { ...tail.readPage, messages: tail.rawMessages, projection: tail.projection }; + return { + projection: tail.projection, + rawMessages: tail.rawMessages, + rawTranscriptSeq: tail.readPage.totalMessages, + totalRawMessages: tail.readPage.totalMessages, + transcriptPath: tail.readPage.transcriptPath, + }; } export function resolveCursorSeq(cursor: string | undefined): number | undefined { @@ -154,10 +164,6 @@ function buildPaginatedSessionHistory(params: { }; } -function resolveMessageSeq(message: SessionHistoryMessage | undefined): number | undefined { - return asPositiveSafeInteger(message?.["__openclaw"]?.seq); -} - function isMessageToolMirrorMessage(message: SessionHistoryMessage): boolean { return message.openclawMessageToolMirror !== undefined; } @@ -229,7 +235,7 @@ export function buildSessionHistorySnapshot(params: { maxChars: params.maxChars ?? DEFAULT_CHAT_HISTORY_TEXT_MAX_CHARS, resolveCurrentUserProfileDisplay, }); - const visibleMessages = toSessionHistoryMessages(projected.messages); + const visibleMessages = projected.messages; const rawHistoryMessages = toSessionHistoryMessages(params.rawMessages); const history = paginateSessionMessages(visibleMessages, params.limit, params.cursor); if ( @@ -343,12 +349,13 @@ export class SessionHistorySseState { // Projection can split, drop, or rewrite raw transcript messages. When one // raw append changes multiple visible rows, callers must refresh instead of // emitting a misleading single SSE item. - const projectedMessages = toSessionHistoryMessages( - projectChatDisplayMessages([...this.sentHistory.messages, nextMessage], { + const projectedMessages = projectChatDisplayMessages( + [...this.sentHistory.messages, nextMessage], + { includeCommentaryFallbacks: true, maxChars: this.maxChars, resolveCurrentUserProfileDisplay, - }), + }, ); const projectedPrefix = projectedMessages.slice(0, this.sentHistory.messages.length); if ( @@ -366,7 +373,7 @@ export class SessionHistorySseState { } if (projectedMessages.length > this.sentHistory.messages.length) { const addedMessages = projectedMessages.slice(this.sentHistory.messages.length); - if (hadPendingTurnBoundary && !this.turnBoundaryPending && addedMessages[0]) { + if (hadPendingTurnBoundary && !this.turnBoundaryPending) { const firstAdded = attachOpenClawTranscriptMeta(addedMessages[0], { turnBoundary: true, }) as SessionHistoryMessage; @@ -380,54 +387,31 @@ export class SessionHistorySseState { }); return { shouldRefresh: true }; } - const projectedMessage = addedMessages[0]; - if (projectedMessage !== undefined) { - const emittedMessage: SessionHistoryMessage = - isMessageToolMirrorMessage(projectedMessage) || - resolveMessageSeq(projectedMessage) === undefined - ? (attachOpenClawTranscriptMeta(projectedMessage, { - seq: this.rawTranscriptSeq, - }) as SessionHistoryMessage) - : projectedMessage; - const nextMessages = [...this.sentHistory.messages, emittedMessage]; - this.sentHistory = buildPaginatedSessionHistory({ - messages: nextMessages, - hasMore: false, - }); - return { - message: emittedMessage, - messageSeq: resolveMessageSeq(emittedMessage), - }; - } - } - const [sanitizedMessage] = toSessionHistoryMessages(nextProjection.messages); - if (!sanitizedMessage) { - if (projectedMessages.length < this.sentHistory.messages.length) { - this.sentHistory = buildPaginatedSessionHistory({ - messages: projectedMessages, - hasMore: false, - }); - return { shouldRefresh: true }; - } - return null; - } - if (projectedMessages.length <= this.sentHistory.messages.length) { + const projectedMessage = expectDefined(addedMessages[0], "projected inline message"); + const emittedMessage: SessionHistoryMessage = + isMessageToolMirrorMessage(projectedMessage) || + resolveMessageSeq(projectedMessage) === undefined + ? (attachOpenClawTranscriptMeta(projectedMessage, { + seq: this.rawTranscriptSeq, + }) as SessionHistoryMessage) + : projectedMessage; this.sentHistory = buildPaginatedSessionHistory({ - messages: projectedMessages, + messages: [...this.sentHistory.messages, emittedMessage], hasMore: false, }); - return { shouldRefresh: true }; + return { message: emittedMessage, messageSeq: resolveMessageSeq(emittedMessage) }; + } + if ( + nextProjection.messages.length === 0 && + projectedMessages.length === this.sentHistory.messages.length + ) { + return null; } - const projectedMessage = projectedMessages.at(-1) ?? sanitizedMessage; - const nextMessages = [...this.sentHistory.messages, projectedMessage]; this.sentHistory = buildPaginatedSessionHistory({ - messages: nextMessages, + messages: projectedMessages, hasMore: false, }); - return { - message: projectedMessage, - messageSeq: resolveMessageSeq(projectedMessage), - }; + return { shouldRefresh: true }; } shouldRefreshForTranscriptPath(updatePath: string | undefined): boolean { @@ -436,7 +420,12 @@ export class SessionHistorySseState { } async refreshAsync(): Promise { - const rawSnapshot = await this.readRawSnapshotAsync(); + const rawSnapshot = await readSessionHistoryRawSnapshotAsync({ + target: this.target, + maxChars: this.maxChars, + limit: this.limit, + cursor: this.cursor, + }); const snapshot = this.buildSnapshot(rawSnapshot); this.rawTranscriptSeq = snapshot.rawTranscriptSeq; this.turnBoundaryPending = snapshot.turnBoundaryPending; @@ -457,42 +446,6 @@ export class SessionHistorySseState { totalRawMessages: rawSnapshot.totalRawMessages, }); } - - private async readRawSnapshotAsync(): Promise { - if (typeof this.limit === "number") { - const snapshot = await readBoundedSessionHistorySnapshotAsync({ - cursor: this.cursor, - target: this.target, - limit: this.limit, - maxChars: this.maxChars, - }); - return { - projection: snapshot.projection, - rawMessages: snapshot.messages, - rawTranscriptSeq: snapshot.totalMessages, - totalRawMessages: snapshot.totalMessages, - transcriptPath: snapshot.transcriptPath, - }; - } - const snapshot = await readSessionMessagesWithSourceAsync( - { - agentId: this.target.agentId, - sessionEntry: this.target.sessionEntry, - sessionId: this.target.sessionId, - sessionKey: this.target.sessionKey, - storePath: this.target.storePath, - }, - { - mode: "full", - reason: "session history cursor pagination", - allowResetArchiveFallback: true, - }, - ); - return { - rawMessages: snapshot.messages, - transcriptPath: snapshot.transcriptPath, - }; - } } function normalizeTranscriptPathForComparison(filePath: string | undefined): string | undefined { diff --git a/src/gateway/session-transcript-readers.ts b/src/gateway/session-transcript-readers.ts index 22c0522306d6..9a27a3993351 100644 --- a/src/gateway/session-transcript-readers.ts +++ b/src/gateway/session-transcript-readers.ts @@ -156,12 +156,6 @@ function projectSqliteHistoryEvents(entries: readonly SessionTranscriptMessageEv }); } -async function readSqliteHistoryMessages(target: ResolvedTranscriptReadTarget): Promise { - return projectSqliteHistoryEvents( - readSessionTranscriptHistoryEvents(toTranscriptReadScope(target)), - ); -} - function readSqliteMessagesSync(target: ResolvedTranscriptReadTarget): unknown[] { return readSqliteMessageRecords(target).map(sqliteRecordMessageWithSeq); } @@ -260,21 +254,7 @@ export async function readSessionMessagesAsync( scope: SessionTranscriptReadScope, opts: ReadSessionMessagesAsyncOptions, ): Promise { - const target = resolveTranscriptReadTarget(scope); - if (opts.mode === "recent") { - const { messages } = await readRecentSqliteMessageRecords(target, opts); - if (messages.length === 0 && opts.allowResetArchiveFallback === true) { - return (await archivedTranscriptReader(target).read({ ...opts, resetArchiveOnly: true })) - .messages; - } - return messages; - } - const messages = await readSqliteHistoryMessages(target); - if (messages.length === 0 && opts.allowResetArchiveFallback === true) { - return (await archivedTranscriptReader(target).read({ ...opts, resetArchiveOnly: true })) - .messages; - } - return messages; + return (await readSessionMessagesWithSourceAsync(scope, opts)).messages; } /** Reads display messages with source metadata through the reader seam. */ @@ -286,7 +266,9 @@ export async function readSessionMessagesWithSourceAsync( const messages = opts.mode === "recent" ? (await readRecentSqliteMessageRecords(target, opts)).messages - : await readSqliteHistoryMessages(target); + : projectSqliteHistoryEvents( + readSessionTranscriptHistoryEvents(toTranscriptReadScope(target)), + ); if (messages.length === 0 && opts.allowResetArchiveFallback === true) { return await archivedTranscriptReader(target).read({ ...opts, resetArchiveOnly: true }); } diff --git a/src/gateway/sessions-history-http.revocation.test.ts b/src/gateway/sessions-history-http.revocation.test.ts index 12d314a9245a..2f42bf29ada6 100644 --- a/src/gateway/sessions-history-http.revocation.test.ts +++ b/src/gateway/sessions-history-http.revocation.test.ts @@ -120,32 +120,16 @@ vi.mock("./session-utils.js", () => ({ resolveSessionTranscriptCandidates: () => ["/tmp/session-1.jsonl"], })); -vi.mock("./session-transcript-readers.js", () => ({ - readRecentSessionMessagesWithStatsAsync: async () => { - if (transcriptReadError) { - throw transcriptReadError; - } - return { messages: [], totalMessages: 0 }; - }, - readSessionMessagesAsync: async () => [], - readSessionMessagesWithSourceAsync: async () => { - if (transcriptReadError) { - throw transcriptReadError; - } - return { messages: [] }; - }, -})); - vi.mock("./session-history-state.js", () => ({ buildSessionHistorySnapshot: () => ({ history: { items: [], nextCursor: null, messages: [] }, }), resolveCursorSeq: (_cursor: string | undefined) => undefined, - readBoundedSessionHistorySnapshotAsync: async () => { + readSessionHistoryRawSnapshotAsync: async () => { if (transcriptReadError) { throw transcriptReadError; } - return { messages: [], totalMessages: 0 }; + return { rawMessages: [] }; }, SessionHistorySseState: { fromRawSnapshot: (_params: unknown) => ({ diff --git a/src/gateway/sessions-history-http.test.ts b/src/gateway/sessions-history-http.test.ts index c67f4fd286f5..b4029d453e02 100644 --- a/src/gateway/sessions-history-http.test.ts +++ b/src/gateway/sessions-history-http.test.ts @@ -235,16 +235,11 @@ async function appendVisibleAssistantMessage(params: { text: string; storePath: string; }) { - const appended = await appendExactAssistantMessageToSessionTranscript({ + return await appendTranscriptMessage({ sessionKey: params.sessionKey, storePath: params.storePath, message: makeTranscriptAssistantMessage({ text: params.text }), }); - expect(appended.ok).toBe(true); - if (!appended.ok) { - throw new Error(`append failed: ${appended.reason}`); - } - return appended.messageId; } async function fetchSessionHistory( diff --git a/src/gateway/sessions-history-http.ts b/src/gateway/sessions-history-http.ts index 0e700a638ad8..cd8772b948dc 100644 --- a/src/gateway/sessions-history-http.ts +++ b/src/gateway/sessions-history-http.ts @@ -36,13 +36,12 @@ import { authorizeOperatorScopesForMethod } from "./method-scopes.js"; import type { GatewayClient } from "./server-methods/shared-types.js"; import { buildSessionHistorySnapshot, - readBoundedSessionHistorySnapshotAsync, + readSessionHistoryRawSnapshotAsync, resolveCursorSeq, SessionHistorySseState, } from "./session-history-state.js"; import { createSessionListEntryFilter, resolveSessionSharingTarget } from "./session-sharing.js"; import { resolveTranscriptPathForComparison } from "./session-transcript-path.js"; -import { readSessionMessagesWithSourceAsync } from "./session-transcript-readers.js"; import { resolveCanonicalSessionEntryFromStoreKeys, resolveGatewaySessionStoreTargetWithStore, @@ -225,29 +224,14 @@ export async function handleSessionHistoryHttpRequest( sessionKey: target.canonicalKey, storePath: target.storePath, }; - let boundedSnapshot: - | Awaited> - | undefined; - let fullSnapshot: Awaited> | undefined; + let rawSnapshot: Awaited>; try { - boundedSnapshot = - typeof limit === "number" - ? await readBoundedSessionHistorySnapshotAsync({ - cursor, - target: historyTarget, - limit, - maxChars: effectiveMaxChars, - }) - : undefined; - // Requests without a limit preserve the public complete-history contract. - fullSnapshot = - boundedSnapshot === undefined - ? await readSessionMessagesWithSourceAsync(historyTarget, { - mode: "full", - reason: "session history cursor pagination", - allowResetArchiveFallback: true, - }) - : undefined; + rawSnapshot = await readSessionHistoryRawSnapshotAsync({ + cursor, + target: historyTarget, + limit, + maxChars: effectiveMaxChars, + }); } catch (error) { if (!isSessionTranscriptProjectionUnavailableError(error)) { throw error; @@ -263,17 +247,9 @@ export async function handleSessionHistoryHttpRequest( }); return true; } - const rawSnapshot = boundedSnapshot?.messages ?? fullSnapshot?.messages ?? []; + const historySnapshot = { ...rawSnapshot, maxChars: effectiveMaxChars, limit, cursor }; if (!shouldStreamSse(req)) { - const history = buildSessionHistorySnapshot({ - projection: boundedSnapshot?.projection, - rawMessages: rawSnapshot, - maxChars: effectiveMaxChars, - limit, - cursor, - rawTranscriptSeq: boundedSnapshot?.totalMessages, - totalRawMessages: boundedSnapshot?.totalMessages, - }).history; + const history = buildSessionHistorySnapshot(historySnapshot).history; sendJson(res, 200, { sessionKey: target.canonicalKey, ...history, @@ -293,15 +269,8 @@ export async function handleSessionHistoryHttpRequest( ); const sseState = SessionHistorySseState.fromRawSnapshot({ - projection: boundedSnapshot?.projection, + ...historySnapshot, target: historyTarget, - rawMessages: rawSnapshot, - rawTranscriptSeq: boundedSnapshot?.totalMessages, - totalRawMessages: boundedSnapshot?.totalMessages, - transcriptPath: boundedSnapshot?.transcriptPath ?? fullSnapshot?.transcriptPath, - maxChars: effectiveMaxChars, - limit, - cursor, }); let sentHistory = sseState.snapshot(); let streamStopped = false;