perf(gateway): bound sessions.list transcript validation (#118380)

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<string, ReturnType<typeof resolveSqliteTargetFromSessionStorePath>>
+>;
+
 const SQLITE_SESSION_SLOW_WRITE_MS = 1_000;
 const SQLITE_SESSION_WRITER_QUEUES = new Map<string, StoreWriterQueue>();
 
@@ -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<typeof resolveSqliteTargetFromSessionStorePath> {
+  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,
   };
 }

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 });

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<string, SessionTranscriptWatermark> {
+  const db = getNodeSqliteKysely<WatermarkDatabase>(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<number>("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<SessionTranscriptWatermark | undefined> = 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<string, SessionTranscriptWatermark>();
+    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 });
+}

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. */

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 } : {}),
   };
 }
 

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 {

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();
   });

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()) {

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();
       }