From 074f2dcfc87d96ec7b5ce6c63d77ed6a526ec0af Mon Sep 17 00:00:00 2001 From: "Jason (Json)" <263060202+fuller-stack-dev@users.noreply.github.com> Date: Sun, 2 Aug 2026 21:01:10 -0600 Subject: [PATCH] perf(gateway): bound sessions.list transcript validation (#118380) --- .../sessions/session-accessor.sqlite-scope.ts | 57 ++++++++-- .../session-accessor.sqlite-title-probes.ts | 4 +- ...on-accessor.sqlite-transcript-watermark.ts | 100 +++++++++++++++++- src/config/sessions/session-accessor.test.ts | 44 ++++++++ .../session-accessor.transcript-target.ts | 27 ++--- src/config/sessions/session-accessor.ts | 1 + .../session-transcript-readers.test.ts | 36 +++++++ .../session-transcript-title-reader.ts | 41 +++++-- src/gateway/session-utils.perf.test.ts | 10 ++ 9 files changed, 290 insertions(+), 30 deletions(-) diff --git a/src/config/sessions/session-accessor.sqlite-scope.ts b/src/config/sessions/session-accessor.sqlite-scope.ts index 040d008bedb8..12203c4abb77 100644 --- a/src/config/sessions/session-accessor.sqlite-scope.ts +++ b/src/config/sessions/session-accessor.sqlite-scope.ts @@ -71,6 +71,11 @@ type ResolvedTranscriptReadScope = ResolvedSqliteReadScope & { sessionId: string; }; +export type SessionSqliteTargetResolutionCache = Map< + NodeJS.ProcessEnv | undefined, + Map> +>; + const SQLITE_SESSION_SLOW_WRITE_MS = 1_000; const SQLITE_SESSION_WRITER_QUEUES = new Map(); @@ -165,6 +170,7 @@ export function resolveSqliteReadScope( SessionTranscriptReadScope, "agentId" | "defaultAgentId" | "env" | "sessionKey" | "storePath" >, + targetCache?: SessionSqliteTargetResolutionCache, ): ResolvedSqliteReadScope { const sessionKey = scope.sessionKey ? normalizeSqliteSessionKey(scope.sessionKey) : undefined; const parsedAgentId = parseAgentSessionKey(sessionKey)?.agentId; @@ -177,11 +183,15 @@ export function resolveSqliteReadScope( : scope.storePath; const effectiveAgentId = incognitoAgentId ?? scopedAgentId; const storeTarget = effectiveStorePath - ? resolveSqliteTargetFromSessionStorePath(effectiveStorePath, { - agentId: effectiveAgentId, - defaultAgentId: scope.defaultAgentId, - ...(scope.env ? { env: scope.env } : {}), - }) + ? resolveCachedSqliteStoreTarget( + { + agentId: effectiveAgentId, + defaultAgentId: scope.defaultAgentId, + env: scope.env, + storePath: effectiveStorePath, + }, + targetCache, + ) : undefined; const agentId = resolveSqliteAgentId({ scopedAgentId: effectiveAgentId, @@ -201,6 +211,40 @@ export function resolveSqliteReadScope( }; } +function resolveCachedSqliteStoreTarget( + params: { + agentId?: string; + defaultAgentId?: string; + env?: NodeJS.ProcessEnv; + storePath: string; + }, + targetCache: SessionSqliteTargetResolutionCache | undefined, +): ReturnType { + if (!targetCache) { + return resolveSqliteTargetFromSessionStorePath(params.storePath, { + agentId: params.agentId, + defaultAgentId: params.defaultAgentId, + ...(params.env ? { env: params.env } : {}), + }); + } + // Store ownership is stable for this batch. Scope the cache to the caller so later requests + // still observe owner changes after migration, install, or doctor flows. + const envCache = targetCache.get(params.env) ?? new Map(); + targetCache.set(params.env, envCache); + const cacheKey = JSON.stringify([params.storePath, params.agentId, params.defaultAgentId]); + const cached = envCache.get(cacheKey); + if (cached) { + return cached; + } + const resolved = resolveSqliteTargetFromSessionStorePath(params.storePath, { + agentId: params.agentId, + defaultAgentId: params.defaultAgentId, + ...(params.env ? { env: params.env } : {}), + }); + envCache.set(cacheKey, resolved); + return resolved; +} + export function resolveSqliteStoreScope( storePath: string, options: { agentId?: string } = {}, @@ -273,9 +317,10 @@ export function resolveSqliteTranscriptReadScope( SessionTranscriptReadScope, "agentId" | "env" | "sessionId" | "sessionKey" | "storePath" >, + targetCache?: SessionSqliteTargetResolutionCache, ): ResolvedTranscriptReadScope { return { - ...resolveSqliteReadScope(scope), + ...resolveSqliteReadScope(scope, targetCache), sessionId: scope.sessionId, }; } diff --git a/src/config/sessions/session-accessor.sqlite-title-probes.ts b/src/config/sessions/session-accessor.sqlite-title-probes.ts index e66f5abb3b42..1a369470132d 100644 --- a/src/config/sessions/session-accessor.sqlite-title-probes.ts +++ b/src/config/sessions/session-accessor.sqlite-title-probes.ts @@ -13,6 +13,7 @@ import type { import { resolveSqliteTranscriptReadScope, toDatabaseOptions, + type SessionSqliteTargetResolutionCache, } from "./session-accessor.sqlite-scope.js"; type TitleProbeDatabase = Pick< @@ -180,8 +181,9 @@ export function readSessionTranscriptTitleProbeBatch( string, { database: OpenClawAgentDatabase; items: Array<{ index: number; sessionId: string }> } >(); + const targetCache: SessionSqliteTargetResolutionCache = new Map(); for (const [index, scope] of scopes.entries()) { - const resolved = resolveSqliteTranscriptReadScope(scope); + const resolved = resolveSqliteTranscriptReadScope(scope, targetCache); const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); const group = groups.get(database.path) ?? { database, items: [] }; group.items.push({ index, sessionId: resolved.sessionId }); diff --git a/src/config/sessions/session-accessor.sqlite-transcript-watermark.ts b/src/config/sessions/session-accessor.sqlite-transcript-watermark.ts index 1e43802ab46f..2212e4a13f9c 100644 --- a/src/config/sessions/session-accessor.sqlite-transcript-watermark.ts +++ b/src/config/sessions/session-accessor.sqlite-transcript-watermark.ts @@ -2,18 +2,26 @@ // validates transcript-derived caches (derived titles, branch summaries). // Kept apart from the active-events reader so cache validation stays a // dependency-light import for gateway callers. -import { executeSqliteQueryTakeFirstSync, getNodeSqliteKysely } from "../../infra/kysely-sync.js"; +import { + executeSqliteQuerySync, + executeSqliteQueryTakeFirstSync, + getNodeSqliteKysely, +} from "../../infra/kysely-sync.js"; import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js"; -import { openOpenClawAgentDatabase } from "../../state/openclaw-agent-db.js"; +import { + openOpenClawAgentDatabase, + type OpenClawAgentDatabase, +} from "../../state/openclaw-agent-db.js"; import type { SessionTranscriptReadScope } from "./session-accessor.sqlite-contract.js"; import { resolveSqliteTranscriptReadScope, toDatabaseOptions, + type SessionSqliteTargetResolutionCache, } from "./session-accessor.sqlite-scope.js"; type WatermarkDatabase = Pick< OpenClawAgentKyselyDatabase, - "transcript_events" | "transcript_rewrite_watermarks" + "session_windows" | "transcript_events" | "transcript_rewrite_watermarks" >; export type SessionTranscriptWatermark = { @@ -21,6 +29,8 @@ export type SessionTranscriptWatermark = { maxSeq: number | null; }; +const SESSION_TRANSCRIPT_WATERMARK_QUERY_CHUNK_SIZE = 400; + /** Reads the append and rewrite tokens that validate transcript-derived caches. */ export function readSessionTranscriptWatermark( scope: SessionTranscriptReadScope, @@ -44,3 +54,87 @@ export function readSessionTranscriptWatermark( )?.generation; return { generation: generation ?? null, maxSeq: maxSeq ?? null }; } + +function readSessionTranscriptWatermarkChunk( + database: OpenClawAgentDatabase, + sessionIds: readonly string[], +): Map { + const db = getNodeSqliteKysely(database.db); + const rows = executeSqliteQuerySync( + database.db, + db + .selectFrom("session_windows as window") + .leftJoin( + "transcript_rewrite_watermarks as rewrite", + "rewrite.session_id", + "window.session_id", + ) + .select((eb) => [ + "window.session_id", + "rewrite.generation", + eb + .selectFrom("transcript_events as event") + .select((inner) => inner.fn.max("event.seq").as("max_seq")) + .whereRef("event.session_id", "=", "window.session_id") + .as("max_seq"), + ]) + .where("window.session_id", "in", sessionIds), + ).rows; + return new Map( + rows.map((row) => [ + row.session_id, + { + generation: row.generation ?? null, + maxSeq: row.max_seq ?? null, + }, + ]), + ); +} + +/** Reads cache-validation tokens in one statement per opened store and SQLite-sized chunk. */ +export function readSessionTranscriptWatermarkBatch( + scopes: readonly SessionTranscriptReadScope[], +): SessionTranscriptWatermark[] { + const results: Array = Array.from({ + length: scopes.length, + }); + const groups = new Map< + string, + { database: OpenClawAgentDatabase; items: Array<{ index: number; sessionId: string }> } + >(); + const targetCache: SessionSqliteTargetResolutionCache = new Map(); + for (const [index, scope] of scopes.entries()) { + const resolved = resolveSqliteTranscriptReadScope(scope, targetCache); + const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); + const group = groups.get(database.path) ?? { database, items: [] }; + group.items.push({ index, sessionId: resolved.sessionId }); + groups.set(database.path, group); + } + for (const group of groups.values()) { + const sessionIds = [...new Set(group.items.map((item) => item.sessionId))]; + const watermarks = new Map(); + for ( + let offset = 0; + offset < sessionIds.length; + offset += SESSION_TRANSCRIPT_WATERMARK_QUERY_CHUNK_SIZE + ) { + const chunk = sessionIds.slice( + offset, + offset + SESSION_TRANSCRIPT_WATERMARK_QUERY_CHUNK_SIZE, + ); + for (const [sessionId, watermark] of readSessionTranscriptWatermarkChunk( + group.database, + chunk, + )) { + watermarks.set(sessionId, watermark); + } + } + for (const item of group.items) { + results[item.index] = watermarks.get(item.sessionId) ?? { + generation: null, + maxSeq: null, + }; + } + } + return results.map((result) => result ?? { generation: null, maxSeq: null }); +} diff --git a/src/config/sessions/session-accessor.test.ts b/src/config/sessions/session-accessor.test.ts index 7b64da17cc4c..3e7a3ef45eda 100644 --- a/src/config/sessions/session-accessor.test.ts +++ b/src/config/sessions/session-accessor.test.ts @@ -3644,5 +3644,49 @@ describe("session accessor seam", () => { storePath, }); }); + + it("preserves a matching preloaded entry identity without rereading the session row", () => { + const sessionKey = "agent:main:preloaded-read"; + const target = resolveSessionTranscriptReadTarget({ + agentId: "main", + sessionEntry: { sessionId: "preloaded-session" }, + sessionId: "preloaded-session", + sessionKey, + storePath, + }); + + expect(target).toEqual({ + agentId: "main", + sessionId: "preloaded-session", + sessionKey, + storePath, + }); + }); + + it("does not trust a preloaded entry for a different session id", async () => { + const sessionKey = "agent:main:mismatched-preloaded-read"; + await upsertSessionEntry( + { sessionKey, storePath }, + { + sessionId: "stored-session", + updatedAt: 10, + }, + ); + + const target = resolveSessionTranscriptReadTarget({ + agentId: "main", + sessionEntry: { sessionId: "different-session" }, + sessionId: "stored-session", + sessionKey, + storePath, + }); + + expect(target).toEqual({ + agentId: "main", + sessionId: "stored-session", + sessionKey, + storePath, + }); + }); }); /* oxlint-disable max-lines -- TODO: split this grandfathered oversized file. */ diff --git a/src/config/sessions/session-accessor.transcript-target.ts b/src/config/sessions/session-accessor.transcript-target.ts index ea8507e5b270..d94d695441d7 100644 --- a/src/config/sessions/session-accessor.transcript-target.ts +++ b/src/config/sessions/session-accessor.transcript-target.ts @@ -77,22 +77,25 @@ export function resolveSessionTranscriptReadTarget( sessionKey, storePath: configuredStorePath, }); - const resolved = sessionKey - ? resolveSessionEntrySelection( - { - agentId, - ...(scope.env ? { env: scope.env } : {}), - sessionKey, - storePath, - }, - { readOnly: true }, - ) - : undefined; + const hasMatchingSessionEntry = scope.sessionEntry?.sessionId === scope.sessionId; + const resolved = + sessionKey && !hasMatchingSessionEntry + ? resolveSessionEntrySelection( + { + agentId, + ...(scope.env ? { env: scope.env } : {}), + sessionKey, + storePath, + }, + { readOnly: true }, + ) + : undefined; + const resolvedSessionKey = hasMatchingSessionEntry ? sessionKey : resolved?.normalizedKey; return { agentId, sessionId: scope.sessionId, storePath, - ...(resolved?.normalizedKey ? { sessionKey: resolved.normalizedKey } : {}), + ...(resolvedSessionKey ? { sessionKey: resolvedSessionKey } : {}), }; } diff --git a/src/config/sessions/session-accessor.ts b/src/config/sessions/session-accessor.ts index 77689cdf11d8..98ef3ac8cc21 100644 --- a/src/config/sessions/session-accessor.ts +++ b/src/config/sessions/session-accessor.ts @@ -245,6 +245,7 @@ export type { } from "./session-accessor.sqlite-active-events.js"; export { readSessionTranscriptWatermark, + readSessionTranscriptWatermarkBatch, type SessionTranscriptWatermark, } from "./session-accessor.sqlite-transcript-watermark.js"; export { diff --git a/src/gateway/session-transcript-readers.test.ts b/src/gateway/session-transcript-readers.test.ts index 2db0345e2292..3b13d3c86d00 100644 --- a/src/gateway/session-transcript-readers.test.ts +++ b/src/gateway/session-transcript-readers.test.ts @@ -1,5 +1,6 @@ import fs from "node:fs"; import path from "node:path"; +import { DatabaseSync } from "node:sqlite"; import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js"; import * as sessionAccessor from "../config/sessions/session-accessor.js"; @@ -36,6 +37,8 @@ vi.mock("../config/sessions/session-accessor.js", async (importOriginal) => { readSessionTranscriptMessageEventPage: vi.fn(actual.readSessionTranscriptMessageEventPage), readSessionTranscriptMessageEvents: vi.fn(actual.readSessionTranscriptMessageEvents), readSessionTranscriptTitleProbeBatch: vi.fn(actual.readSessionTranscriptTitleProbeBatch), + readSessionTranscriptWatermark: vi.fn(actual.readSessionTranscriptWatermark), + readSessionTranscriptWatermarkBatch: vi.fn(actual.readSessionTranscriptWatermarkBatch), }; }); @@ -500,10 +503,41 @@ describe("session transcript reader facade", () => { expect(readSessionTitleFieldsFromTranscriptBatch([scope])).toEqual([ { firstUserMessage: "cached batch prompt", lastMessagePreview: "cached batch reply" }, ]); + expect(sessionAccessor.readSessionTranscriptWatermarkBatch).toHaveBeenCalledOnce(); + expect(sessionAccessor.readSessionTranscriptWatermark).not.toHaveBeenCalled(); expect(sessionAccessor.readSessionTranscriptTitleProbeBatch).not.toHaveBeenCalled(); expect(sessionAccessor.readSessionTranscriptMessageEventPage).not.toHaveBeenCalled(); }); + test("resolves SQLite store ownership once for a multi-row transcript batch", async () => { + const scopes: SessionTranscriptReadScope[] = []; + for (let index = 0; index < 30; index += 1) { + scopes.push( + await writeSqliteMessages(`reader-title-target-batch-${index}`, [ + { role: "user", content: `prompt ${index}` }, + { role: "assistant", content: `reply ${index}` }, + ]), + ); + } + const prepareSpy = vi.spyOn(DatabaseSync.prototype, "prepare"); + try { + expect(sessionAccessor.readSessionTranscriptTitleProbeBatch(scopes)).toHaveLength(30); + const titleSchemaReads = prepareSpy.mock.calls.filter(([sql]) => + sql.toLowerCase().includes("pragma user_version"), + ); + expect(titleSchemaReads).toHaveLength(1); + + prepareSpy.mockClear(); + expect(sessionAccessor.readSessionTranscriptWatermarkBatch(scopes)).toHaveLength(30); + const watermarkSchemaReads = prepareSpy.mock.calls.filter(([sql]) => + sql.toLowerCase().includes("pragma user_version"), + ); + expect(watermarkSchemaReads).toHaveLength(1); + } finally { + prepareSpy.mockRestore(); + } + }); + test("reprobes cached batch title fields after an append advances max seq", async () => { const sessionId = "reader-title-batch-cache-append"; const scope = await writeSqliteMessages(sessionId, [ @@ -525,6 +559,8 @@ describe("session transcript reader facade", () => { expect(readSessionTitleFieldsFromTranscriptBatch([scope])[0]?.lastMessagePreview).toBe( "appended batch reply", ); + expect(sessionAccessor.readSessionTranscriptWatermarkBatch).toHaveBeenCalledOnce(); + expect(sessionAccessor.readSessionTranscriptWatermark).not.toHaveBeenCalled(); expect(sessionAccessor.readSessionTranscriptTitleProbeBatch).toHaveBeenCalledOnce(); expect(sessionAccessor.readSessionTranscriptMessageEventPage).not.toHaveBeenCalled(); }); diff --git a/src/gateway/session-transcript-title-reader.ts b/src/gateway/session-transcript-title-reader.ts index ca628c68a625..28c15eedb7a8 100644 --- a/src/gateway/session-transcript-title-reader.ts +++ b/src/gateway/session-transcript-title-reader.ts @@ -4,6 +4,7 @@ import { readSessionTranscriptMessageEventPage, readSessionTranscriptTitleProbeBatch, readSessionTranscriptWatermark, + readSessionTranscriptWatermarkBatch, type SessionTranscriptMessageEvent, type SessionTranscriptReadScope, } from "../config/sessions/session-accessor.js"; @@ -168,6 +169,14 @@ export function readSessionTitleFieldsFromTranscriptBatch( scope: SessionTranscriptReadScope; target: ResolvedTranscriptReadTarget; }> = []; + const cachedCandidates: Array<{ + cacheKey: string; + cached: SqliteTitleFieldCacheEntry; + cachedFields: SessionTitleFields; + index: number; + scope: SessionTranscriptReadScope; + target: ResolvedTranscriptReadTarget; + }> = []; for (const [index, scope] of scopes.entries()) { const target = resolveTranscriptReadTarget(scope); @@ -176,18 +185,34 @@ export function readSessionTitleFieldsFromTranscriptBatch( const cached = sqliteTitleFieldCache.get(cacheKey); const cachedFields = cached?.fields[variant]; if (cached && cachedFields) { - // Keep the single-row generation/maxSeq validity contract, but validate only warm rows; - // cold or changed rows still collapse into the one store-batched probe below. - const watermark = readSessionTranscriptWatermark(scope); - if (cached.generation === watermark.generation && cached.maxSeq === watermark.maxSeq) { - setSqliteTitleFieldCache(cacheKey, cached); - results.set(index, { ...cachedFields }); - continue; - } + cachedCandidates.push({ cacheKey, cached, cachedFields, index, scope, target }); + continue; } misses.push({ cacheKey, index, scope, target }); } + const watermarks = readSessionTranscriptWatermarkBatch( + cachedCandidates.map((candidate) => candidate.scope), + ); + for (const [candidateIndex, candidate] of cachedCandidates.entries()) { + const watermark = watermarks[candidateIndex]; + if ( + watermark && + candidate.cached.generation === watermark.generation && + candidate.cached.maxSeq === watermark.maxSeq + ) { + setSqliteTitleFieldCache(candidate.cacheKey, candidate.cached); + results.set(candidate.index, { ...candidate.cachedFields }); + continue; + } + misses.push({ + cacheKey: candidate.cacheKey, + index: candidate.index, + scope: candidate.scope, + target: candidate.target, + }); + } + const probes = misses.length > 0 ? readSessionTranscriptTitleProbeBatch(misses.map((miss) => miss.scope)) : []; for (const [probeIndex, miss] of misses.entries()) { diff --git a/src/gateway/session-utils.perf.test.ts b/src/gateway/session-utils.perf.test.ts index a0cdf3ea22a7..508e26a5fe23 100644 --- a/src/gateway/session-utils.perf.test.ts +++ b/src/gateway/session-utils.perf.test.ts @@ -311,6 +311,16 @@ describe("listSessionsFromStore resolver cache", () => { derivedTitle: "title 29", lastMessagePreview: "last 29", }); + + titleBatchSpy.mockClear(); + await listSessionsFromStoreAsync({ + cfg, + storePath, + store, + opts: { includeDerivedTitles: false, includeLastMessage: false, limit: 30 }, + }); + expect(titleBatchSpy).toHaveBeenCalledOnce(); + expect(titleBatchSpy).toHaveBeenCalledWith([]); } finally { titleBatchSpy.mockRestore(); }