diff --git a/src/config/sessions/session-accessor.sqlite-handle-lifecycle.test.ts b/src/config/sessions/session-accessor.sqlite-handle-lifecycle.test.ts new file mode 100644 index 000000000000..d8824bdc7564 --- /dev/null +++ b/src/config/sessions/session-accessor.sqlite-handle-lifecycle.test.ts @@ -0,0 +1,177 @@ +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; +import { + closeOpenClawAgentDatabaseByPath, + closeOpenClawAgentDatabasesForTest, + openOpenClawAgentDatabase, +} from "../../state/openclaw-agent-db.js"; +import { + applySessionEntryLifecycleMutation, + appendTranscriptMessage, + loadSessionEntry, + loadTranscriptEvents, + persistSessionTranscriptTurn, + replaceSessionEntry, + withTranscriptWriteLock, +} from "./session-accessor.js"; +import { readSessionTranscriptMessageEventPage } from "./session-accessor.sqlite-active-events.js"; +import { replaceTranscriptEvents } from "./session-accessor.sqlite-transcript-write.js"; +import { enforceSqliteSessionHistoryDiskBudget } from "./session-history-eviction.js"; +import { resolveSqliteTargetFromSessionStorePath } from "./session-sqlite-target.js"; +import { + startSessionTranscriptIndexReconcile, + waitForSessionTranscriptIndexReconcile, + waitForSessionTranscriptProjection, +} from "./session-transcript-reconcile.js"; + +const archiveMaterializationHook = vi.hoisted(() => ({ + afterMaterialize: undefined as (() => void) | undefined, +})); + +// Close the cached handle after the real archive worker yields back to its caller. +vi.mock("./session-accessor.sqlite-archive.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + materializeSessionStateDeletePlans: async ( + ...args: Parameters + ) => { + const result = await actual.materializeSessionStateDeletePlans(...args); + archiveMaterializationHook.afterMaterialize?.(); + return result; + }, + }; +}); + +const tempDirs = useAutoCleanupTempDirTracker(afterEach); + +describe("SQLite session handle lifecycle", () => { + let scope: { sessionId: string; sessionKey: string; storePath: string }; + let databasePath: string; + + beforeEach(async () => { + scope = { + sessionId: "handle-session", + sessionKey: "agent:main:handle-session", + storePath: path.join(tempDirs.make("openclaw-session-handle-"), "sessions.json"), + }; + await replaceSessionEntry(scope, { sessionId: scope.sessionId, updatedAt: 1 }); + databasePath = resolveSqliteTargetFromSessionStorePath(scope.storePath).path!; + }); + + afterEach(() => { + archiveMaterializationHook.afterMaterialize = undefined; + closeOpenClawAgentDatabasesForTest(); + }); + + it.each(["events", "message facts"])( + "reads %s after a locked callback loses its handle", + async (kind) => { + const message = { role: "user", content: "retained", idempotencyKey: "handle-message" }; + await appendTranscriptMessage(scope, { message }); + + await withTranscriptWriteLock(scope, async (transcript) => { + const before = await transcript.readEvents(); + expect(closeOpenClawAgentDatabaseByPath(databasePath)).toBe(true); + if (kind === "events") { + await expect(transcript.readEvents()).resolves.toEqual(before); + } else { + const facts = await transcript.readMessageFacts({ + idempotencyKeys: [message.idempotencyKey], + }); + expect(facts.messagesByIdempotencyKey.get(message.idempotencyKey)).toMatchObject(message); + } + }); + }, + ); + + it("commits a turn after its async predicate loses the cached handle", async () => { + const result = await persistSessionTranscriptTurn(scope, { + messages: [ + { + message: { role: "user", content: "append after close" }, + shouldAppend: async () => { + expect(closeOpenClawAgentDatabaseByPath(databasePath)).toBe(true); + return true; + }, + }, + ], + updateMode: "none", + }); + + expect(result.appendedCount).toBe(1); + await expect(loadTranscriptEvents(scope)).resolves.toContainEqual( + expect.objectContaining({ + message: expect.objectContaining({ content: "append after close" }), + }), + ); + }); + + it("commits a lifecycle projection after its async builder loses the cached handle", async () => { + await expect( + applySessionEntryLifecycleMutation({ + storePath: scope.storePath, + skipMaintenance: true, + upserts: [ + { + sessionKey: scope.sessionKey, + buildEntry: async ({ currentEntry }) => { + expect(closeOpenClawAgentDatabaseByPath(databasePath)).toBe(true); + return { ...currentEntry!, label: "built after close" }; + }, + }, + ], + }), + ).resolves.toMatchObject({ afterCount: 1 }); + expect(loadSessionEntry(scope)).toMatchObject({ label: "built after close" }); + }); + + it("waits for projection repair after its polling handle closes", async () => { + await persistSessionTranscriptTurn(scope, { + messages: [{ eventId: "target", message: { role: "user", content: "target" } }], + touchSessionEntry: false, + }); + const databaseOptions = { agentId: "main", path: databasePath }; + const database = openOpenClawAgentDatabase(databaseOptions); + database.db.prepare("UPDATE session_transcript_index_state SET needs_rebuild = 1").run(); + startSessionTranscriptIndexReconcile(databaseOptions); + try { + const ready = waitForSessionTranscriptProjection(scope); + expect(closeOpenClawAgentDatabaseByPath(database.path)).toBe(true); + await ready; + expect( + readSessionTranscriptMessageEventPage(scope, { maxMessages: 0, offset: 0 }).totalMessages, + ).toBe(1); + } finally { + await waitForSessionTranscriptIndexReconcile(databaseOptions); + } + }); + + it("completes a disk-budget sweep after its handle closes during archive materialization", async () => { + const { sessionKey, sessionId, storePath } = scope; + await replaceTranscriptEvents({ sessionKey, sessionId, storePath }, [ + { type: "session", id: sessionId, content: "retained history" }, + ]); + await replaceSessionEntry( + { sessionKey, storePath }, + { sessionId: "current-session", updatedAt: 2 }, + ); + const closeHandle = vi.fn(() => { + expect(closeOpenClawAgentDatabaseByPath(databasePath)).toBe(true); + }); + archiveMaterializationHook.afterMaterialize = closeHandle; + + await expect( + enforceSqliteSessionHistoryDiskBudget({ + storePath, + mode: "enforce", + maintenance: { maxDiskBytes: 1, highWaterBytes: 0 }, + }), + ).resolves.toMatchObject({ removedEntries: 1, removedFiles: 1 }); + + expect(closeHandle).toHaveBeenCalledOnce(); + expect(loadSessionEntry({ sessionKey, storePath })?.sessionId).toBe("current-session"); + await expect(loadTranscriptEvents({ sessionKey, sessionId, storePath })).resolves.toEqual([]); + }); +}); diff --git a/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts b/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts index 9b1cb5ef8dcb..e16aefc29ba0 100644 --- a/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts +++ b/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts @@ -6,7 +6,9 @@ import { import { normalizeAgentId, parseAgentSessionKey } from "../../routing/session-key.js"; import { isIncognitoOpenClawAgentDatabase, + openOpenClawAgentDatabase, type OpenClawAgentDatabase, + type OpenClawAgentDatabaseOptions, } from "../../state/openclaw-agent-db.js"; import { persistSessionTranscriptArchive } from "./session-accessor.sqlite-archive-store.js"; import type { @@ -307,7 +309,7 @@ export function readSessionGenerationIdsForKeys( // Projects removals and upserts before archive materialization so same-call // upserts can keep a transcript live without producing a spurious archive. export async function projectSessionEntryLifecycleMutation( - database: OpenClawAgentDatabase, + databaseOptions: OpenClawAgentDatabaseOptions, params: { allowCanonicalRepair?: boolean; archiveDirectory: string; @@ -315,7 +317,9 @@ export async function projectSessionEntryLifecycleMutation( upserts: readonly SessionEntryLifecycleUpsert[]; }, ): Promise { - const store = readSessionEntryStore(database, { + // openclaw-agent-db.ts cache rule: keep handles within synchronous sections. + const removalDatabase = openOpenClawAgentDatabase(databaseOptions); + const store = readSessionEntryStore(removalDatabase, { allowCanonicalRepair: params.allowCanonicalRepair === true, }); const removedEntries: Array<{ archiveTranscript: boolean; entry: SessionEntry }> = []; @@ -326,7 +330,7 @@ export async function projectSessionEntryLifecycleMutation( const sessionKey = removal.exactStoredKey ? removal.sessionKey : removal.sessionKey.trim(); let entry = removal.exactStoredKey || sessionKey ? store[sessionKey] : undefined; if (removal.expectedRawEntryJson !== undefined) { - const currentRawEntryJson = readExactSessionEntryJson(database, sessionKey); + const currentRawEntryJson = readExactSessionEntryJson(removalDatabase, sessionKey); if (currentRawEntryJson !== removal.expectedRawEntryJson) { throw new Error( `SQLite session entry changed before raw lifecycle removal for ${sessionKey}`, @@ -342,7 +346,7 @@ export async function projectSessionEntryLifecycleMutation( if ( !sessionId || !sqliteSessionStateDeleteSnapshotsEqual( - readSessionStateDeleteSnapshot(database.db, sessionId), + readSessionStateDeleteSnapshot(removalDatabase.db, sessionId), removal.expectedTranscriptSnapshot, ) ) { @@ -400,6 +404,8 @@ export async function projectSessionEntryLifecycleMutation( ...(upsert.resetBoundary ? { resetBoundary: upsert.resetBoundary } : {}), }); } + // openclaw-agent-db.ts cache rule: LRU eviction may close idle handles during buildEntry awaits. + const database = openOpenClawAgentDatabase(databaseOptions); const referencedSessionIds = collectProjectedReferencedSessionIds({ database, excludedSessionKeys: changedSessionKeys, diff --git a/src/config/sessions/session-accessor.sqlite-projection.ts b/src/config/sessions/session-accessor.sqlite-projection.ts index e869e9fd5eb5..528cc458a836 100644 --- a/src/config/sessions/session-accessor.sqlite-projection.ts +++ b/src/config/sessions/session-accessor.sqlite-projection.ts @@ -279,8 +279,7 @@ export async function applySessionEntryLifecycleMutation(params: { let materializedRemovalPlans: MaterializedSessionStateDeletePlan[] = []; let removalArchiveMaterializationFailed = false; const committed = await runPreparedSqliteSessionWrite(resolved, async () => { - const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); - projected = await projectSessionEntryLifecycleMutation(database, { + projected = await projectSessionEntryLifecycleMutation(toDatabaseOptions(resolved), { ...(params.allowCanonicalRepair ? { allowCanonicalRepair: true } : {}), archiveDirectory: resolveSqliteTranscriptArchiveDirectory(resolved), removals, diff --git a/src/config/sessions/session-accessor.sqlite-transcript-write.ts b/src/config/sessions/session-accessor.sqlite-transcript-write.ts index 90577cb84102..746b61124fcd 100644 --- a/src/config/sessions/session-accessor.sqlite-transcript-write.ts +++ b/src/config/sessions/session-accessor.sqlite-transcript-write.ts @@ -16,15 +16,11 @@ import type { TranscriptMessageAppendOptions, TranscriptMessageAppendResult, } from "./session-accessor.sqlite-contract.js"; -import { - runPreparedSqliteSessionWrite, - runSqliteSessionDeletionTransaction as runOpenClawAgentWriteTransaction, -} from "./session-accessor.sqlite-deletion.js"; +import { runSqliteSessionDeletionTransaction as runOpenClawAgentWriteTransaction } from "./session-accessor.sqlite-deletion.js"; import { assertSessionEntrySelectionUnchanged } from "./session-accessor.sqlite-entry-equality.js"; import type { ResolvedSessionEntryRow } from "./session-accessor.sqlite-entry-store.js"; import { collectSessionEntryLookupKeys, - deleteLegacySessionEntryRows, readSessionEntryRow, readSessionEntrySelectionSnapshot, readSessionIdentitySnapshot, @@ -375,111 +371,93 @@ export async function appendExpectedSessionTranscriptTurn( ...scope, sessionId: options.expectedSessionId, }); - return await runPreparedSqliteSessionWrite( - resolved, - async () => { - const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); - const preparedEntry = readSessionEntryRow(database, resolved.sessionKey); - if (!sessionMatchesExpectedTranscriptTurn(preparedEntry, options)) { - return { - deletedEntries: [], - commit: () => sqliteSessionTranscriptTurnRebound(preparedEntry, options.sessionFile), - }; + return await runExclusiveSqliteSessionWrite(resolved, async () => { + // openclaw-agent-db.ts cache rule: LRU can close idle handles during shouldAppend awaits. + const preparedEntry = readSessionEntryRow( + openOpenClawAgentDatabase(toDatabaseOptions(resolved)), + resolved.sessionKey, + ); + if (!sessionMatchesExpectedTranscriptTurn(preparedEntry, options)) { + return sqliteSessionTranscriptTurnRebound(preparedEntry, options.sessionFile); + } + const messages = await selectAppendableSqliteTranscriptTurnMessages( + { + agentId: resolved.agentId, + sessionId: options.expectedSessionId, + sessionKey: resolved.sessionKey, + ...(scope.storePath ? { storePath: scope.storePath } : {}), + }, + options.messages, + ); + let result: SqliteExpectedSessionTranscriptTurnResult = sqliteSessionTranscriptTurnRebound( + preparedEntry, + options.sessionFile, + ); + let previousIdentity = new Map(); + let currentIdentity = new Map(); + runOpenClawAgentWriteTransaction((transactionDb) => { + const fresh = readSessionEntryRow(transactionDb, resolved.sessionKey); + if (!sessionMatchesExpectedTranscriptTurn(fresh, options)) { + result = sqliteSessionTranscriptTurnRebound(fresh, options.sessionFile); + return; + } + const appendedMessages: TranscriptMessageAppendResult[] = []; + for (const append of messages) { + const { shouldAppend: _shouldAppend, ...appendOptions } = append; + const appended = appendTranscriptMessageInTransaction(transactionDb, resolved, { + ...appendOptions, + messageAlreadyRedacted: options.atomicGroup === true, + ...((append.cwd ?? options.cwd) ? { cwd: append.cwd ?? options.cwd } : {}), + ...((append.config ?? options.config) ? { config: append.config ?? options.config } : {}), + }); + if (appended) { + appendedMessages.push(appended); + } + } + if ( + options.atomicGroup && + (appendedMessages.length !== messages.length || + appendedMessages.some((message) => message.appended) !== + appendedMessages.every((message) => message.appended)) + ) { + throw new Error("SQLite transcript batch was not wholly inserted or replayed"); } - const messages = await selectAppendableSqliteTranscriptTurnMessages( - { - agentId: resolved.agentId, - sessionId: options.expectedSessionId, - sessionKey: resolved.sessionKey, - ...(scope.storePath ? { storePath: scope.storePath } : {}), - }, - options.messages, - ); - let result: SqliteExpectedSessionTranscriptTurnResult = sqliteSessionTranscriptTurnRebound( - preparedEntry, - options.sessionFile, - ); - let previousIdentity = new Map(); - let currentIdentity = new Map(); - const aliases = readSessionIdentitySnapshot( - database, - preparedEntry.legacyKeys.filter((key) => key !== resolved.sessionKey), - ); - return { - deletedEntries: [...aliases].map(([sessionKey, entry]) => ({ sessionKey, entry })), - commit: () => { - runOpenClawAgentWriteTransaction((transactionDb) => { - const fresh = readSessionEntryRow(transactionDb, resolved.sessionKey); - if (!sessionMatchesExpectedTranscriptTurn(fresh, options)) { - result = sqliteSessionTranscriptTurnRebound(fresh, options.sessionFile); - return; - } - const appendedMessages: TranscriptMessageAppendResult[] = []; - for (const append of messages) { - const { shouldAppend: _shouldAppend, ...appendOptions } = append; - const appended = appendTranscriptMessageInTransaction(transactionDb, resolved, { - ...appendOptions, - messageAlreadyRedacted: options.atomicGroup === true, - ...((append.cwd ?? options.cwd) ? { cwd: append.cwd ?? options.cwd } : {}), - ...((append.config ?? options.config) - ? { config: append.config ?? options.config } - : {}), - }); - if (appended) { - appendedMessages.push(appended); - } - } - if ( - options.atomicGroup && - (appendedMessages.length !== messages.length || - appendedMessages.some((message) => message.appended) !== - appendedMessages.every((message) => message.appended)) - ) { - throw new Error("SQLite transcript batch was not wholly inserted or replayed"); - } - // Later explicit parents can abandon earlier rows. Capture every cursor - // from the final active projection before this atomic transaction commits. - rememberCommittedTranscriptMessageSequencesInTransaction( - transactionDb, - resolved.sessionId, - appendedMessages, - ); + // Later explicit parents can abandon earlier rows. Capture every cursor + // from the final active projection before this atomic transaction commits. + rememberCommittedTranscriptMessageSequencesInTransaction( + transactionDb, + resolved.sessionId, + appendedMessages, + ); - const sessionPatch = buildExpectedTranscriptTurnSessionPatch({ - appendedMessages, - currentEntry: fresh.entry, - expectedSessionState: options.expectedSessionState, - sessionFile: options.sessionFile, - sessionLifecyclePatch: options.sessionLifecyclePatch, - touchSessionEntry: options.touchSessionEntry, - }); - const next = - Object.keys(sessionPatch).length > 0 - ? mergeSessionEntry(fresh.entry, sessionPatch) - : fresh.entry; - if (next !== fresh.entry) { - const identityKeys = collectSessionEntryLookupKeys( - transactionDb, - resolved.sessionKey, - ); - previousIdentity = readSessionIdentitySnapshot(transactionDb, identityKeys); - writeSessionEntry(transactionDb, resolved.sessionKey, next); - deleteLegacySessionEntryRows(transactionDb, fresh.legacyKeys, resolved.sessionKey); - currentIdentity = readSessionIdentitySnapshot(transactionDb, identityKeys); - } - result = { - appendedMessages, - sessionEntry: cloneSessionEntry(next), - sessionFile: options.sessionFile, - }; - }, toDatabaseOptions(resolved)); - emitCommittedSessionIdentityDiff(previousIdentity, currentIdentity); - return result; - }, + const sessionPatch = buildExpectedTranscriptTurnSessionPatch({ + appendedMessages, + currentEntry: fresh.entry, + expectedSessionState: options.expectedSessionState, + sessionFile: options.sessionFile, + sessionLifecyclePatch: options.sessionLifecyclePatch, + touchSessionEntry: options.touchSessionEntry, + }); + const next = + Object.keys(sessionPatch).length > 0 + ? mergeSessionEntry(fresh.entry, sessionPatch) + : fresh.entry; + if (next !== fresh.entry) { + const identityKeys = collectSessionEntryLookupKeys(transactionDb, resolved.sessionKey); + previousIdentity = readSessionIdentitySnapshot(transactionDb, identityKeys); + writeSessionEntry(transactionDb, resolved.sessionKey, next); + currentIdentity = readSessionIdentitySnapshot(transactionDb, identityKeys); + } + result = { + appendedMessages, + sessionEntry: cloneSessionEntry(next), + sessionFile: options.sessionFile, }; - }, - ); + }, toDatabaseOptions(resolved)); + emitCommittedSessionIdentityDiff(previousIdentity, currentIdentity); + return result; + }); } function sqliteSessionTranscriptTurnRebound( @@ -568,16 +546,20 @@ export async function withTranscriptWriteLock( run: (context: SqliteTranscriptWriteLockContext) => Promise | T, ): Promise { const resolved = resolveSqliteTranscriptScope(scope); + const databaseOptions = toDatabaseOptions(resolved); return await runExclusiveSqliteSessionWrite(resolved, async () => { - const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); let transcriptSnapshot: SqliteTranscriptSnapshotState | undefined; return await run({ readEvents: async () => { + // openclaw-agent-db.ts cache rule: LRU eviction closes idle handles across caller awaits. + const database = openOpenClawAgentDatabase(databaseOptions); const snapshot = readTranscriptSnapshot(database, resolved.sessionId); transcriptSnapshot = { kind: "current", rows: snapshot.rows }; return snapshot.events; }, - readMessageFacts: async (params) => readTranscriptMirrorFacts(database, resolved, params), + // openclaw-agent-db.ts cache rule: never retain a handle across caller awaits; LRU may close it. + readMessageFacts: async (params) => + readTranscriptMirrorFacts(openOpenClawAgentDatabase(databaseOptions), resolved, params), replaceEvents: async (events) => { if (transcriptSnapshot?.kind === "stale") { throw new SqliteTranscriptMutationConflictError(resolved.sessionId); @@ -595,7 +577,7 @@ export async function withTranscriptWriteLock( } replaceSqliteTranscriptEventsInTransaction(writeDatabase, resolved, events); return readTranscriptEventRows(writeDatabase, resolved.sessionId); - }, toDatabaseOptions(resolved)); + }, databaseOptions); transcriptSnapshot = { kind: "current", rows: nextSnapshot }; }, appendMessage: async (options) => { @@ -620,7 +602,7 @@ export async function withTranscriptWriteLock( } : { kind: "stale" }; } - }, toDatabaseOptions(resolved)); + }, databaseOptions); transcriptSnapshot = nextSnapshotState; return result as TranscriptMessageAppendResult | undefined; }, @@ -637,7 +619,7 @@ export async function withTranscriptWriteLock( ); messageSeq = readCommittedTranscriptMessageSequence(result); } - }, toDatabaseOptions(resolved)); + }, databaseOptions); return { ...(messageSeq !== undefined ? { messageSeq } : {}), result: result as TranscriptMessageAppendResult | undefined, diff --git a/src/config/sessions/session-history-eviction.test.ts b/src/config/sessions/session-history-eviction.test.ts index c423ea9eae25..4b577022b4cb 100644 --- a/src/config/sessions/session-history-eviction.test.ts +++ b/src/config/sessions/session-history-eviction.test.ts @@ -21,6 +21,7 @@ vi.mock("../../logging/subsystem.js", async () => { import { executeSqliteQuerySync } from "../../infra/kysely-sync.js"; import { beginSessionWorkAdmission } from "../../sessions/session-lifecycle-admission.js"; import { + closeOpenClawAgentDatabaseByPath, closeOpenClawAgentDatabasesForTest, openOpenClawAgentDatabase, } from "../../state/openclaw-agent-db.js"; @@ -61,6 +62,7 @@ describe("SQLite historical session disk budget", () => { }); afterEach(async () => { + vi.restoreAllMocks(); await enforceSqliteSessionHistoryDiskBudget({ storePath, mode: "warn", @@ -208,6 +210,14 @@ describe("SQLite historical session disk budget", () => { settlePhysicalUsage(); const before = await measureSessionPhysicalDiskUsage(storePath); + const databasePath = database().path; + const rm = fs.promises.rm.bind(fs.promises); + const removeArchive = vi.spyOn(fs.promises, "rm").mockImplementation(async (...args) => { + if (args[0] === archivePath) { + expect(closeOpenClawAgentDatabaseByPath(databasePath)).toBe(true); + } + return await rm(...args); + }); const result = await enforceSqliteSessionHistoryDiskBudget({ storePath, mode: "enforce", @@ -219,6 +229,7 @@ describe("SQLite historical session disk budget", () => { expect(result).toMatchObject({ removedEntries: 0, removedFiles: 1 }); expect(fs.existsSync(archivePath ?? "")).toBe(false); + expect(removeArchive).toHaveBeenCalledWith(archivePath); expect( database() .db.prepare("SELECT 1 FROM session_transcript_archives WHERE session_id = ?") @@ -394,6 +405,36 @@ describe("SQLite historical session disk budget", () => { }); }); + it("inspects history after the archive probe loses its cached handle", async () => { + await createHistoricalTranscript({ + content: "inspect retained history", + nextSessionId: "inspect-live", + sessionId: "inspect-old", + sessionKey: "agent:main:inspect-history", + updatedAt: 1, + }); + const databasePath = database().path; + const diskBudget = await import("./disk-budget.js"); + const probe = diskBudget.hasRetainedSessionTranscriptArchives; + const probeSpy = vi + .spyOn(diskBudget, "hasRetainedSessionTranscriptArchives") + .mockImplementation(async (pathname) => { + const retained = await probe(pathname); + expect(closeOpenClawAgentDatabaseByPath(databasePath)).toBe(true); + return retained; + }); + + await expect( + inspectSqliteSessionHistoryDiskBudget({ + storePath, + mode: "enforce", + maintenance: { maxDiskBytes: 1, highWaterBytes: 0 }, + }), + ).resolves.toMatchObject({ wouldMutate: true }); + expect(probeSpy).toHaveBeenCalledOnce(); + expect(sessionExists("inspect-old")).toBe(true); + }); + it("warn mode reports physical overage without extracting or deleting history", async () => { await createHistoricalTranscript({ content: "warn history", diff --git a/src/config/sessions/session-history-eviction.ts b/src/config/sessions/session-history-eviction.ts index 1d1c4d944ded..61c3ee47a4e1 100644 --- a/src/config/sessions/session-history-eviction.ts +++ b/src/config/sessions/session-history-eviction.ts @@ -12,6 +12,7 @@ import { openOpenClawAgentDatabase, runOpenClawAgentWriteTransaction, type OpenClawAgentDatabase, + type OpenClawAgentDatabaseOptions, } from "../../state/openclaw-agent-db.js"; import { hasRetainedSessionTranscriptArchives, @@ -98,20 +99,17 @@ export async function inspectSqliteSessionHistoryDiskBudget( sessionKey: "", storePath: params.storePath, }); - const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); + const databaseOptions = toDatabaseOptions(resolved); if ( - hasCanonicalSessionTranscriptArchives(database) || + hasCanonicalSessionTranscriptArchives(databaseOptions) || (await hasRetainedSessionTranscriptArchives(params.storePath)) ) { return { diskBudget, wouldMutate: true }; } const candidates = readHistoricalSessionIds({ - database, - protectedSessionIds: collectInitialProtectedHistoricalSessionIds({ - database, - preserveRecentMs: params.maintenance.preserveRecentMs, - storePath: params.storePath, - }), + databaseOptions, + preserveRecentMs: params.maintenance.preserveRecentMs, + storePath: params.storePath, }); return { diskBudget, wouldMutate: candidates.length > 0 }; } @@ -127,18 +125,6 @@ function collectProtectedHistoricalSessionIds(params: { return protectedSessionIds; } -function collectInitialProtectedHistoricalSessionIds(params: { - database: OpenClawAgentDatabase; - preserveRecentMs?: number | null; - storePath: string; -}): Set { - const protectedSessionIds = collectProtectedHistoricalSessionIds(params); - for (const sessionId of collectRecentSessionHistoryIds(params)) { - protectedSessionIds.add(sessionId); - } - return protectedSessionIds; -} - function collectRecentSessionHistoryIds(params: { database: OpenClawAgentDatabase; preserveRecentMs?: number | null; @@ -277,21 +263,31 @@ export function collectAdmissionProtectedSessionIds(params: { } function readHistoricalSessionIds(params: { - database: OpenClawAgentDatabase; - protectedSessionIds: ReadonlySet; + databaseOptions: OpenClawAgentDatabaseOptions; + preserveRecentMs?: number | null; + storePath: string; }): string[] { - const db = getSessionKysely(params.database.db); + // openclaw-agent-db.ts cache rule: LRU eviction closes idle handles across awaits. + const database = openOpenClawAgentDatabase(params.databaseOptions); + const scope = { ...params, database }; + const protectedSessionIds = collectProtectedHistoricalSessionIds(scope); + for (const sessionId of collectRecentSessionHistoryIds(scope)) { + protectedSessionIds.add(sessionId); + } + const db = getSessionKysely(database.db); return executeSqliteQuerySync( - params.database.db, + database.db, db .selectFrom("session_windows") .select("session_id") .orderBy("updated_at", "asc") .orderBy("session_id", "asc"), - ).rows.flatMap((row) => (params.protectedSessionIds.has(row.session_id) ? [] : [row.session_id])); + ).rows.flatMap((row) => (protectedSessionIds.has(row.session_id) ? [] : [row.session_id])); } -function reclaimSqliteFreePages(database: OpenClawAgentDatabase): void { +function reclaimSqliteFreePages(databaseOptions: OpenClawAgentDatabaseOptions): void { + // openclaw-agent-db.ts cache rule: LRU eviction closes idle handles across awaits. + const database = openOpenClawAgentDatabase(databaseOptions); // Committed row deletion first lands in the WAL. TRUNCATE makes that shrink immediately; // incremental vacuum can then return free tail pages from the main file without a rewrite. database.walMaintenance.checkpoint(); @@ -305,7 +301,11 @@ function reclaimSqliteFreePages(database: OpenClawAgentDatabase): void { database.walMaintenance.checkpoint(); } -function hasCanonicalSessionTranscriptArchives(database: OpenClawAgentDatabase): boolean { +function hasCanonicalSessionTranscriptArchives( + databaseOptions: OpenClawAgentDatabaseOptions, +): boolean { + // openclaw-agent-db.ts cache rule: LRU eviction closes idle handles across awaits. + const database = openOpenClawAgentDatabase(databaseOptions); const db = getSessionKysely(database.db); const table = executeSqliteQuerySync( database.db, @@ -331,8 +331,10 @@ function hasCanonicalSessionTranscriptArchives(database: OpenClawAgentDatabase): } function readUnpublishedSessionTranscriptArchiveNames( - database: OpenClawAgentDatabase, + databaseOptions: OpenClawAgentDatabaseOptions, ): Set { + // openclaw-agent-db.ts cache rule: LRU eviction closes idle handles across awaits. + const database = openOpenClawAgentDatabase(databaseOptions); const db = getSessionKysely(database.db); const table = executeSqliteQuerySync( database.db, @@ -358,16 +360,18 @@ function readUnpublishedSessionTranscriptArchiveNames( async function pruneCanonicalSessionTranscriptArchivesToHighWater(params: { archiveDirectory: string; - database: OpenClawAgentDatabase; + databaseOptions: OpenClawAgentDatabaseOptions; highWaterBytes: number; storePath: string; }): Promise<{ removedFiles: number; usage: SessionPhysicalDiskUsage }> { let usage = await measureSessionPhysicalDiskUsage(params.storePath); let removedFiles = 0; while (usage.totalBytes > params.highWaterBytes) { - const db = getSessionKysely(params.database.db); + // openclaw-agent-db.ts cache rule: LRU eviction closes idle handles across awaits. + const database = openOpenClawAgentDatabase(params.databaseOptions); + const db = getSessionKysely(database.db); const row = executeSqliteQuerySync( - params.database.db, + database.db, db .selectFrom("session_transcript_archives") .select(["archive_name", "generation", "session_id"]) @@ -397,20 +401,17 @@ async function pruneCanonicalSessionTranscriptArchivesToHighWater(params: { break; } } - runOpenClawAgentWriteTransaction( - (transactionDb) => { - const transactionKysely = getSessionKysely(transactionDb.db); - executeSqliteQuerySync( - transactionDb.db, - transactionKysely - .deleteFrom("session_transcript_archives") - .where("session_id", "=", row.session_id) - .where("generation", "=", row.generation), - ); - }, - { agentId: params.database.agentId, path: params.database.path }, - ); - reclaimSqliteFreePages(params.database); + runOpenClawAgentWriteTransaction((transactionDb) => { + const transactionKysely = getSessionKysely(transactionDb.db); + executeSqliteQuerySync( + transactionDb.db, + transactionKysely + .deleteFrom("session_transcript_archives") + .where("session_id", "=", row.session_id) + .where("generation", "=", row.generation), + ); + }, params.databaseOptions); + reclaimSqliteFreePages(params.databaseOptions); usage = await measureSessionPhysicalDiskUsage(params.storePath); } return { removedFiles, usage }; @@ -418,7 +419,7 @@ async function pruneCanonicalSessionTranscriptArchivesToHighWater(params: { async function pruneAllSessionTranscriptArchivesToHighWater(params: { archiveDirectory: string; - database: OpenClawAgentDatabase; + databaseOptions: OpenClawAgentDatabaseOptions; highWaterBytes: number; storePath: string; }): Promise<{ removedFiles: number; usage: SessionPhysicalDiskUsage }> { @@ -426,14 +427,14 @@ async function pruneAllSessionTranscriptArchivesToHighWater(params: { removedFiles: 0, usage: await measureSessionPhysicalDiskUsage(params.storePath), }; - if (hasCanonicalSessionTranscriptArchives(params.database)) { + if (hasCanonicalSessionTranscriptArchives(params.databaseOptions)) { canonical = await pruneCanonicalSessionTranscriptArchivesToHighWater(params); } if (canonical.usage.totalBytes <= params.highWaterBytes) { return canonical; } const legacy = await pruneSessionTranscriptArchivesToHighWater({ - excludeNames: readUnpublishedSessionTranscriptArchiveNames(params.database), + excludeNames: readUnpublishedSessionTranscriptArchiveNames(params.databaseOptions), highWaterBytes: params.highWaterBytes, storePath: params.storePath, }); @@ -567,10 +568,10 @@ async function enforceSessionHistoryMaintenanceSerialized( sessionKey: "", storePath: params.storePath, }); - const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); + const databaseOptions = toDatabaseOptions(resolved); const archiveDirectory = resolveSqliteTranscriptArchiveDirectory(resolved); let usage: SessionPhysicalDiskUsage = await runExclusiveSqliteSessionWrite(resolved, async () => { - reclaimSqliteFreePages(database); + reclaimSqliteFreePages(databaseOptions); return await measureSessionPhysicalDiskUsage(params.storePath); }); let removedEntries = 0; @@ -581,7 +582,7 @@ async function enforceSessionHistoryMaintenanceSerialized( const archiveSweep = await runExclusiveSqliteSessionWrite(resolved, async () => pruneAllSessionTranscriptArchivesToHighWater({ archiveDirectory, - database, + databaseOptions, highWaterBytes, storePath: params.storePath, }), @@ -590,12 +591,9 @@ async function enforceSessionHistoryMaintenanceSerialized( usage = archiveSweep.usage; } const candidates = readHistoricalSessionIds({ - database, - protectedSessionIds: collectInitialProtectedHistoricalSessionIds({ - database, - preserveRecentMs: params.maintenance.preserveRecentMs, - storePath: params.storePath, - }), + databaseOptions, + preserveRecentMs: params.maintenance.preserveRecentMs, + storePath: params.storePath, }); for (const sessionId of candidates) { @@ -607,13 +605,15 @@ async function enforceSessionHistoryMaintenanceSerialized( identities: [sessionId], run: async () => { const plan = await runExclusiveSqliteSessionWrite(resolved, async () => { + // openclaw-agent-db.ts cache rule: LRU eviction closes idle handles across awaits. + const database = openOpenClawAgentDatabase(databaseOptions); const protectedBeforeArchive = collectCandidateProtectedHistoricalSessionIds({ database, preserveRecentMs: params.maintenance.preserveRecentMs, sessionId, storePath: params.storePath, }); - const candidate = planSessionStateDeleteIfUnreferenced({ + return planSessionStateDeleteIfUnreferenced({ archiveDirectory, archiveTranscript: true, database, @@ -621,10 +621,6 @@ async function enforceSessionHistoryMaintenanceSerialized( referencedSessionIds: protectedBeforeArchive, sessionId, }); - if (!candidate) { - return null; - } - return candidate; }); if (!plan) { return null; @@ -656,7 +652,7 @@ async function enforceSessionHistoryMaintenanceSerialized( .select("session_id") .where("session_id", "=", sessionId), ).rows.length === 0; - }, toDatabaseOptions(resolved)); + }, databaseOptions); if (!deleted) { return null; } @@ -664,7 +660,7 @@ async function enforceSessionHistoryMaintenanceSerialized( // The deletion is committed; checkpoint/incremental-vacuum failure // must not hide it from accounting or observers. Pages reclaim on // a later pass instead. - reclaimSqliteFreePages(database); + reclaimSqliteFreePages(databaseOptions); } catch { // Best-effort reclamation only. } @@ -701,7 +697,7 @@ async function enforceSessionHistoryMaintenanceSerialized( const repruned = await runExclusiveSqliteSessionWrite(resolved, async () => pruneAllSessionTranscriptArchivesToHighWater({ archiveDirectory, - database, + databaseOptions, highWaterBytes, storePath: params.storePath, }), @@ -717,7 +713,7 @@ async function enforceSessionHistoryMaintenanceSerialized( const finalPrune = await runExclusiveSqliteSessionWrite(resolved, async () => pruneAllSessionTranscriptArchivesToHighWater({ archiveDirectory, - database, + databaseOptions, highWaterBytes, storePath: params.storePath, }), diff --git a/src/config/sessions/session-transcript-reconcile.ts b/src/config/sessions/session-transcript-reconcile.ts index 97d65970b72f..18c0ef59c123 100644 --- a/src/config/sessions/session-transcript-reconcile.ts +++ b/src/config/sessions/session-transcript-reconcile.ts @@ -417,10 +417,13 @@ export async function waitForSessionTranscriptProjection( ): Promise { const resolved = resolveSqliteTranscriptReadScope(scope); const databaseOptions = toDatabaseOptions(resolved); - const database = openOpenClawAgentDatabase(databaseOptions); + // openclaw-agent-db.ts cache rule: LRU eviction closes idle handles across polling awaits. while ( isSessionTranscriptIndexReconcileRunning(databaseOptions) && - sessionTranscriptIndexNeedsReconcile(database.db, resolved.sessionId) + sessionTranscriptIndexNeedsReconcile( + openOpenClawAgentDatabase(databaseOptions).db, + resolved.sessionId, + ) ) { await delay(PROJECTION_READY_POLL_MS); }