diff --git a/scripts/check-kysely-guardrails.mjs b/scripts/check-kysely-guardrails.mjs index 22a5d6db73f4..bd636cc5c48e 100644 --- a/scripts/check-kysely-guardrails.mjs +++ b/scripts/check-kysely-guardrails.mjs @@ -105,6 +105,9 @@ const rawSqliteAllowPathGroups = { "src/infra/state-migrations.media-persistence.ts", ], "shared database stores with direct DatabaseSync access": ["src/proxy-capture/store.sqlite.ts"], + "session entry cache connection-local validity counters": [ + "src/config/sessions/session-accessor.sqlite-entry-cache.ts", + ], "Kysely-backed stores that own a DatabaseSync boundary": [ "src/acp/event-ledger.ts", "src/state/user-profiles.ts", diff --git a/src/commands/doctor-telegram-general-topic-conversations.ts b/src/commands/doctor-telegram-general-topic-conversations.ts index 1390972e783c..eda936e7badc 100644 --- a/src/commands/doctor-telegram-general-topic-conversations.ts +++ b/src/commands/doctor-telegram-general-topic-conversations.ts @@ -13,11 +13,10 @@ import { resolveAllAgentSessionStoreTargetsSync } from "../config/sessions/targe import type { SessionEntry } from "../config/sessions/types.js"; import type { OpenClawConfig } from "../config/types.openclaw.js"; import { executeSqliteQuerySync } from "../infra/kysely-sync.js"; -import { runSqliteImmediateTransactionSync } from "../infra/sqlite-transaction.js"; import { buildConversationRef } from "../routing/conversation-ref.js"; import { withOpenClawAgentDatabaseReadOnly } from "../state/openclaw-agent-db-readonly.js"; import { - openOpenClawAgentDatabase, + runOpenClawAgentWriteTransaction, type OpenClawAgentDatabase, } from "../state/openclaw-agent-db.js"; @@ -316,10 +315,8 @@ export async function repairTelegramGeneralTopicConversations(params: { let repaired = 0; for (const { scope } of resolveRepairScopes(params.cfg, env)) { await runExclusiveSqliteSessionWrite(scope, async () => { - const database = openOpenClawAgentDatabase(toDatabaseOptions(scope)); - repaired += runSqliteImmediateTransactionSync( - database.db, - () => { + repaired += runOpenClawAgentWriteTransaction( + (database) => { // Detection is advisory. Re-read every candidate after BEGIN so a live // session write cannot turn the doctor repair into a stale merge. let databaseRepairs = 0; @@ -330,7 +327,8 @@ export async function repairTelegramGeneralTopicConversations(params: { } return databaseRepairs; }, - { databaseLabel: database.path, operationLabel: "doctor-telegram-general-topic" }, + toDatabaseOptions(scope), + { operationLabel: "doctor-telegram-general-topic" }, ); }); } diff --git a/src/config/sessions/session-accessor.conformance.test.ts b/src/config/sessions/session-accessor.conformance.test.ts index 27465af3060a..db86a1ddb749 100644 --- a/src/config/sessions/session-accessor.conformance.test.ts +++ b/src/config/sessions/session-accessor.conformance.test.ts @@ -684,7 +684,7 @@ describe.each([publicAccessorAdapter, sqliteAdapter])( }); }); - it("does not parse unrelated SQLite entry blobs for keyed loads", async () => { + it("parses SQLite entry blobs once across keyed loads", async () => { const scope = sqliteAdapter.entryScope(paths); for (let index = 0; index < 20; index += 1) { await upsertSqliteSessionEntry( @@ -711,7 +711,13 @@ describe.each([publicAccessorAdapter, sqliteAdapter])( model: "target", sessionId: "target-session", }); - expect(parseSpy.mock.calls.length).toBeLessThanOrEqual(2); + const initialParseCount = parseSpy.mock.calls.length; + expect(initialParseCount).toBe(21); + expect(loadSqliteSessionEntry(scope)).toMatchObject({ + model: "target", + sessionId: "target-session", + }); + expect(parseSpy).toHaveBeenCalledTimes(initialParseCount); } finally { parseSpy.mockRestore(); } diff --git a/src/config/sessions/session-accessor.entry.ts b/src/config/sessions/session-accessor.entry.ts index 29404b2e6cc5..4549f83ee792 100644 --- a/src/config/sessions/session-accessor.entry.ts +++ b/src/config/sessions/session-accessor.entry.ts @@ -350,18 +350,16 @@ export function listSessionEntries(scope: SessionEntryListScope = {}): SessionEn /** * Borrowed keyed view over one resolved store for synchronous read-only hot paths. * Unlike loadSessionEntry, `get` is a raw exact persisted-key probe with no alias - * or canonical-key resolution and no row scans, so large stores stay cheap until - * `entries` is called. Rows are borrowed, not cloned: callers must not mutate them - * and must drop the view before any await. + * or canonical-key resolution. The first probe materializes one validated store + * snapshot; later probes and `entries` reuse its parsed rows. Rows are borrowed, + * not cloned: callers must not mutate them and must drop the view before any await. */ export function openSessionEntryReadView( scope: Omit = {}, ): SessionEntryReadView { - // Exact-key probes read single SQLite rows; entries() materializes the full - // list only when raw probes cannot settle the caller's lookup. return { - get: (sessionKey) => loadExactSessionEntry({ ...scope, sessionKey })?.entry, - entries: () => listSqliteSessionEntries(scope), + get: (sessionKey) => loadExactSessionEntry({ ...scope, clone: false, sessionKey })?.entry, + entries: () => listSqliteSessionEntries({ ...scope, clone: false }), }; } diff --git a/src/config/sessions/session-accessor.sqlite-data-version.test.ts b/src/config/sessions/session-accessor.sqlite-data-version.test.ts new file mode 100644 index 000000000000..0732ae60983b --- /dev/null +++ b/src/config/sessions/session-accessor.sqlite-data-version.test.ts @@ -0,0 +1,328 @@ +import path from "node:path"; +import { DatabaseSync } from "node:sqlite"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; +import { configureSqliteConnectionPragmas } from "../../infra/sqlite-wal.js"; +import { + closeOpenClawAgentDatabasesForTest, + openOpenClawAgentDatabase, + runOpenClawAgentWriteTransaction, +} from "../../state/openclaw-agent-db.js"; +import { closeOpenClawStateDatabaseForTest } from "../../state/openclaw-state-db.js"; +import { + listSessionEntries, + listSessionEntryKeysReadOnly, + loadSessionEntry, + openSessionEntryReadView, + upsertSessionEntry, +} from "./session-accessor.js"; +import { ensureTranscriptSessionRoot } from "./session-accessor.sqlite-transcript-state.js"; + +const parseSessionEntryCalls = vi.hoisted(() => vi.fn()); + +vi.mock("./session-accessor.sqlite-status.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + parseSqliteSessionEntryJson: ( + row: Parameters[0], + ) => { + parseSessionEntryCalls(); + return actual.parseSqliteSessionEntryJson(row); + }, + }; +}); + +const tempDirs = useAutoCleanupTempDirTracker(afterEach); + +beforeEach(() => { + parseSessionEntryCalls.mockClear(); +}); + +afterEach(() => { + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); +}); + +function readDataVersion(database: DatabaseSync): number { + const row = database.prepare("PRAGMA data_version").get() as { data_version: number }; + return row.data_version; +} + +function readTotalChanges(database: DatabaseSync): number { + const row = database.prepare("SELECT total_changes() AS value").get() as { value: number }; + return row.value; +} + +describe("SQLite entry cache validity counters", () => { + it("separately tracks same-connection and other-connection commits", () => { + const databasePath = path.join(tempDirs.make("openclaw-data-version-"), "probe.sqlite"); + const first = new DatabaseSync(databasePath); + const firstMaintenance = configureSqliteConnectionPragmas(first, { + checkpointIntervalMs: 0, + databaseLabel: "data-version-first", + databasePath, + foreignKeys: true, + synchronous: "NORMAL", + }); + first.exec("CREATE TABLE probe (value TEXT NOT NULL) STRICT;"); + const second = new DatabaseSync(databasePath); + const secondMaintenance = configureSqliteConnectionPragmas(second, { + checkpointIntervalMs: 0, + databaseLabel: "data-version-second", + databasePath, + foreignKeys: true, + synchronous: "NORMAL", + }); + + try { + expect(first.prepare("PRAGMA journal_mode").get()).toEqual({ journal_mode: "wal" }); + expect(second.prepare("PRAGMA journal_mode").get()).toEqual({ journal_mode: "wal" }); + + const firstVersion = readDataVersion(first); + const firstChanges = readTotalChanges(first); + first.exec("BEGIN IMMEDIATE; INSERT INTO probe VALUES ('first'); COMMIT;"); + expect(readDataVersion(first)).toBe(firstVersion); + expect(readTotalChanges(first)).toBe(firstChanges + 1); + + const secondVersion = readDataVersion(second); + const secondChanges = readTotalChanges(second); + second.exec("BEGIN IMMEDIATE; INSERT INTO probe VALUES ('second'); COMMIT;"); + expect(readDataVersion(second)).toBe(secondVersion); + expect(readTotalChanges(second)).toBe(secondChanges + 1); + expect(readDataVersion(first)).not.toBe(firstVersion); + expect(readTotalChanges(first)).toBe(firstChanges + 1); + } finally { + secondMaintenance.close(); + second.close(); + firstMaintenance.close(); + first.close(); + } + }); +}); + +function createSessionScope(label: string) { + const stateDir = tempDirs.make(`openclaw-entry-cache-${label}-`); + return { + agentId: "main", + env: { ...process.env, OPENCLAW_STATE_DIR: stateDir }, + sessionKey: `agent:main:${label}`, + }; +} + +describe("SQLite session entry cache", () => { + it("reuses parsed entries on the second list", async () => { + const scope = createSessionScope("second-list"); + await upsertSessionEntry(scope, { label: "first", sessionId: "first", updatedAt: 1 }); + await upsertSessionEntry( + { ...scope, sessionKey: "agent:main:second-list-2" }, + { label: "second", sessionId: "second", updatedAt: 2 }, + ); + + parseSessionEntryCalls.mockClear(); + const first = listSessionEntries(scope); + const firstParseCount = parseSessionEntryCalls.mock.calls.length; + const second = listSessionEntries(scope); + + expect(firstParseCount).toBe(2); + expect(parseSessionEntryCalls).toHaveBeenCalledTimes(firstParseCount); + expect(second).toEqual(first); + }); + + it("reloads after another connection commits", async () => { + const scope = createSessionScope("external-write"); + await upsertSessionEntry(scope, { label: "before", sessionId: "external", updatedAt: 1 }); + const before = listSessionEntries(scope)[0]?.entry; + expect(before).toBeDefined(); + if (!before) { + throw new Error("missing seeded external-write entry"); + } + const database = openOpenClawAgentDatabase(scope); + const external = new DatabaseSync(database.path); + const maintenance = configureSqliteConnectionPragmas(external, { + checkpointIntervalMs: 0, + databaseLabel: "session-entry-external-writer", + databasePath: database.path, + foreignKeys: true, + synchronous: "NORMAL", + }); + try { + const updated = { ...before, label: "after", updatedAt: 2 }; + external + .prepare( + "UPDATE session_nodes SET entry_json = ?, label = ?, updated_at = ? WHERE session_key = ?", + ) + .run(JSON.stringify(updated), updated.label, updated.updatedAt, scope.sessionKey); + + parseSessionEntryCalls.mockClear(); + expect(listSessionEntries(scope)[0]?.entry.label).toBe("after"); + expect(parseSessionEntryCalls).toHaveBeenCalledTimes(1); + } finally { + maintenance.close(); + external.close(); + } + }); + + it("reloads after a raw insert on the cached connection", async () => { + const scope = createSessionScope("same-connection-insert"); + await upsertSessionEntry(scope, { + label: "existing", + sessionId: "same-connection-existing", + updatedAt: 1, + }); + listSessionEntries(scope); + + const database = openOpenClawAgentDatabase(scope); + const insertedKey = "agent:main:same-connection-inserted"; + const insertedEntry = { + label: "inserted", + sessionId: "same-connection-inserted", + updatedAt: 2, + }; + database.db + .prepare( + "INSERT INTO session_nodes (session_key, current_session_id, entry_json, updated_at) VALUES (?, ?, ?, ?)", + ) + .run( + insertedKey, + insertedEntry.sessionId, + JSON.stringify(insertedEntry), + insertedEntry.updatedAt, + ); + + parseSessionEntryCalls.mockClear(); + const entries = listSessionEntries(scope); + + expect(entries.map((row) => row.sessionKey)).toEqual([scope.sessionKey, insertedKey]); + expect(entries[1]?.entry).toMatchObject(insertedEntry); + expect(parseSessionEntryCalls).toHaveBeenCalledTimes(2); + }); + + it("reloads after a tracked same-process upsert", async () => { + const scope = createSessionScope("write-through"); + await upsertSessionEntry(scope, { label: "before", sessionId: "write-through", updatedAt: 1 }); + listSessionEntries(scope); + + await upsertSessionEntry(scope, { label: "after", updatedAt: 2 }); + parseSessionEntryCalls.mockClear(); + + expect(listSessionEntries(scope)[0]?.entry.label).toBe("after"); + expect(parseSessionEntryCalls).toHaveBeenCalledTimes(1); + }); + + it("does not let a tracked write mask an earlier raw connection write", async () => { + const scope = createSessionScope("raw-before-tracked"); + const trackedScope = { ...scope, sessionKey: "agent:main:tracked-after-raw" }; + await upsertSessionEntry(scope, { label: "raw-before", sessionId: "raw", updatedAt: 1 }); + await upsertSessionEntry(trackedScope, { + label: "tracked-before", + sessionId: "tracked", + updatedAt: 1, + }); + listSessionEntries(scope); + + const database = openOpenClawAgentDatabase(scope); + const rawEntry = { label: "raw-after", sessionId: "raw", updatedAt: Date.now() }; + database.db + .prepare("UPDATE session_nodes SET entry_json = ?, updated_at = ? WHERE session_key = ?") + .run(JSON.stringify(rawEntry), rawEntry.updatedAt, scope.sessionKey); + await upsertSessionEntry(trackedScope, { label: "tracked-after", updatedAt: 2 }); + + parseSessionEntryCalls.mockClear(); + const entries = listSessionEntries(scope); + const entriesBySessionId = new Map(entries.map((row) => [row.entry.sessionId, row.entry])); + + expect(entriesBySessionId.get("raw")).toMatchObject(rawEntry); + expect(entriesBySessionId.get("tracked")).toMatchObject({ + label: "tracked-after", + sessionId: "tracked", + }); + expect(parseSessionEntryCalls).toHaveBeenCalledTimes(2); + }); + + it("invalidates cached keys when transcript creation inserts a placeholder node", async () => { + const scope = createSessionScope("placeholder-key"); + await upsertSessionEntry(scope, { sessionId: "entry", updatedAt: 1 }); + expect(listSessionEntryKeysReadOnly({ agentId: scope.agentId, env: scope.env })).toEqual([ + scope.sessionKey, + ]); + + const placeholderKey = "agent:main:placeholder-only"; + runOpenClawAgentWriteTransaction((database) => { + ensureTranscriptSessionRoot( + database, + { + agentId: scope.agentId, + env: scope.env, + sessionId: "placeholder-only", + sessionKey: placeholderKey, + }, + 2, + ); + }, scope); + + expect(listSessionEntryKeysReadOnly({ agentId: scope.agentId, env: scope.env })).toEqual([ + scope.sessionKey, + placeholderKey, + ]); + }); + + it("bypasses the cache in a transaction and reloads persisted state after rollback", async () => { + const scope = createSessionScope("transaction-rollback"); + await upsertSessionEntry(scope, { label: "before", sessionId: "rollback", updatedAt: 1 }); + const borrowedBefore = openSessionEntryReadView(scope).get(scope.sessionKey); + expect(borrowedBefore?.label).toBe("before"); + if (!borrowedBefore) { + throw new Error("missing seeded rollback entry"); + } + + expect(() => + runOpenClawAgentWriteTransaction((database) => { + const updated = { ...borrowedBefore, label: "uncommitted", updatedAt: 2 }; + database.db + .prepare("UPDATE session_nodes SET entry_json = ?, updated_at = ? WHERE session_key = ?") + .run(JSON.stringify(updated), updated.updatedAt, scope.sessionKey); + expect(loadSessionEntry({ ...scope, clone: false })?.label).toBe("uncommitted"); + throw new Error("roll back cache probe"); + }, scope), + ).toThrow("roll back cache probe"); + + parseSessionEntryCalls.mockClear(); + const borrowedAfter = openSessionEntryReadView(scope).get(scope.sessionKey); + expect(borrowedAfter).not.toBe(borrowedBefore); + expect(borrowedAfter?.label).toBe("before"); + expect(parseSessionEntryCalls).toHaveBeenCalledTimes(1); + }); + + it("isolates cloned results while borrowed views retain stable references", async () => { + const scope = createSessionScope("clone-borrow"); + await upsertSessionEntry(scope, { label: "original", sessionId: "clone", updatedAt: 1 }); + + const cloned = listSessionEntries(scope)[0]?.entry; + expect(cloned).toBeDefined(); + if (cloned) { + cloned.label = "mutated"; + } + expect(loadSessionEntry(scope)?.label).toBe("original"); + + const view = openSessionEntryReadView(scope); + const first = view.get(scope.sessionKey); + expect(view.get(scope.sessionKey)).toBe(first); + expect(view.entries()[0]?.entry).toBe(first); + }); + + it("honors latest reads after an untracked own-connection write", async () => { + const scope = createSessionScope("latest"); + await upsertSessionEntry(scope, { label: "cached", sessionId: "latest", updatedAt: 1 }); + expect(loadSessionEntry(scope)?.label).toBe("cached"); + + const database = openOpenClawAgentDatabase(scope); + const updated = { label: "latest", sessionId: "latest", updatedAt: 2 }; + database.db + .prepare("UPDATE session_nodes SET entry_json = ?, updated_at = ? WHERE session_key = ?") + .run(JSON.stringify(updated), updated.updatedAt, scope.sessionKey); + + expect(loadSessionEntry(scope)?.label).toBe("latest"); + expect(loadSessionEntry({ ...scope, readConsistency: "latest" })?.label).toBe("latest"); + }); +}); diff --git a/src/config/sessions/session-accessor.sqlite-entry-cache.ts b/src/config/sessions/session-accessor.sqlite-entry-cache.ts new file mode 100644 index 000000000000..d472f183ef2e --- /dev/null +++ b/src/config/sessions/session-accessor.sqlite-entry-cache.ts @@ -0,0 +1,137 @@ +import type { DatabaseSync } from "node:sqlite"; +import { executeSqliteQuerySync } from "../../infra/kysely-sync.js"; +import { + deferOpenClawAgentPostCommitPublication, + type OpenClawAgentDatabase, +} from "../../state/openclaw-agent-db.js"; +import { getSessionKysely } from "./session-accessor.sqlite-scope.js"; +import { parseSqliteSessionEntryJson } from "./session-accessor.sqlite-status.js"; +import type { SessionEntry } from "./types.js"; + +type SessionEntryCacheDatabase = Pick; + +export type SqliteSessionEntryCacheSnapshot = { + entries: Map; + keys: string[]; + listEntries: Map; +}; + +type SqliteSessionEntryCache = SqliteSessionEntryCacheSnapshot & { + connection: DatabaseSync; + validityToken: SqliteSessionEntryCacheValidityToken; +}; + +type SqliteSessionEntryCacheValidityToken = { + dataVersion: number; + totalChanges: number; +}; + +// One parsed snapshot per opened agent database bounds memory to the process's database set. +// The connection-local validity token plus tracked-write invalidation keeps it current; +// without both, every read would re-query and re-parse every entry_json document. +const sessionEntryCaches = new Map(); + +function readDataVersion(database: DatabaseSync): number { + const row = database.prepare("PRAGMA data_version").get() as { data_version?: unknown }; + if (typeof row.data_version !== "number") { + throw new Error("SQLite did not return a numeric PRAGMA data_version"); + } + return row.data_version; +} + +function readTotalChanges(database: DatabaseSync): number { + const row = database.prepare("SELECT total_changes() AS value").get() as { value?: unknown }; + if (typeof row.value !== "number") { + throw new Error("SQLite did not return a numeric total_changes() value"); + } + return row.value; +} + +function readCacheValidityToken(database: DatabaseSync): SqliteSessionEntryCacheValidityToken { + return { + dataVersion: readDataVersion(database), + totalChanges: readTotalChanges(database), + }; +} + +function cacheValidityTokensEqual( + left: SqliteSessionEntryCacheValidityToken, + right: SqliteSessionEntryCacheValidityToken, +): boolean { + return left.dataVersion === right.dataVersion && left.totalChanges === right.totalChanges; +} + +function createListProjection(entry: SessionEntry): SessionEntry { + const projected = structuredClone(entry); + delete projected.skillsSnapshot; + delete projected.systemPromptReport; + return projected; +} + +function loadSessionEntrySnapshot( + database: SessionEntryCacheDatabase, +): SqliteSessionEntryCacheSnapshot { + const db = getSessionKysely(database.db); + const rows = executeSqliteQuerySync( + database.db, + db.selectFrom("session_nodes").select(["session_key", "entry_json"]).orderBy("session_key"), + ).rows; + const entries = new Map(); + const listEntries = new Map(); + for (const row of rows) { + const entry = parseSqliteSessionEntryJson(row); + if (!entry) { + continue; + } + entries.set(row.session_key, entry); + listEntries.set(row.session_key, createListProjection(entry)); + } + return { + entries, + keys: rows.map((row) => row.session_key), + listEntries, + }; +} + +export function readSqliteSessionEntryCache( + database: SessionEntryCacheDatabase, + options: { cache: boolean; latest?: boolean }, +): SqliteSessionEntryCacheSnapshot { + if (!options.cache || options.latest || database.db.isTransaction) { + return loadSessionEntrySnapshot(database); + } + const validityToken = readCacheValidityToken(database.db); + const cached = sessionEntryCaches.get(database.path); + if ( + cached?.connection === database.db && + cacheValidityTokensEqual(cached.validityToken, validityToken) + ) { + return cached; + } + const loaded = loadSessionEntrySnapshot(database); + const next = { ...loaded, connection: database.db, validityToken }; + sessionEntryCaches.set(database.path, next); + return next; +} + +function invalidateTrackedCache(database: OpenClawAgentDatabase): void { + const invalidate = () => { + const cached = sessionEntryCaches.get(database.path); + if (cached?.connection === database.db) { + sessionEntryCaches.delete(database.path); + } + }; + if (deferOpenClawAgentPostCommitPublication(database, invalidate)) { + return; + } + if (database.db.isTransaction) { + throw new Error( + "SQLite session entry writes must use runOpenClawAgentWriteTransaction for cache publication", + ); + } + invalidate(); +} + +export function publishSqliteSessionEntryCacheInvalidation(database: OpenClawAgentDatabase): void { + invalidateTrackedCache(database); +} diff --git a/src/config/sessions/session-accessor.sqlite-entry-store.ts b/src/config/sessions/session-accessor.sqlite-entry-store.ts index 6ca8f2442f66..5db62a0d119a 100644 --- a/src/config/sessions/session-accessor.sqlite-entry-store.ts +++ b/src/config/sessions/session-accessor.sqlite-entry-store.ts @@ -11,6 +11,7 @@ import { prepareSessionConversation, upsertConversationIdentity, } from "./session-accessor.sqlite-conversation.js"; +import { publishSqliteSessionEntryCacheInvalidation } from "./session-accessor.sqlite-entry-cache.js"; import { clearSessionCollaborationForKey, deleteSessionNodeArtifacts, @@ -352,12 +353,14 @@ export function deleteSqliteSessionEntryRows( sessionKey, updatedAt: remainingWindow.updated_at, }); + publishSqliteSessionEntryCacheInvalidation(database); return; } executeSqliteQuerySync( database.db, db.deleteFrom("session_nodes").where("session_key", "=", sessionKey), ); + publishSqliteSessionEntryCacheInvalidation(database); } /** Remove the logical entry while retaining its node-owned transcript windows. */ @@ -479,6 +482,7 @@ export function deleteLegacySessionEntryRows( database.db, db.deleteFrom("session_nodes").where("session_key", "=", legacyKey), ); + publishSqliteSessionEntryCacheInvalidation(database); } } @@ -634,6 +638,7 @@ export function writeSessionEntry( updatedAt, }); } + publishSqliteSessionEntryCacheInvalidation(database); } /** Resolves the parent fork decision using SQLite transcript rows when totals are stale. */ diff --git a/src/config/sessions/session-accessor.sqlite-entry.ts b/src/config/sessions/session-accessor.sqlite-entry.ts index a3675444c4d7..d48c2e6b51e6 100644 --- a/src/config/sessions/session-accessor.sqlite-entry.ts +++ b/src/config/sessions/session-accessor.sqlite-entry.ts @@ -1,12 +1,9 @@ -import type { DatabaseSync } from "node:sqlite"; import type { MsgContext } from "../../auto-reply/templating.js"; -import { - executeSqliteQuerySync, - executeSqliteQueryTakeFirstSync, -} from "../../infra/kysely-sync.js"; +import { executeSqliteQueryTakeFirstSync } from "../../infra/kysely-sync.js"; import type { ChannelRouteRef } from "../../plugin-sdk/channel-route.js"; import { withOpenClawAgentDatabaseReadOnly } from "../../state/openclaw-agent-db-readonly.js"; import { + isIncognitoOpenClawAgentSqlitePath, openOpenClawAgentDatabase, resolveOpenClawAgentSqlitePath, runOpenClawAgentWriteTransaction, @@ -26,13 +23,16 @@ import type { SessionEntryTargetPatchScope, SessionTranscriptReadScope, } from "./session-accessor.sqlite-contract.js"; +import { + readSqliteSessionEntryCache, + type SqliteSessionEntryCacheSnapshot, +} from "./session-accessor.sqlite-entry-cache.js"; import { assertSqliteLifecycleTargetSnapshotUnchanged, assertSqliteSessionEntrySelectionUnchanged, collectSessionEntryLookupKeys, createSqliteSessionIdentitySnapshot, deleteLegacySessionEntryRows, - readExactSessionEntryRow, readSessionEntryRow, readSqliteLifecycleTargetSnapshot, readSqliteSessionEntrySelectionSnapshot, @@ -62,15 +62,13 @@ import { toDatabaseOptions, type ResolvedSqliteScope, } from "./session-accessor.sqlite-scope.js"; -import { - parseSqliteSessionEntryJson as parseSessionEntryRow, - readSqliteSessionEntriesByStatus, -} from "./session-accessor.sqlite-status.js"; +import { readSqliteSessionEntriesByStatus } from "./session-accessor.sqlite-status.js"; import type { SessionEntryListScope } from "./session-accessor.types.js"; import { preserveSqliteSameKeySessionRolloverLineage } from "./session-entry-lineage.js"; import { buildSessionCreationStamp } from "./session-entry-provenance.js"; import { kickSessionHistoryDiskBudgetMaintenance } from "./session-history-eviction.js"; import { resolveSessionStorePathForScope } from "./session-store-path.js"; +import { resolveSessionEntryCandidates } from "./store-entry.js"; import type { GroupKeyResolution, SessionEntry } from "./types.js"; import { mergeSessionEntry, mergeSessionEntryPreserveActivity } from "./types.js"; @@ -92,11 +90,22 @@ export function resolveSqliteSessionEntry( options: { readOnly?: boolean } = {}, ): ResolvedSqliteSessionEntry { const resolved = resolveSqliteScope(scope); - const read = (database: Pick): ResolvedSqliteSessionEntry => { - const row = readSessionEntryRow(database, resolved.sessionKey); + const read = ( + database: Pick, + ): ResolvedSqliteSessionEntry => { + const snapshot = readSessionEntrySnapshot(database, resolved, scope.readConsistency); + const selected = resolveSessionEntryCandidates({ + entries: [...snapshot.entries].map(([sessionKey, entry]) => ({ entry, sessionKey })), + sessionKey: resolved.sessionKey, + }); + const existing = selected.existing?.entry; return { - existing: row?.entry, - legacyKeys: row?.legacyKeys ?? [], + existing: existing + ? scope.clone === false + ? existing + : cloneSessionEntry(existing) + : undefined, + legacyKeys: selected.legacyKeys, normalizedKey: resolved.sessionKey, }; }; @@ -131,21 +140,21 @@ export function loadExactSqliteSessionEntry( } const resolved = resolveSqliteScope(scope); const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); - const row = readExactSessionEntryRow(database, sessionKey); - return row ? { sessionKey, entry: row.entry } : undefined; + const entry = readSessionEntrySnapshot(database, resolved, scope.readConsistency).entries.get( + sessionKey, + ); + return entry + ? { sessionKey, entry: scope.clone === false ? entry : cloneSessionEntry(entry) } + : undefined; } -/** Lists persisted session keys only, skipping entry_json parsing entirely. */ +/** Lists persisted session keys from the data-version-validated entry snapshot. */ export function listSqliteSessionEntryKeysReadOnly( scope: Partial> = {}, ): string[] { const resolved = resolveSqliteScope({ ...scope, sessionKey: "" }); const result = withOpenClawAgentDatabaseReadOnly((database) => { - const db = getSessionKysely(database.db); - return executeSqliteQuerySync( - database.db, - db.selectFrom("session_nodes").select(["session_key"]).orderBy("session_key", "asc"), - ).rows.map((row) => row.session_key); + return [...readSessionEntrySnapshot(database, resolved, scope.readConsistency).keys]; }, toDatabaseOptions(resolved)); return result.found ? result.value : []; } @@ -160,10 +169,16 @@ export function loadExactSqliteSessionEntryReadOnly( } const resolved = resolveSqliteScope(scope); const result = withOpenClawAgentDatabaseReadOnly( - (database) => readExactSessionEntryRow(database, sessionKey)?.entry, + (database) => + readSessionEntrySnapshot(database, resolved, scope.readConsistency).entries.get(sessionKey), toDatabaseOptions(resolved), ); - return result.found && result.value ? { sessionKey, entry: result.value } : undefined; + return result.found && result.value + ? { + sessionKey, + entry: scope.clone === false ? result.value : cloneSessionEntry(result.value), + } + : undefined; } /** Resolves the persisted session key for a SQLite transcript session id. */ @@ -188,7 +203,7 @@ export function resolveSqliteSessionKeyBySessionId( export function listSqliteSessionEntries(scope: SessionEntryListScope = {}): SessionEntrySummary[] { const resolved = resolveSqliteScope({ ...scope, sessionKey: "" }); const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); - return listSqliteSessionEntriesFromDatabase(database, scope.projection); + return listSqliteSessionEntriesFromDatabase(database, resolved, scope); } /** @@ -201,46 +216,49 @@ export function listSqliteSessionEntriesReadOnly( ): SessionEntrySummary[] { const resolved = resolveSqliteScope({ ...scope, sessionKey: "" }); const result = withOpenClawAgentDatabaseReadOnly( - (database) => listSqliteSessionEntriesFromDatabase(database, scope.projection), + (database) => listSqliteSessionEntriesFromDatabase(database, resolved, scope), toDatabaseOptions(resolved), ); return result.found ? result.value : []; } function listSqliteSessionEntriesFromDatabase( - database: { db: DatabaseSync }, - projection: SessionEntryListScope["projection"] = "full", -) { - const db = getSessionKysely(database.db); - const baseQuery = db.selectFrom("session_nodes").select("session_key"); - const query = - projection === "list" - ? baseQuery.select((eb) => - eb - .case() - .when(eb.fn("json_valid", [eb.ref("entry_json")]), "=", 1) - .then( - eb.fn("json_remove", [ - eb.ref("entry_json"), - eb.val("$.systemPromptReport"), - eb.val("$.skillsSnapshot"), - ]), - ) - .else(eb.ref("entry_json")) - .end() - .as("entry_json"), - ) - : baseQuery.select("entry_json"); - const rows = executeSqliteQuerySync(database.db, query.orderBy("session_key", "asc")).rows; - return rows - .map((row) => { - if (isInternalSessionEffectsKey(row.session_key)) { - return undefined; - } - const entry = parseSessionEntryRow(row); - return entry ? { sessionKey: row.session_key, entry } : undefined; - }) - .filter((entry): entry is SessionEntrySummary => entry !== undefined); + database: Pick, + resolved: ResolvedSqliteScope, + scope: SessionEntryListScope, +): SessionEntrySummary[] { + const snapshot = readSessionEntrySnapshot(database, resolved, scope.readConsistency); + const entries = scope.projection === "list" ? snapshot.listEntries : snapshot.entries; + return snapshot.keys.flatMap((sessionKey) => { + if (isInternalSessionEffectsKey(sessionKey)) { + return []; + } + const entry = entries.get(sessionKey); + if (!entry) { + return []; + } + return [ + { + sessionKey, + entry: scope.clone === false ? entry : cloneSessionEntry(entry), + }, + ]; + }); +} + +function readSessionEntrySnapshot( + database: Pick, + resolved: ResolvedSqliteScope, + readConsistency: SessionAccessScope["readConsistency"], +): SqliteSessionEntryCacheSnapshot { + const cache = !isIncognitoOpenClawAgentSqlitePath(database.path, { + agentId: database.agentId, + env: resolved.env, + }); + return readSqliteSessionEntryCache(database, { + cache, + latest: readConsistency === "latest", + }); } /** Lists only entries whose normalized session row has one of the requested statuses. */ diff --git a/src/config/sessions/session-accessor.sqlite-transcript-state.ts b/src/config/sessions/session-accessor.sqlite-transcript-state.ts index f448adc0a04f..5b63dcf49863 100644 --- a/src/config/sessions/session-accessor.sqlite-transcript-state.ts +++ b/src/config/sessions/session-accessor.sqlite-transcript-state.ts @@ -4,6 +4,7 @@ import { executeSqliteQueryTakeFirstSync, } from "../../infra/kysely-sync.js"; import type { OpenClawAgentDatabase } from "../../state/openclaw-agent-db.js"; +import { publishSqliteSessionEntryCacheInvalidation } from "./session-accessor.sqlite-entry-cache.js"; import { normalizeSqliteNumber } from "./session-accessor.sqlite-normalize.js"; import { getSessionKysely, type ResolvedTranscriptScope } from "./session-accessor.sqlite-scope.js"; import { deleteSessionTranscriptIndexInTransaction } from "./session-transcript-index.js"; @@ -69,7 +70,7 @@ export function ensureTranscriptSessionRoot( updatedAt: number, ): void { const db = getSessionKysely(database.db); - executeSqliteQuerySync( + const insertedNode = executeSqliteQuerySync( database.db, db .insertInto("session_nodes") @@ -81,6 +82,9 @@ export function ensureTranscriptSessionRoot( }) .onConflict((conflict) => conflict.column("session_key").doNothing()), ); + if ((insertedNode.numAffectedRows ?? 0n) > 0n) { + publishSqliteSessionEntryCacheInvalidation(database); + } executeSqliteQuerySync( database.db, db diff --git a/src/config/sessions/session-accessor.test.ts b/src/config/sessions/session-accessor.test.ts index 02b1e642db0c..cde6b1a3bbd3 100644 --- a/src/config/sessions/session-accessor.test.ts +++ b/src/config/sessions/session-accessor.test.ts @@ -942,7 +942,7 @@ describe("session accessor seam", () => { expect(fs.existsSync(storePath)).toBe(false); }); - it("resolves canonical candidate and transcript rows without parsing unrelated sessions", async () => { + it("parses the store once across canonical candidate and transcript reads", async () => { const sessionKey = "agent:main:focused-session"; await upsertSessionEntry( { agentId: "main", sessionKey, storePath }, @@ -969,6 +969,7 @@ describe("session accessor seam", () => { legacyKeys: [], normalizedKey: sessionKey, }); + expect(parse.mock.calls.filter(([value]) => value === unrelatedEntryJson)).toHaveLength(1); expect( resolveSessionEntryCandidateTarget({ agentId: "main", @@ -984,7 +985,7 @@ describe("session accessor seam", () => { storePath, }), ).toMatchObject({ agentId: "main", sessionId: "focused-session", sessionKey }); - expect(parse).not.toHaveBeenCalledWith(unrelatedEntryJson); + expect(parse.mock.calls.filter(([value]) => value === unrelatedEntryJson)).toHaveLength(1); } finally { parse.mockRestore(); } diff --git a/src/gateway/server.sessions.list-store-materialization.test.ts b/src/gateway/server.sessions.list-store-materialization.test.ts index 2fdc12f38743..7dac995ec248 100644 --- a/src/gateway/server.sessions.list-store-materialization.test.ts +++ b/src/gateway/server.sessions.list-store-materialization.test.ts @@ -6,6 +6,7 @@ import { expect, test, vi } from "vitest"; import * as sessionAccessor from "../config/sessions/session-accessor.js"; import { resolveSqliteTargetFromSessionStorePath } from "../config/sessions/session-sqlite-target.js"; +import type { SessionEntry } from "../config/sessions/types.js"; import { openOpenClawAgentDatabase } from "../state/openclaw-agent-db.js"; import { testState, writeSessionStore } from "./test-helpers.js"; import { @@ -87,9 +88,11 @@ test("sessions.list projects out prompt snapshots without changing full entry re const stored = database.db .prepare("SELECT session_key, entry_json FROM session_nodes LIMIT 1") .get() as { session_key: string; entry_json: string }; - database.db.prepare("UPDATE session_nodes SET entry_json = ? WHERE session_key = ?").run( - JSON.stringify({ - ...(JSON.parse(stored.entry_json) as Record), + const storedEntry = JSON.parse(stored.entry_json) as SessionEntry; + await sessionAccessor.replaceSessionEntry( + { agentId: "main", sessionKey: stored.session_key, storePath }, + { + ...storedEntry, skillsSnapshot: { prompt: "large skill prompt", skills: [{ name: "test" }] }, systemPromptReport: { source: "run", @@ -99,8 +102,7 @@ test("sessions.list projects out prompt snapshots without changing full entry re skills: { promptChars: 0, entries: [] }, tools: { listChars: 0, schemaChars: 0, entries: [] }, }, - }), - stored.session_key, + }, ); database.db .prepare(