From ef91ce57124b09d22b7ffd3713b140d2462dc4be Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sun, 19 Jul 2026 03:38:51 -0700 Subject: [PATCH] feat: keep reset session history searchable (#111194) * feat(sessions): retain reset history in sqlite with physical disk budget * fix(sessions): satisfy test-types, knip export scan, and docs map * fix(sessions): align behavioral suites and flip proof with retained history, route-aware cleanup plans --- docs/concepts/main-session.md | 16 +- docs/docs_map.md | 2 +- .../session-management-compaction.md | 12 +- src/config/sessions/artifacts.ts | 5 + src/config/sessions/cleanup-service.ts | 44 +- src/config/sessions/disk-budget.test.ts | 48 +- src/config/sessions/disk-budget.ts | 89 ++++ .../session-accessor.conformance.test.ts | 38 +- .../sessions/session-accessor.lifecycle.ts | 8 - .../session-accessor.sqlite-archive.ts | 11 + .../session-accessor.sqlite-disk-budget.ts | 108 ----- .../sessions/session-accessor.sqlite-entry.ts | 12 + ...session-accessor.sqlite-lifecycle-state.ts | 51 +- .../session-accessor.sqlite-lifecycle.ts | 253 +++++++--- .../session-accessor.sqlite-maintenance.ts | 178 +------ .../sessions/session-accessor.sqlite.ts | 1 - src/config/sessions/session-accessor.ts | 1 - .../sessions/session-history-eviction.test.ts | 331 +++++++++++++ .../sessions/session-history-eviction.ts | 444 ++++++++++++++++++ src/config/sessions/store-maintenance.ts | 9 +- .../store.pruning.integration.test.ts | 95 +++- .../store.session-lifecycle-mutation.test.ts | 182 +++++-- .../server.sessions.reset-hooks.test.ts | 11 +- src/gateway/server.sessions.store-rpc.test.ts | 4 +- .../sqlite-sessions-transcripts-flip-proof.ts | 14 +- ...anscripts-flip-proof.built-cli.e2e.test.ts | 6 +- ...essions-transcripts-flip-proof.e2e.test.ts | 6 +- 27 files changed, 1497 insertions(+), 482 deletions(-) delete mode 100644 src/config/sessions/session-accessor.sqlite-disk-budget.ts create mode 100644 src/config/sessions/session-history-eviction.test.ts create mode 100644 src/config/sessions/session-history-eviction.ts diff --git a/docs/concepts/main-session.md b/docs/concepts/main-session.md index c12ebeb89598..893f3b31178d 100644 --- a/docs/concepts/main-session.md +++ b/docs/concepts/main-session.md @@ -60,20 +60,26 @@ continuity comes from layers around it: is enabled by default; any configured DM isolation turns it off unless you opt in explicitly. See [Memory configuration](/reference/memory-config). -## A rolling session, not an immortal one +## A rolling session with durable history The main session rolls forward through resets and compaction rather than -growing forever: +making the model carry its entire history at once: - By default there is no automatic reset; compaction keeps the active context bounded while preserving the rolling session. Daily and idle resets are opt-in (see [Session management](/concepts/session)). On `/new` and `/reset`, the tail of the ending conversation is saved to daily memory notes, and the - next session re-primes recent notes. + next session re-primes recent notes. Reset assigns a new live session id but + keeps the previous SQLite transcript searchable under the same main-session + key. - When the conversation approaches the context window, compaction summarizes and continues in place — the transcript history stays in the session store. -- The per-agent session store keeps archived transcripts until a disk budget - (default 10 GB) evicts the oldest ones. +- Session lists show the current live conversation, not every historical + session id behind it. +- When the per-agent store's physical database, WAL, and session artifacts + exceed the disk budget (default 10 GB), OpenClaw extracts the oldest + unreferenced history to a verified compressed archive before removing its + database rows. Live, routed, and in-flight sessions are never budget victims. ## When you want isolation instead diff --git a/docs/docs_map.md b/docs/docs_map.md index fd06386245d1..e65d372769e8 100644 --- a/docs/docs_map.md +++ b/docs/docs_map.md @@ -2493,7 +2493,7 @@ Do not edit it by hand; run `pnpm docs:map:gen`. - H2: Home - H2: What flows into the main session - H2: Memory across resets and conversations - - H2: A rolling session, not an immortal one + - H2: A rolling session with durable history - H2: When you want isolation instead - H2: Related diff --git a/docs/reference/session-management-compaction.md b/docs/reference/session-management-compaction.md index f34b614ff420..ceb466d38d10 100644 --- a/docs/reference/session-management-compaction.md +++ b/docs/reference/session-management-compaction.md @@ -53,19 +53,15 @@ Per agent, on the Gateway host (resolved via `src/config/sessions.ts`): | `maxDiskBytes` | `10gb` | per-agent sessions disk budget; `false` disables | | `highWaterBytes` | 80% of `maxDiskBytes` | target after budget cleanup | -Archived transcripts are kept by default and compressed with zstd (`*.jsonl...zst`) when the runtime supports it, so deleting or resetting a session never silently discards conversation history. The disk budget evicts the oldest archives first, before touching live sessions. +Reset advances the live `sessionKey -> sessionId` mapping but keeps the previous SQLite session, transcript, trajectory, and search rows. That history remains searchable under the same session key; ordinary entry and session lists show only the new live mapping. Retained reset history is bounded by the disk budget, not by `resetArchiveRetention`, which only ages archive artifacts. Explicit deletion is different: it writes and verifies a compressed transcript archive (`*.jsonl.deleted..zst` when zstd is available) before removing the deleted session's rows. -Active SQLite enforcement of `maxDiskBytes` measures session-row JSON plus transcript-event JSON bytes per session; legacy offline-maintenance enforcement measures files in the selected sessions directory. +`maxDiskBytes` enforcement uses physical bytes: the per-agent SQLite main file, its `-wal` file, and counted files in the agent sessions directory. It never estimates row JSON sizes or subtracts logical row sizes from that total. Gateway model-run probe sessions (keys matching `agent:*:explicit:model-run-`) get a separate, fixed `24h` retention. This pruning is pressure-gated: it only runs when session-entry maintenance/cap pressure is reached, and only before the global stale-entry cleanup/cap step. Other explicit sessions do not use this retention. -Enforcement order for disk-budget cleanup (`mode: "enforce"`): +When combined physical usage exceeds `maxDiskBytes`, `mode: "enforce"` first reclaims checkpointable database space, then removes the oldest retained reset/delete archives. If usage is still above `highWaterBytes`, it walks historical SQLite sessions by `sessions.updated_at`, oldest first. Historical means the session id is not referenced by a live session entry, a route target, or an admitted/in-flight run. For each victim, cleanup writes, fsyncs, and reads back the compressed archive before a write transaction removes the session row and its transcript, trajectory, active, index, and FTS projections. This includes sessions that contain trajectory events but no transcript events. Cleanup rechecks route, entry, and admission references at deletion time, remeasures physical usage after each archive or session victim, and stops at `highWaterBytes`. -1. Remove oldest archived transcript artifacts, orphan legacy artifacts, or orphan trajectory artifacts first. -2. If still above target, evict oldest session entries and their transcript rows or trajectory artifacts. -3. Repeat until usage is at or below `highWaterBytes`. - -`mode: "warn"` reports potential evictions without mutating the store or files. +Committed writes and deletion first land in the WAL. Cleanup checkpoints it so the WAL can shrink immediately, then uses incremental vacuum to return eligible free tail pages from the main file; pages that are not yet reclaimable stay in the main file and therefore remain counted on the next physical measurement. `mode: "warn"` reports the current physical overage without checkpointing, writing an archive, or deleting rows. Run maintenance on demand: diff --git a/src/config/sessions/artifacts.ts b/src/config/sessions/artifacts.ts index 6c484cf33a80..2a5d5769d886 100644 --- a/src/config/sessions/artifacts.ts +++ b/src/config/sessions/artifacts.ts @@ -38,6 +38,11 @@ export function isSessionArchiveArtifactName(fileName: string): boolean { ); } +/** Returns true for retained reset/delete transcript archives counted by the session budget. */ +export function isRetainedSessionTranscriptArchiveName(fileName: string): boolean { + return hasArchiveSuffix(fileName, "deleted") || hasArchiveSuffix(fileName, "reset"); +} + /** Returns true for migration rollback archives retained beside their legacy source. */ export function isMigrationArchiveArtifactName(fileName: string): boolean { return MIGRATION_ARCHIVE_RE.test(fileName); diff --git a/src/config/sessions/cleanup-service.ts b/src/config/sessions/cleanup-service.ts index 2d816661fdf5..14167d2ed7e0 100644 --- a/src/config/sessions/cleanup-service.ts +++ b/src/config/sessions/cleanup-service.ts @@ -19,10 +19,13 @@ import { applySessionEntryLifecycleMutation, listSessionEntries, loadTranscriptEventsSync, - previewSessionDiskBudget, purgeDeletedAgentSessionEntries, type SessionEntryLifecycleRemoval, } from "./session-accessor.js"; +import { + enforceSqliteSessionHistoryDiskBudget, + inspectSqliteSessionHistoryDiskBudget, +} from "./session-history-eviction.js"; import { resolveSqliteTargetFromSessionStorePath } from "./session-sqlite-target.js"; import { cloneSessionStoreRecord } from "./store-cache.js"; import { collectSessionMaintenancePreserveKeysForStore } from "./store-maintenance-preserve.js"; @@ -435,15 +438,13 @@ async function previewStoreCleanup(params: { keys: dmScopeRetiredKeys, }); const diskBudgetPreview = fs.existsSync(resolveCleanupSqlitePath(params.target)) - ? previewSessionDiskBudget({ + ? await inspectSqliteSessionHistoryDiskBudget({ agentId: params.target.agentId, - store: previewStore, storePath: params.target.storePath, - activeSessionKey: params.activeKey, - preserveKeys: preserveSessionKeys, + mode: params.mode, maintenance: params.maintenance, }) - : { diskBudget: null, removedKeys: new Set() }; + : { diskBudget: null, wouldMutate: false }; const diskBudget = diskBudgetPreview.diskBudget; const unreferencedArtifacts = await pruneUnreferencedSessionArtifacts({ store: previewStore, @@ -452,7 +453,7 @@ async function previewStoreCleanup(params: { dryRun: true, excludeCanonicalPaths: entryCleanupArtifactPaths, }); - const budgetEvictedKeys = diskBudgetPreview.removedKeys; + const budgetEvictedKeys = new Set(); const beforeCount = Object.keys(beforeStore).length; const afterPreviewCount = Object.keys(previewStore).length; const wouldMutate = @@ -463,7 +464,8 @@ async function previewStoreCleanup(params: { capped > 0 || unreferencedArtifacts.removedFiles > 0 || (diskBudget?.removedEntries ?? 0) > 0 || - (diskBudget?.removedFiles ?? 0) > 0; + (diskBudget?.removedFiles ?? 0) > 0 || + diskBudgetPreview.wouldMutate; const summary: SessionCleanupSummary = { agentId: params.target.agentId, @@ -606,6 +608,12 @@ export async function runSessionsCleanup(params: { freedBytes: 0, olderThanMs: maintenance.pruneAfterMs, }); + const appliedDiskBudget = await enforceSqliteSessionHistoryDiskBudget({ + agentId: target.agentId, + storePath: target.storePath, + mode, + maintenance, + }); const preview = previewResults.find( (result) => result.summary.storePath === target.storePath, ); @@ -631,8 +639,16 @@ export async function runSessionsCleanup(params: { }), dryRun: false, unreferencedArtifacts, + diskBudget: appliedDiskBudget, wouldMutate: - (preview?.summary.wouldMutate ?? false) || unreferencedArtifacts.removedFiles > 0, + removedSessionKeys.size > 0 || + unreferencedArtifacts.removedFiles > 0 || + (appliedDiskBudget?.removedEntries ?? 0) > 0 || + (appliedDiskBudget?.removedFiles ?? 0) > 0 || + // Checkpoint/incremental-vacuum reclamation mutates the store + // even when no session or archive was removed. + (appliedDiskBudget != null && + appliedDiskBudget.totalBytesAfter < appliedDiskBudget.totalBytesBefore), applied: true, appliedCount: lifecycleResult.afterCount, } @@ -649,7 +665,7 @@ export async function runSessionsCleanup(params: { pruned: appliedReport.pruned, capped: appliedReport.capped, unreferencedArtifacts, - diskBudget: appliedReport.diskBudget, + diskBudget: appliedDiskBudget, wouldMutate: missingApplied > 0 || dmScopeRetiredApplied > 0 || @@ -657,8 +673,12 @@ export async function runSessionsCleanup(params: { appliedReport.pruned > 0 || appliedReport.capped > 0 || unreferencedArtifacts.removedFiles > 0 || - (appliedReport.diskBudget?.removedEntries ?? 0) > 0 || - (appliedReport.diskBudget?.removedFiles ?? 0) > 0, + (appliedDiskBudget?.removedEntries ?? 0) > 0 || + (appliedDiskBudget?.removedFiles ?? 0) > 0 || + // Checkpoint/incremental-vacuum reclamation mutates the store + // even when no session or archive was removed. + (appliedDiskBudget != null && + appliedDiskBudget.totalBytesAfter < appliedDiskBudget.totalBytesBefore), applied: true, appliedCount: lifecycleResult.afterCount, }; diff --git a/src/config/sessions/disk-budget.test.ts b/src/config/sessions/disk-budget.test.ts index a25ad7c2856a..7395f754741e 100644 --- a/src/config/sessions/disk-budget.test.ts +++ b/src/config/sessions/disk-budget.test.ts @@ -10,7 +10,12 @@ import { resolveTrajectoryPointerFilePath, } from "../../trajectory/paths.js"; import { formatSessionArchiveTimestamp } from "./artifacts.js"; -import { enforceSessionDiskBudget, pruneUnreferencedSessionArtifacts } from "./disk-budget.js"; +import { + enforceSessionDiskBudget, + measureSessionPhysicalDiskUsage, + pruneUnreferencedSessionArtifacts, +} from "./disk-budget.js"; +import { resolveSqliteTargetFromSessionStorePath } from "./session-sqlite-target.js"; import { saveSessionStore } from "./store.js"; import type { SessionEntry } from "./types.js"; @@ -54,6 +59,47 @@ function refreshPathBeforeSecondStat(targetPath: string): ReturnType { + it("counts the SQLite main file and WAL as physical session usage", async () => { + await withTempDir({ prefix: "openclaw-disk-budget-sqlite-" }, async (dir) => { + const storePath = path.join(dir, "sessions.json"); + const databasePath = resolveSqliteTargetFromSessionStorePath(storePath).path; + if (!databasePath) { + throw new Error("expected a SQLite database path"); + } + await fs.writeFile(databasePath, Buffer.alloc(321)); + await fs.writeFile(`${databasePath}-wal`, Buffer.alloc(654)); + + const usage = await measureSessionPhysicalDiskUsage(storePath); + + expect(usage).toEqual({ + databaseMainBytes: 321, + databaseWalBytes: 654, + sessionFilesBytes: 0, + totalBytes: 975, + }); + }); + }); + + it("excludes migration archives from physical SQLite usage (#106875)", async () => { + await withTempDir({ prefix: "openclaw-disk-budget-sqlite-" }, async (dir) => { + const storePath = path.join(dir, "sessions.json"); + const databasePath = resolveSqliteTargetFromSessionStorePath(storePath).path; + if (!databasePath) { + throw new Error("expected a SQLite database path"); + } + await fs.writeFile(databasePath, Buffer.alloc(100)); + // Rollback archives are recovery artifacts outside the session budget; + // counting them would evict live history to pay for unreclaimable bytes. + await fs.writeFile(path.join(dir, "legacy.jsonl.migrated"), Buffer.alloc(4096)); + await fs.writeFile(path.join(dir, "legacy.jsonl.migrated.2"), Buffer.alloc(4096)); + + const usage = await measureSessionPhysicalDiskUsage(storePath); + + expect(usage.totalBytes).toBe(100); + expect(usage.sessionFilesBytes).toBe(0); + }); + }); + it("excludes migration archives from the session disk budget (#106875)", async () => { await withTempDir({ prefix: "openclaw-disk-budget-" }, async (dir) => { const storePath = path.join(dir, "sessions.json"); diff --git a/src/config/sessions/disk-budget.ts b/src/config/sessions/disk-budget.ts index efc0d33e2966..2733a06d493b 100644 --- a/src/config/sessions/disk-budget.ts +++ b/src/config/sessions/disk-budget.ts @@ -14,12 +14,14 @@ import { isCompactionCheckpointTranscriptFileName, isMigrationArchiveArtifactName, isPrimarySessionTranscriptFileName, + isRetainedSessionTranscriptArchiveName, isSessionArchiveArtifactName, isSessionStoreTempArtifactName, SESSION_STORE_TEMP_STALE_MS, isTrajectorySessionArtifactName, } from "./artifacts.js"; import { resolveSessionFilePath } from "./paths.js"; +import { resolveSqliteTargetFromSessionStorePath } from "./session-sqlite-target.js"; import { projectSessionStoreForPersistence } from "./skill-prompt-blobs.js"; import { shouldPreserveMaintenanceEntry } from "./store-maintenance.js"; import type { SessionEntry } from "./types.js"; @@ -47,6 +49,13 @@ export type SessionUnreferencedArtifactSweepResult = { olderThanMs: number; }; +export type SessionPhysicalDiskUsage = { + databaseMainBytes: number; + databaseWalBytes: number; + sessionFilesBytes: number; + totalBytes: number; +}; + type SessionDiskBudgetLogger = { warn: (message: string, context?: Record) => void; info: (message: string, context?: Record) => void; @@ -248,6 +257,86 @@ async function readSessionsDirFiles(sessionsDir: string): Promise Boolean(file)); } +async function readSqliteDatabaseFiles(storePath: string): Promise { + const databasePath = resolveSqliteTargetFromSessionStorePath(storePath).path; + if (!databasePath) { + return []; + } + const files: SessionsDirFileStat[] = []; + for (const filePath of [databasePath, `${databasePath}-wal`]) { + const stat = await fs.promises.stat(filePath).catch(() => null); + if (!stat?.isFile()) { + continue; + } + files.push({ + path: filePath, + canonicalPath: canonicalizePathForComparison(filePath), + name: path.basename(filePath), + size: stat.size, + mtimeMs: stat.mtimeMs, + }); + } + return files; +} + +/** Measures current physical session artifacts plus the agent SQLite main file and WAL. */ +export async function measureSessionPhysicalDiskUsage( + storePath: string, +): Promise { + const sessionsDirFiles = await readSessionsDirFiles(path.dirname(storePath)); + const promptBlobFiles = await readSessionPromptBlobFiles(path.dirname(storePath)); + const databaseFiles = await readSqliteDatabaseFiles(storePath); + const databasePath = resolveSqliteTargetFromSessionStorePath(storePath).path; + const databaseMainPath = databasePath ? canonicalizePathForComparison(databasePath) : undefined; + const databaseWalPath = databasePath + ? canonicalizePathForComparison(`${databasePath}-wal`) + : undefined; + const uniqueFiles = new Map(); + for (const file of [...sessionsDirFiles, ...promptBlobFiles, ...databaseFiles]) { + uniqueFiles.set(file.canonicalPath, file); + } + const databaseMainBytes = databaseMainPath ? (uniqueFiles.get(databaseMainPath)?.size ?? 0) : 0; + const databaseWalBytes = databaseWalPath ? (uniqueFiles.get(databaseWalPath)?.size ?? 0) : 0; + const totalBytes = [...uniqueFiles.values()].reduce((sum, file) => sum + file.size, 0); + return { + databaseMainBytes, + databaseWalBytes, + sessionFilesBytes: totalBytes - databaseMainBytes - databaseWalBytes, + totalBytes, + }; +} + +export async function hasRetainedSessionTranscriptArchives(storePath: string): Promise { + const files = await readSessionsDirFiles(path.dirname(storePath)); + return files.some((file) => isRetainedSessionTranscriptArchiveName(file.name)); +} + +/** Removes oldest retained reset/delete archives, remeasuring physical usage after each file. */ +export async function pruneSessionTranscriptArchivesToHighWater(params: { + highWaterBytes: number; + storePath: string; +}): Promise<{ removedFiles: number; usage: SessionPhysicalDiskUsage }> { + // Oldest-first is the hard-cap sacrifice order: under extreme pressure this + // may prune an archive the current pass just extracted, which is preferred + // over evicting additional sessions' searchable rows to spare a copy. + const files = (await readSessionsDirFiles(path.dirname(params.storePath))) + .filter((file) => isRetainedSessionTranscriptArchiveName(file.name)) + .toSorted((left, right) => left.mtimeMs - right.mtimeMs); + let usage = await measureSessionPhysicalDiskUsage(params.storePath); + let removedFiles = 0; + for (const file of files) { + if (usage.totalBytes <= params.highWaterBytes) { + break; + } + if ((await removeFileIfExists(file.path)) <= 0) { + continue; + } + removedFiles += 1; + usage = await measureSessionPhysicalDiskUsage(params.storePath); + } + return { removedFiles, usage }; +} + async function readSessionPromptBlobFiles(sessionsDir: string): Promise { const root = path.join(sessionsDir, "skills-prompts", "sha256"); const prefixEntries = await fs.promises.readdir(root, { withFileTypes: true }).catch(() => []); diff --git a/src/config/sessions/session-accessor.conformance.test.ts b/src/config/sessions/session-accessor.conformance.test.ts index 70f842d8b4a7..01c5fd9eef5a 100644 --- a/src/config/sessions/session-accessor.conformance.test.ts +++ b/src/config/sessions/session-accessor.conformance.test.ts @@ -441,7 +441,12 @@ describe.each([publicAccessorAdapter, sqliteAdapter])( orphanTranscriptMinAgeMs: 300_000, nowMs, }), - ).resolves.toEqual({ removedEntries: 2, archivedTranscriptArtifacts: 2 }); + ).resolves.toEqual({ + // Only the removed entry's transcript is archived: the orphan's route + // row still targets it, and route-referenced history is retained. + removedEntries: 2, + archivedTranscriptArtifacts: 1, + }); expect( adapter.loadSessionEntry(scopedEntry("agent:main:lifecycle-cleanup-missing")), @@ -500,7 +505,11 @@ describe.each([publicAccessorAdapter, sqliteAdapter])( file.startsWith("orphan-lifecycle.jsonl.deleted."), ); expect(removedArchive).toBeDefined(); - expect(orphanArchive).toBeDefined(); + // Route-referenced orphan history is retained in SQLite, not archived. + expect(orphanArchive).toBeUndefined(); + await expect( + adapter.loadTranscriptEvents(scopedTranscript("agent:main:orphan", "orphan-lifecycle")), + ).resolves.not.toEqual([]); expect( readSessionArchiveContentSync( path.join(path.dirname(cleanupStorePath), removedArchive ?? ""), @@ -1616,7 +1625,7 @@ describe("sqlite session normalization", () => { }); }); - it("evicts old SQLite transcript rows only when no remaining entry references them", async () => { + it("keeps live entries and transcripts under byte pressure at save time", async () => { vi.mocked(getRuntimeConfig).mockReturnValue({ session: { maintenance: { @@ -1697,6 +1706,9 @@ describe("sqlite session normalization", () => { modelOverride: "gpt-5.5", })); + // Live sessions are never save-time budget victims: byte pressure is + // handled by the async physical-budget pass, which only reclaims + // historical generations no entry, route, or admission references. expect( listSqliteSessionEntries({ agentId: "main", @@ -1705,7 +1717,7 @@ describe("sqlite session normalization", () => { }) .map((summary) => summary.sessionKey) .toSorted(), - ).toEqual(["agent:main:active-budget"]); + ).toEqual(["agent:main:active-budget", "agent:main:old-budget", "agent:main:unshared-budget"]); await expect( loadSqliteTranscriptEvents({ agentId: "main", @@ -1725,18 +1737,14 @@ describe("sqlite session normalization", () => { sessionId: "unshared-budget-session", storePath: paths.sqlitePath, }), - ).resolves.toEqual([]); - const archivedOldBudget = fs - .readdirSync(paths.tempDir) - .filter((file) => file.startsWith("old-budget-session.jsonl.deleted.")); - expect(archivedOldBudget).toHaveLength(0); - const archivedUnsharedBudget = fs - .readdirSync(paths.tempDir) - .filter((file) => file.startsWith("unshared-budget-session.jsonl.deleted.")); - expect(archivedUnsharedBudget).toHaveLength(1); + ).resolves.toEqual([ + expect.objectContaining({ + id: "unshared-budget-event", + }), + ]); expect( - readSessionArchiveContentSync(path.join(paths.tempDir, archivedUnsharedBudget[0] ?? "")), - ).toContain("unshared-budget-event"); + fs.readdirSync(paths.tempDir).filter((file) => file.includes(".jsonl.deleted.")), + ).toEqual([]); }); it("resolves confirmed lowercased legacy SQLite session aliases", async () => { diff --git a/src/config/sessions/session-accessor.lifecycle.ts b/src/config/sessions/session-accessor.lifecycle.ts index 3bb5efc1b60c..2a7d79655dd5 100644 --- a/src/config/sessions/session-accessor.lifecycle.ts +++ b/src/config/sessions/session-accessor.lifecycle.ts @@ -22,7 +22,6 @@ import { applySqliteSessionStoreProjection, cleanupSqliteSessionLifecycleArtifacts, deleteSqliteSessionEntryLifecycle, - previewSqliteSessionDiskBudget, purgeSqliteDeletedAgentSessionEntries, rollbackSqliteAgentHarnessSessionEntryLifecycle, rollbackSqlitePluginOwnedSessionEntryLifecycle, @@ -381,13 +380,6 @@ export async function applySessionEntryLifecycleMutation(params: { return await applySqliteSessionEntryLifecycleMutation(params); } -/** Previews SQLite row-byte disk-budget cleanup without mutating persisted rows. */ -export function previewSessionDiskBudget( - params: Parameters[0], -) { - return previewSqliteSessionDiskBudget(params); -} - /** Purges session entries owned by a deleted agent at the storage boundary. */ export async function purgeDeletedAgentSessionEntries( params: DeletedAgentSessionEntryPurgeParams, diff --git a/src/config/sessions/session-accessor.sqlite-archive.ts b/src/config/sessions/session-accessor.sqlite-archive.ts index 57048fd2a76d..3137a4d7aa44 100644 --- a/src/config/sessions/session-accessor.sqlite-archive.ts +++ b/src/config/sessions/session-accessor.sqlite-archive.ts @@ -105,6 +105,13 @@ function writeSqliteTranscriptArchive(params: { writeDurableFileExclusive(tempPath, encoded.bytes); fs.renameSync(tempPath, archivePath); fsyncDirectory(params.archiveDirectory); + // Full readback is bounded by the same single-generation content the + // delete plan already buffers (Node string limits cap both); a partial + // or corrupt archive must fail here, before any rows are reclaimed. + if (readSessionArchiveContentSync(archivePath) !== params.content) { + fs.rmSync(archivePath, { force: true }); + throw new Error(`SQLite transcript archive verification failed for ${params.sessionId}`); + } return archivePath; } catch (error) { fs.rmSync(tempPath, { force: true }); @@ -149,6 +156,10 @@ export function materializeSqliteSessionStateDeletePlans( plans: readonly SqliteSessionStateDeletePlan[], ): MaterializedSqliteSessionStateDeletePlan[] { return dedupeSqliteSessionStateDeletePlans(plans).map((plan) => { + // Empty content means no transcript to preserve (e.g. trajectory-only + // sessions). Writing a verified-but-empty archive would fake extraction; + // trajectory runtime events are diagnostic telemetry and are reclaimed + // without an archive artifact. const archivedTranscript = plan.archiveTranscript && plan.content.length > 0 ? { diff --git a/src/config/sessions/session-accessor.sqlite-disk-budget.ts b/src/config/sessions/session-accessor.sqlite-disk-budget.ts deleted file mode 100644 index aa259c230c5b..000000000000 --- a/src/config/sessions/session-accessor.sqlite-disk-budget.ts +++ /dev/null @@ -1,108 +0,0 @@ -import type { SessionDiskBudgetSweepResult } from "./disk-budget.js"; -import { shouldPreserveMaintenanceEntry } from "./store-maintenance.js"; -import type { SessionEntry } from "./types.js"; - -export type SqliteSessionRowBytes = { - entryBytesByKey: Map; - trajectoryBytesBySessionId: Map; - transcriptBytesBySessionId: Map; -}; - -function getSessionStateBytes(rowBytes: SqliteSessionRowBytes, sessionId: string): number { - return ( - (rowBytes.transcriptBytesBySessionId.get(sessionId) ?? 0) + - (rowBytes.trajectoryBytesBySessionId.get(sessionId) ?? 0) - ); -} - -function getEntryUpdatedAt(entry?: SessionEntry): number { - return entry?.updatedAt ?? Number.NEGATIVE_INFINITY; -} - -export function enforceSqliteSessionDiskBudget(params: { - collectStateIds: (entry: SessionEntry) => readonly string[]; - maintenance: { maxDiskBytes: number | null; highWaterBytes: number | null }; - onRemoveEntry?: (removed: { key: string; entry: SessionEntry }) => void; - preserveKeys?: ReadonlySet; - rowBytes: SqliteSessionRowBytes; - store: Record; -}): SessionDiskBudgetSweepResult | null { - const { maxDiskBytes, highWaterBytes } = params.maintenance; - if (maxDiskBytes == null || highWaterBytes == null) { - return null; - } - - let totalBytes = 0; - const entryBytesByKey = new Map(); - const sessionIdsByKey = new Map(); - const sessionIdRefCounts = new Map(); - // State rows can be shared through usage-family references. Count each - // session once; release its bytes only after the last entry is removed. - for (const [key, entry] of Object.entries(params.store)) { - const entryBytes = params.rowBytes.entryBytesByKey.get(key) ?? 0; - const sessionIds = params.collectStateIds(entry); - entryBytesByKey.set(key, entryBytes); - sessionIdsByKey.set(key, sessionIds); - totalBytes += entryBytes; - for (const sessionId of sessionIds) { - sessionIdRefCounts.set(sessionId, (sessionIdRefCounts.get(sessionId) ?? 0) + 1); - } - } - for (const sessionId of sessionIdRefCounts.keys()) { - totalBytes += getSessionStateBytes(params.rowBytes, sessionId); - } - - const totalBytesBefore = totalBytes; - if (totalBytes <= maxDiskBytes) { - return { - totalBytesBefore, - totalBytesAfter: totalBytes, - removedFiles: 0, - removedEntries: 0, - freedBytes: 0, - maxBytes: maxDiskBytes, - highWaterBytes, - overBudget: false, - }; - } - - let removedEntries = 0; - const keys = Object.keys(params.store).toSorted( - (left, right) => getEntryUpdatedAt(params.store[left]) - getEntryUpdatedAt(params.store[right]), - ); - for (const key of keys) { - if (totalBytes <= highWaterBytes) { - break; - } - const entry = params.store[key]; - if ( - !entry || - shouldPreserveMaintenanceEntry({ key, entry, preserveKeys: params.preserveKeys }) - ) { - continue; - } - params.onRemoveEntry?.({ key, entry }); - delete params.store[key]; - removedEntries += 1; - totalBytes -= entryBytesByKey.get(key) ?? 0; - for (const sessionId of sessionIdsByKey.get(key) ?? []) { - const nextRefCount = (sessionIdRefCounts.get(sessionId) ?? 0) - 1; - if (nextRefCount > 0) { - sessionIdRefCounts.set(sessionId, nextRefCount); - continue; - } - sessionIdRefCounts.delete(sessionId); - totalBytes -= getSessionStateBytes(params.rowBytes, sessionId); - } - } - return { - totalBytesBefore, - totalBytesAfter: totalBytes, - removedFiles: 0, - removedEntries, - freedBytes: Math.max(0, totalBytesBefore - totalBytes), - maxBytes: maxDiskBytes, - highWaterBytes, - overBudget: true, - }; -} diff --git a/src/config/sessions/session-accessor.sqlite-entry.ts b/src/config/sessions/session-accessor.sqlite-entry.ts index d0c8edfca577..00de71a2f1e7 100644 --- a/src/config/sessions/session-accessor.sqlite-entry.ts +++ b/src/config/sessions/session-accessor.sqlite-entry.ts @@ -63,6 +63,8 @@ import { readSqliteSessionEntriesByStatus, } from "./session-accessor.sqlite-status.js"; import { preserveSqliteSameKeySessionRolloverLineage } from "./session-entry-lineage.js"; +import { kickSessionHistoryDiskBudgetMaintenance } from "./session-history-eviction.js"; +import { resolveSessionStorePathForScope } from "./session-store-path.js"; import type { GroupKeyResolution, SessionEntry } from "./types.js"; import { mergeSessionEntry, mergeSessionEntryPreserveActivity } from "./types.js"; @@ -300,6 +302,11 @@ export async function patchSqliteSessionEntry( }, toDatabaseOptions(resolved)); emitCommittedSessionIdentityDiff(previousIdentity, currentIdentity); finalizeSqliteSessionEntryMaintenancePlansBestEffort(resolved, maintenancePlans); + kickSessionHistoryDiskBudgetMaintenance({ + ...(resolved.agentId ? { agentId: resolved.agentId } : {}), + storePath: resolveSessionStorePathForScope(scope), + ...(options.maintenanceConfig ? { maintenanceConfig: options.maintenanceConfig } : {}), + }); return result; }); } @@ -370,6 +377,11 @@ export async function patchSqliteSessionEntryTarget( }, toDatabaseOptions(resolved)); emitCommittedSessionIdentityDiff(previousIdentity, currentIdentity); finalizeSqliteSessionEntryMaintenancePlansBestEffort(resolved, maintenancePlans); + kickSessionHistoryDiskBudgetMaintenance({ + ...(resolved.agentId ? { agentId: resolved.agentId } : {}), + storePath: resolveSessionStorePathForScope(scope), + ...(options.maintenanceConfig ? { maintenanceConfig: options.maintenanceConfig } : {}), + }); return result; }); } diff --git a/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts b/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts index f90f7cf16922..63656396e33f 100644 --- a/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts +++ b/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts @@ -116,7 +116,8 @@ function sqliteTranscriptStateHasMarker(params: { return rows.some((row) => row.event_json.includes(params.transcriptContentMarker)); } -function readReferencedSqliteSessionIds(database: OpenClawAgentDatabase): Set { +/** Session ids protected by live entry state or durable route targets. */ +export function readReferencedSqliteSessionIds(database: OpenClawAgentDatabase): Set { const db = getSessionKysely(database.db); const rows = executeSqliteQuerySync( database.db, @@ -133,6 +134,13 @@ function readReferencedSqliteSessionIds(database: OpenClawAgentDatabase): Set, ): SessionLifecycleArchivedTranscript[] { const archivedTranscripts: SessionLifecycleArchivedTranscript[] = []; const referencedSessionIds = readReferencedSqliteSessionIds(database); + for (const sessionId of protectedSessionIds ?? []) { + referencedSessionIds.add(sessionId); + } for (const plan of plans) { if (referencedSessionIds.has(plan.sessionId)) { continue; @@ -266,6 +287,22 @@ export function planSqliteSessionStateAfterEntryRemoval(params: { return plans; } +/** Ids of every persisted generation owned by the given logical session keys. */ +export function readSqliteSessionGenerationIdsForKeys( + database: OpenClawAgentDatabase, + keys: Iterable, +): string[] { + const sessionKeys = uniqueStrings([...keys].map((key) => key.trim())); + if (sessionKeys.length === 0) { + return []; + } + const db = getSessionKysely(database.db); + return executeSqliteQuerySync( + database.db, + db.selectFrom("sessions").select("session_id").where("session_key", "in", sessionKeys), + ).rows.map((row) => row.session_id); +} + // Projects removals and upserts before archive materialization so same-call // upserts can keep a transcript live without producing a spurious archive. export async function projectSqliteSessionEntryLifecycleMutation( @@ -382,6 +419,18 @@ export function collectProjectedReferencedSqliteSessionIds(params: { for (const sessionId of collectReferencedSqliteSessionIdsFromStore(params.projectedStore)) { sessionIds.add(sessionId); } + // Routes protect their target session unless the cleanup removes that key's + // route in the same pass; mirroring the post-cleanup state here keeps the + // plan from writing archives for sessions the delete stage will retain. + const routeRows = executeSqliteQuerySync( + params.database.db, + db.selectFrom("session_routes").select(["session_id", "session_key"]), + ).rows; + for (const row of routeRows) { + if (!excludedSessionKeys.has(row.session_key)) { + sessionIds.add(row.session_id); + } + } return sessionIds; } diff --git a/src/config/sessions/session-accessor.sqlite-lifecycle.ts b/src/config/sessions/session-accessor.sqlite-lifecycle.ts index 4f9710d9f64c..06a1fb62e7bb 100644 --- a/src/config/sessions/session-accessor.sqlite-lifecycle.ts +++ b/src/config/sessions/session-accessor.sqlite-lifecycle.ts @@ -37,6 +37,8 @@ import { deleteMaterializedSqliteSessionStatePlans, deletePlannedSqliteLifecycleArtifactEntries, planSqliteSessionLifecycleArtifactCleanup, + planSqliteSessionStateDeleteIfUnreferenced, + readSqliteSessionGenerationIdsForKeys, planSqliteSessionStateAfterEntryRemoval, readReferencedSqliteSessionIdsAfterTargetMutation, } from "./session-accessor.sqlite-lifecycle-state.js"; @@ -48,6 +50,10 @@ import { runExclusiveSqliteSessionWrite, toDatabaseOptions, } from "./session-accessor.sqlite-scope.js"; +import { + collectAdmissionProtectedSessionIds, + kickSessionHistoryDiskBudgetMaintenance, +} from "./session-history-eviction.js"; import type { ResetSessionEntryLifecycleMutation } from "./store.js"; import type { SessionEntry } from "./types.js"; @@ -140,72 +146,68 @@ export async function resetSqliteSessionEntryLifecycle( params: ResetSessionEntryLifecycleParams, ): Promise { const resolved = resolveSqliteStoreScope(params.storePath, { agentId: params.agentId }); - return await runExclusiveSqliteSessionWrite(resolved, async () => { - const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); - const targetSnapshot = readSqliteLifecycleTargetSnapshot(database, params.target); - const current = targetSnapshot.primary; - const nextEntry = await params.buildNextEntry({ - currentEntry: current ? cloneSessionEntry(current.entry) : undefined, - primaryKey: params.target.canonicalKey, + // Retained reset history is the store's growth event; give the throttled + // budget pass a chance to extract-and-evict once we finish. + try { + return await runExclusiveSqliteSessionWrite(resolved, async () => { + const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); + const targetSnapshot = readSqliteLifecycleTargetSnapshot(database, params.target); + const current = targetSnapshot.primary; + const nextEntry = await params.buildNextEntry({ + currentEntry: current ? cloneSessionEntry(current.entry) : undefined, + primaryKey: params.target.canonicalKey, + }); + const mutation: ResetSessionEntryLifecycleMutation = { + nextEntry: cloneSessionEntry(nextEntry), + ...(current ? { previousEntry: cloneSessionEntry(current.entry) } : {}), + ...(current?.entry.sessionFile ? { previousSessionFile: current.entry.sessionFile } : {}), + ...(current?.entry.sessionId ? { previousSessionId: current.entry.sessionId } : {}), + }; + runOpenClawAgentWriteTransaction((transactionDb) => { + assertSqliteLifecycleTargetUnchanged(transactionDb, params.target, current?.entry, "reset"); + deleteSqliteLifecycleTargetRows(transactionDb, params.target); + writeSessionEntry(transactionDb, params.target.canonicalKey, nextEntry); + // Reset only advances the live entry and route. Historical rows stay searchable; + // disk-budget cleanup owns durable extraction before reclaiming them. + }, toDatabaseOptions(resolved)); + if (current) { + emitSessionIdentityMutation({ + kind: "reset", + previous: { + ...(current.entry.sessionId ? { sessionId: current.entry.sessionId } : {}), + sessionKeys: targetSnapshot.rows.map((row) => row.sessionKey), + }, + current: { + ...(nextEntry.sessionId ? { sessionId: nextEntry.sessionId } : {}), + sessionKeys: [params.target.canonicalKey], + }, + }); + } else { + emitSessionIdentityMutation({ + kind: "create", + previous: { sessionKeys: [] }, + current: { + ...(nextEntry.sessionId ? { sessionId: nextEntry.sessionId } : {}), + sessionKeys: [params.target.canonicalKey], + }, + }); + } + await params.afterEntryMutation?.(mutation); + return { + ...mutation, + archivedTranscripts: [], + }; }); - const mutation: ResetSessionEntryLifecycleMutation = { - nextEntry: cloneSessionEntry(nextEntry), - ...(current ? { previousEntry: cloneSessionEntry(current.entry) } : {}), - ...(current?.entry.sessionFile ? { previousSessionFile: current.entry.sessionFile } : {}), - ...(current?.entry.sessionId ? { previousSessionId: current.entry.sessionId } : {}), - }; - let archivedTranscripts: SessionLifecycleArchivedTranscript[] = []; - const referencedAfterReset = current?.entry.sessionId - ? readReferencedSqliteSessionIdsAfterTargetMutation(database, params.target, nextEntry) - : new Set(); - const deletePlans = current?.entry.sessionId - ? planSqliteSessionStateAfterEntryRemoval({ - archiveDirectory: resolveSqliteTranscriptArchiveDirectory(resolved), - database, - entry: current.entry, - reason: "reset", - referencedSessionIds: referencedAfterReset, - }) - : []; - const materializedPlans = materializeSqliteSessionStateDeletePlans(deletePlans); - runOpenClawAgentWriteTransaction((transactionDb) => { - assertSqliteLifecycleTargetUnchanged(transactionDb, params.target, current?.entry, "reset"); - deleteSqliteLifecycleTargetRows(transactionDb, params.target); - writeSessionEntry(transactionDb, params.target.canonicalKey, nextEntry); - archivedTranscripts = deleteMaterializedSqliteSessionStatePlans( - transactionDb, - materializedPlans, - ); - }, toDatabaseOptions(resolved)); - if (current) { - emitSessionIdentityMutation({ - kind: "reset", - previous: { - ...(current.entry.sessionId ? { sessionId: current.entry.sessionId } : {}), - sessionKeys: targetSnapshot.rows.map((row) => row.sessionKey), - }, - current: { - ...(nextEntry.sessionId ? { sessionId: nextEntry.sessionId } : {}), - sessionKeys: [params.target.canonicalKey], - }, - }); - } else { - emitSessionIdentityMutation({ - kind: "create", - previous: { sessionKeys: [] }, - current: { - ...(nextEntry.sessionId ? { sessionId: nextEntry.sessionId } : {}), - sessionKeys: [params.target.canonicalKey], - }, - }); - } - await params.afterEntryMutation?.(mutation); - emitArchivedSqliteTranscriptUpdates(archivedTranscripts); - return { - ...mutation, - archivedTranscripts, - }; - }); + } finally { + // Reset is what turns the old generation into an eviction candidate; a + // throttled kick could be suppressed by a recent pre-reset pass and never + // retried if the agent idles, leaving an over-budget store unreclaimed. + kickSessionHistoryDiskBudgetMaintenance({ + ...(resolved.agentId ? { agentId: resolved.agentId } : {}), + storePath: params.storePath, + force: true, + }); + } } async function deleteSqliteSessionEntryLifecycleInternal( @@ -214,6 +216,31 @@ async function deleteSqliteSessionEntryLifecycleInternal( expectedPluginOwnerId?: string, ): Promise { const resolved = resolveSqliteStoreScope(params.storePath, { agentId: params.agentId }); + try { + return await deleteSqliteSessionEntryLifecycleLocked( + resolved, + params, + allowLockedEntryRemoval, + expectedPluginOwnerId, + ); + } finally { + // Deletion writes an archive per retained generation before reclaiming + // rows, so usage can spike well past the budget; force a pass instead of + // waiting up to the throttle interval (or forever, if the agent idles). + kickSessionHistoryDiskBudgetMaintenance({ + ...(params.agentId ? { agentId: params.agentId } : {}), + storePath: params.storePath, + force: true, + }); + } +} + +async function deleteSqliteSessionEntryLifecycleLocked( + resolved: ReturnType, + params: DeleteSessionEntryLifecycleParams, + allowLockedEntryRemoval: boolean, + expectedPluginOwnerId?: string, +): Promise { return await runExclusiveSqliteSessionWrite(resolved, async () => { let result: DeleteSessionEntryLifecycleResult = { archivedTranscripts: [], @@ -246,10 +273,11 @@ async function deleteSqliteSessionEntryLifecycleInternal( ); // SQLite transcript state is keyed by session id; sessionFile is only its // marker. Materialization dedupes aliases that share the same state owner. - const deletePlans = params.archiveTranscript + const archiveDirectory = resolveSqliteTranscriptArchiveDirectory(resolved); + const entryPlans = params.archiveTranscript ? targetSnapshot.rows.flatMap(({ entry }) => planSqliteSessionStateAfterEntryRemoval({ - archiveDirectory: resolveSqliteTranscriptArchiveDirectory(resolved), + archiveDirectory, archiveTranscript: true, database, entry, @@ -258,7 +286,93 @@ async function deleteSqliteSessionEntryLifecycleInternal( }), ) : []; - const materializedPlans = materializeSqliteSessionStateDeletePlans(deletePlans); + const entryPlanIds = new Set(entryPlans.map((plan) => plan.sessionId)); + // Ids only — planning (which loads full transcript content) happens + // lazily one generation at a time after the main transaction. + const historicalGenerationIds = params.archiveTranscript + ? readSqliteSessionGenerationIdsForKeys(database, [ + params.target.canonicalKey, + ...params.target.storeKeys, + ...targetSnapshot.rows.map((row) => row.sessionKey), + ]).filter((sessionId) => !entryPlanIds.has(sessionId)) + : []; + // Historical generations are reclaimed BEFORE the entry-removing + // transaction, one generation per transaction: an archive or delete + // failure aborts the whole deletion while the live entry still exists, + // so a retry rediscovers the remaining history. Acknowledging deletion + // first would let surviving generations become unreachable via delete. + // Preflight the admission fence over every generation BEFORE deleting + // anything, so an in-flight run rejects the whole deletion instead of + // aborting it midway through committed removals. + const preflightFence = collectAdmissionProtectedSessionIds({ + database, + storePath: params.storePath, + }); + for (const sessionId of historicalGenerationIds) { + if (preflightFence.has(sessionId) && !referencedAfterDelete.has(sessionId)) { + throw new Error( + `cannot delete session history while work is in flight for ${sessionId}; retry after the run completes`, + ); + } + } + const historicalArchivedTranscripts: SessionLifecycleArchivedTranscript[] = []; + for (const sessionId of historicalGenerationIds) { + if (referencedAfterDelete.has(sessionId)) { + // Another live entry or route still owns this generation; deleting + // this entry must not destroy shared history. + continue; + } + // Recomputed per generation so a run admitted after preflight cannot + // lose the generation it just adopted; such a race aborts the deletion, + // and since the live entry is still present a retry finishes the rest. + const admissionProtected = collectAdmissionProtectedSessionIds({ + database, + storePath: params.storePath, + }); + if (admissionProtected.has(sessionId)) { + throw new Error( + `cannot delete session history while work is in flight for ${sessionId}; retry after the run completes`, + ); + } + const plan = planSqliteSessionStateDeleteIfUnreferenced({ + archiveDirectory, + archiveTranscript: true, + database, + reason: "deleted", + referencedSessionIds: referencedAfterDelete, + sessionId, + }); + if (!plan) { + continue; + } + const materializedGeneration = materializeSqliteSessionStateDeletePlans([plan]); + const archivedGeneration: SessionLifecycleArchivedTranscript[] = []; + runOpenClawAgentWriteTransaction((transactionDb) => { + // Authoritative fence: admissions are process-local sync state and this + // callback runs synchronously, so a run admitted after the pre-checks + // cannot interleave past this point. An outer lifecycle-mutation hold + // (as eviction uses) would invert lock order against the exclusive + // SQLite write this function already holds. + const fenceAtDelete = collectAdmissionProtectedSessionIds({ + database: transactionDb, + storePath: params.storePath, + }); + if (fenceAtDelete.has(sessionId)) { + throw new Error( + `cannot delete session history while work is in flight for ${sessionId}; retry after the run completes`, + ); + } + archivedGeneration.push( + ...deleteMaterializedSqliteSessionStatePlans(transactionDb, materializedGeneration), + ); + }, toDatabaseOptions(resolved)); + // Publish each committed generation immediately: a later archive or + // transaction failure aborts the deletion, and observers must still see + // the removals that already happened (retry completes the remainder). + emitArchivedSqliteTranscriptUpdates(archivedGeneration); + historicalArchivedTranscripts.push(...archivedGeneration); + } + const materializedPlans = materializeSqliteSessionStateDeletePlans(entryPlans); runOpenClawAgentWriteTransaction((transactionDb) => { const transactionSnapshot = readSqliteLifecycleTargetSnapshot(transactionDb, params.target); assertSqliteLifecycleTargetSnapshotUnchanged( @@ -298,6 +412,9 @@ async function deleteSqliteSessionEntryLifecycleInternal( }); } emitArchivedSqliteTranscriptUpdates(result.archivedTranscripts); + // Historical generations were emitted per commit above; merge them into + // the result after the final emit so callers still see every archive. + result.archivedTranscripts.push(...historicalArchivedTranscripts); return result; }); } diff --git a/src/config/sessions/session-accessor.sqlite-maintenance.ts b/src/config/sessions/session-accessor.sqlite-maintenance.ts index aa90bb7d5191..fc7f415e6888 100644 --- a/src/config/sessions/session-accessor.sqlite-maintenance.ts +++ b/src/config/sessions/session-accessor.sqlite-maintenance.ts @@ -1,25 +1,15 @@ import { uniqueStrings } from "@openclaw/normalization-core/string-normalization"; -import { sql } from "kysely"; -import { - executeSqliteQuerySync, - executeSqliteQueryTakeFirstSync, -} from "../../infra/kysely-sync.js"; +import { executeSqliteQuerySync } from "../../infra/kysely-sync.js"; import { getChildLogger } from "../../logging/logger.js"; import { - openOpenClawAgentDatabase, runOpenClawAgentWriteTransaction, type OpenClawAgentDatabase, } from "../../state/openclaw-agent-db.js"; -import type { SessionDiskBudgetSweepResult } from "./disk-budget.js"; import { materializeSqliteSessionStateDeletePlans, type SqliteSessionStateDeletePlan, } from "./session-accessor.sqlite-archive.js"; import type { SessionLifecycleArchivedTranscript } from "./session-accessor.sqlite-contract.js"; -import { - enforceSqliteSessionDiskBudget, - type SqliteSessionRowBytes, -} from "./session-accessor.sqlite-disk-budget.js"; import { readSqliteSessionEntryCount } from "./session-accessor.sqlite-entry-store.js"; import { emitCommittedSessionEntryRemovals } from "./session-accessor.sqlite-identity.js"; import { @@ -30,11 +20,9 @@ import { planSqliteSessionStateDeleteIfUnreferenced, } from "./session-accessor.sqlite-lifecycle-state.js"; import type { SqliteSessionEntryMaintenancePlan } from "./session-accessor.sqlite-lifecycle-types.js"; -import { normalizeSqliteNumber } from "./session-accessor.sqlite-normalize.js"; import { cloneSessionEntry, getSessionKysely, - resolveSqliteScope, toDatabaseOptions, type ResolvedSqliteReadScope, } from "./session-accessor.sqlite-scope.js"; @@ -52,7 +40,7 @@ import { } from "./store-maintenance.js"; import type { SessionEntry } from "./types.js"; -// Entry pruning and disk-budget owner. Produces plans inside writes; finalizes archives afterward. +// Live-entry pruning owner. Produces plans inside writes; finalizes archives afterward. function collectSqliteSessionMaintenanceBaseKeys( store: Record, @@ -69,158 +57,6 @@ function collectSqliteSessionMaintenanceBaseKeys( return keys; } -function sumEventJsonBytes() { - return ( - // kysely-allow-raw: SQLite byte accounting needs LENGTH(CAST(... AS BLOB)), - // which Kysely does not expose as a typed aggregate helper. - sql`COALESCE(SUM(length(CAST(event_json AS BLOB))), 0)`.as("event_json_bytes") - ); -} - -function sumSessionEntryJsonBytes() { - return ( - // kysely-allow-raw: SQLite byte accounting needs LENGTH(CAST(... AS BLOB)), - // which Kysely does not expose as a typed aggregate helper. - sql`COALESCE(SUM(length(CAST(entry_json AS BLOB))), 0)`.as("entry_json_bytes") - ); -} - -function readSqliteSessionRowBytes(database: OpenClawAgentDatabase): SqliteSessionRowBytes { - const db = getSessionKysely(database.db); - const entryRows = executeSqliteQuerySync( - database.db, - db.selectFrom("session_entries").select(["session_key", "entry_json"]), - ).rows; - const transcriptRows = executeSqliteQuerySync( - database.db, - db - .selectFrom("transcript_events") - .select(["session_id"]) - .select(sumEventJsonBytes()) - .groupBy("session_id"), - ).rows; - const trajectoryRows = executeSqliteQuerySync( - database.db, - db - .selectFrom("trajectory_runtime_events") - .select(["session_id"]) - .select(sumEventJsonBytes()) - .groupBy("session_id"), - ).rows; - const entryBytesByKey = new Map(); - for (const row of entryRows) { - entryBytesByKey.set(row.session_key, Buffer.byteLength(row.entry_json, "utf8")); - } - const transcriptBytesBySessionId = new Map(); - for (const row of transcriptRows) { - const bytes = row.event_json_bytes; - transcriptBytesBySessionId.set(row.session_id, normalizeSqliteNumber(bytes ?? 0)); - } - const trajectoryBytesBySessionId = new Map(); - for (const row of trajectoryRows) { - const bytes = row.event_json_bytes; - trajectoryBytesBySessionId.set(row.session_id, normalizeSqliteNumber(bytes ?? 0)); - } - return { entryBytesByKey, trajectoryBytesBySessionId, transcriptBytesBySessionId }; -} - -function hasSqliteSessionDiskBudgetOverflow( - database: OpenClawAgentDatabase, - maintenance: ResolvedSessionMaintenanceConfig, -): boolean { - if (maintenance.maxDiskBytes == null || maintenance.highWaterBytes == null) { - return false; - } - const db = getSessionKysely(database.db); - const entryRow = executeSqliteQueryTakeFirstSync( - database.db, - db.selectFrom("session_entries").select(sumSessionEntryJsonBytes()), - ); - const transcriptRow = executeSqliteQueryTakeFirstSync( - database.db, - db.selectFrom("transcript_events").select(sumEventJsonBytes()), - ); - const trajectoryRow = executeSqliteQueryTakeFirstSync( - database.db, - db.selectFrom("trajectory_runtime_events").select(sumEventJsonBytes()), - ); - const entryBytes = normalizeSqliteNumber(entryRow?.entry_json_bytes ?? 0); - const transcriptBytes = normalizeSqliteNumber(transcriptRow?.event_json_bytes ?? 0); - const trajectoryBytes = normalizeSqliteNumber(trajectoryRow?.event_json_bytes ?? 0); - return entryBytes + transcriptBytes + trajectoryBytes > maintenance.maxDiskBytes; -} - -function applySqliteSessionDiskBudget(params: { - database: OpenClawAgentDatabase; - store: Record; - maintenance: ResolvedSessionMaintenanceConfig; - preserveKeys: ReadonlySet; - rememberRemovedEntry: (removed: { key: string; entry: SessionEntry }) => void; -}): void { - enforceSqliteSessionDiskBudgetInStore({ - database: params.database, - store: params.store, - maintenance: params.maintenance, - preserveKeys: params.preserveKeys, - onRemoveEntry: params.rememberRemovedEntry, - }); -} - -function enforceSqliteSessionDiskBudgetInStore(params: { - database: OpenClawAgentDatabase; - store: Record; - maintenance: Pick; - preserveKeys?: ReadonlySet; - onRemoveEntry?: (removed: { key: string; entry: SessionEntry }) => void; -}): SessionDiskBudgetSweepResult | null { - return enforceSqliteSessionDiskBudget({ - collectStateIds: collectSqliteSessionStateIdsForEntry, - maintenance: params.maintenance, - onRemoveEntry: params.onRemoveEntry, - preserveKeys: params.preserveKeys, - rowBytes: readSqliteSessionRowBytes(params.database), - store: params.store, - }); -} - -export function previewSqliteSessionDiskBudget(params: { - agentId?: string; - activeSessionKey?: string; - store: Record; - storePath: string; - maintenance: Pick; - preserveKeys?: ReadonlySet; -}): { diskBudget: SessionDiskBudgetSweepResult | null; removedKeys: Set } { - const removedKeys = new Set(); - if (params.maintenance.maxDiskBytes == null || params.maintenance.highWaterBytes == null) { - return { diskBudget: null, removedKeys }; - } - const resolved = resolveSqliteScope({ - ...(params.agentId ? { agentId: params.agentId } : {}), - sessionKey: "", - storePath: params.storePath, - }); - const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); - const baseKeys = collectSqliteSessionMaintenanceBaseKeys( - params.store, - params.activeSessionKey ?? "", - ); - const preserveKeys = - baseKeys.length > 0 || params.preserveKeys - ? new Set([...(params.preserveKeys ?? []), ...baseKeys]) - : undefined; - const diskBudget = enforceSqliteSessionDiskBudgetInStore({ - database, - store: params.store, - maintenance: params.maintenance, - preserveKeys, - onRemoveEntry: ({ key }) => { - removedKeys.add(key); - }, - }); - return { diskBudget, removedKeys }; -} - function hasStaleSqliteSessionEntryCandidate( database: OpenClawAgentDatabase, pruneAfterMs: number, @@ -264,12 +100,10 @@ export function applySqliteSessionEntryMaintenance( maintenance.pruneAfterMs, preserveCandidateKeys, ); - const hasDiskBudgetOverflow = hasSqliteSessionDiskBudgetOverflow(database, maintenance); const shouldLoadStore = params.forceMaintenance === true || entryCount > maintenance.maxEntries || hasStaleCandidate || - hasDiskBudgetOverflow || shouldRunModelRunPrune({ maintenance, entryCount, @@ -348,14 +182,6 @@ export function applySqliteSessionEntryMaintenance( preserveKeys, }); } - applySqliteSessionDiskBudget({ - database, - store, - maintenance, - preserveKeys, - rememberRemovedEntry, - }); - const referencedSessionIds = collectProjectedReferencedSqliteSessionIds({ database, excludedSessionKeys: removedKeys, diff --git a/src/config/sessions/session-accessor.sqlite.ts b/src/config/sessions/session-accessor.sqlite.ts index 362f385904f3..f0697f17b471 100644 --- a/src/config/sessions/session-accessor.sqlite.ts +++ b/src/config/sessions/session-accessor.sqlite.ts @@ -58,7 +58,6 @@ export { } from "./session-accessor.sqlite-transcript-write.js"; export { publishSqliteTranscriptUpdate } from "./session-accessor.sqlite-events.js"; export { readSqliteTranscriptRawDelta } from "./session-accessor.sqlite-delta.js"; -export { previewSqliteSessionDiskBudget } from "./session-accessor.sqlite-maintenance.js"; export { findSqliteTranscriptEvent, loadLatestSqliteAssistantText, diff --git a/src/config/sessions/session-accessor.ts b/src/config/sessions/session-accessor.ts index f35c93335bc8..e00e0cda2180 100644 --- a/src/config/sessions/session-accessor.ts +++ b/src/config/sessions/session-accessor.ts @@ -159,7 +159,6 @@ export { cleanupSessionLifecycleArtifacts, deleteSessionEntryLifecycle, preserveTemporarySessionMapping, - previewSessionDiskBudget, purgeDeletedAgentSessionEntries, resetSessionEntryLifecycle, restoreSessionFromCompactionCheckpoint, diff --git a/src/config/sessions/session-history-eviction.test.ts b/src/config/sessions/session-history-eviction.test.ts new file mode 100644 index 000000000000..0b17bf974508 --- /dev/null +++ b/src/config/sessions/session-history-eviction.test.ts @@ -0,0 +1,331 @@ +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { executeSqliteQuerySync } from "../../infra/kysely-sync.js"; +import { beginSessionWorkAdmission } from "../../sessions/session-lifecycle-admission.js"; +import { + closeOpenClawAgentDatabasesForTest, + openOpenClawAgentDatabase, +} from "../../state/openclaw-agent-db.js"; +import { appendSqliteTrajectoryRuntimeEvents } from "../../trajectory/runtime-store.sqlite.js"; +import type { TrajectoryEvent } from "../../trajectory/types.js"; +import { measureSessionPhysicalDiskUsage } from "./disk-budget.js"; +import { + appendTranscriptMessage, + replaceSessionEntry, + resetSessionEntryLifecycle, +} from "./session-accessor.js"; +import { getSessionKysely } from "./session-accessor.sqlite-scope.js"; +import { + enforceSqliteSessionHistoryDiskBudget, + kickSessionHistoryDiskBudgetMaintenance, + inspectSqliteSessionHistoryDiskBudget, +} from "./session-history-eviction.js"; +import { resolveSqliteTargetFromSessionStorePath } from "./session-sqlite-target.js"; + +describe("SQLite historical session disk budget", () => { + let tempDir: string; + let storePath: string; + + beforeEach(() => { + tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-session-history-budget-")); + storePath = path.join(tempDir, "sessions.json"); + }); + + afterEach(() => { + closeOpenClawAgentDatabasesForTest(); + fs.rmSync(tempDir, { recursive: true, force: true }); + }); + + it("evicts the oldest historical session and stops after reaching high water", async () => { + const sessionKey = "agent:main:history-order"; + await createHistoricalTranscript({ + content: "oldest " + "x".repeat(64 * 1024), + nextSessionId: "newer-history", + sessionId: "oldest-history", + sessionKey, + updatedAt: 10, + }); + await appendTranscriptMessage( + { sessionId: "newer-history", sessionKey, storePath }, + { message: { role: "user", content: "newer " + "y".repeat(64 * 1024) } }, + ); + await resetSessionEntryLifecycle({ + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + buildNextEntry: () => ({ sessionId: "live-history", updatedAt: 30 }), + }); + setSessionUpdatedAt("newer-history", 20); + settlePhysicalUsage(); + const before = await measureSessionPhysicalDiskUsage(storePath); + + const result = await enforceSqliteSessionHistoryDiskBudget({ + storePath, + mode: "enforce", + maintenance: { + maxDiskBytes: before.totalBytes - 1, + highWaterBytes: before.totalBytes - 1, + }, + }); + + expect(result?.removedEntries).toBe(1); + expect(result?.totalBytesAfter).toBeLessThanOrEqual(before.totalBytes - 1); + expect(sessionExists("oldest-history")).toBe(false); + expect(sessionExists("newer-history")).toBe(true); + expect(sessionExists("live-history")).toBe(true); + expect(readArchiveNames("oldest-history")).toHaveLength(1); + expect(readArchiveNames("newer-history")).toHaveLength(0); + }); + + it("removes counted archives before evicting searchable history", async () => { + await createHistoricalTranscript({ + content: "keep searchable history", + nextSessionId: "archive-live", + sessionId: "archive-history", + sessionKey: "agent:main:archive-pressure", + updatedAt: 1, + }); + database().walMaintenance.checkpoint(); + const oldArchive = path.join( + tempDir, + "already-extracted.jsonl.deleted.2026-01-01T00-00-00.000Z", + ); + fs.writeFileSync(oldArchive, Buffer.alloc(256 * 1024)); + const before = await measureSessionPhysicalDiskUsage(storePath); + + const result = await enforceSqliteSessionHistoryDiskBudget({ + storePath, + mode: "enforce", + maintenance: { + maxDiskBytes: before.totalBytes - 1, + highWaterBytes: before.totalBytes - 64 * 1024, + }, + }); + + expect(result).toMatchObject({ removedEntries: 0, removedFiles: 1 }); + expect(fs.existsSync(oldArchive)).toBe(false); + expect(sessionExists("archive-history")).toBe(true); + }); + + it("excludes entry, route, and admitted ids while evicting trajectory-only history", async () => { + const sessionKey = "agent:main:history-protection"; + await replaceSessionEntry( + { sessionKey, storePath }, + { sessionId: "admitted-history", updatedAt: 1 }, + ); + await appendTranscriptMessage( + { sessionId: "admitted-history", sessionKey, storePath }, + { message: { role: "user", content: "admitted" } }, + ); + await resetSessionEntryLifecycle({ + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + buildNextEntry: () => ({ sessionId: "route-history", updatedAt: 2 }), + }); + await appendTranscriptMessage( + { sessionId: "route-history", sessionKey, storePath }, + { message: { role: "user", content: "route protected" } }, + ); + await resetSessionEntryLifecycle({ + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + buildNextEntry: () => ({ sessionId: "trajectory-history", updatedAt: 3 }), + }); + appendSqliteTrajectoryRuntimeEvents({ sessionId: "trajectory-history", storePath }, [ + createTrajectoryEvent("trajectory-history", sessionKey), + ]); + await resetSessionEntryLifecycle({ + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + buildNextEntry: () => ({ sessionId: "live-history", updatedAt: 4 }), + }); + addRouteReference("route-only", "route-history"); + const admission = await beginSessionWorkAdmission({ + scope: storePath, + identities: ["admitted-history"], + assertAllowed: () => {}, + }); + try { + const before = await measureSessionPhysicalDiskUsage(storePath); + const result = await enforceSqliteSessionHistoryDiskBudget({ + storePath, + mode: "enforce", + maintenance: { maxDiskBytes: before.totalBytes - 1, highWaterBytes: 0 }, + }); + + expect(result?.removedEntries).toBe(1); + expect(sessionExists("trajectory-history")).toBe(false); + // Trajectory-only sessions carry no transcript; eviction reclaims their + // diagnostic telemetry without writing an empty archive artifact. + expect(readArchiveNames("trajectory-history")).toHaveLength(0); + expect(sessionExists("admitted-history")).toBe(true); + expect(sessionExists("route-history")).toBe(true); + expect(sessionExists("live-history")).toBe(true); + } finally { + admission.release(); + } + }); + + it("warn mode reports physical overage without extracting or deleting history", async () => { + await createHistoricalTranscript({ + content: "warn history", + nextSessionId: "warn-live", + sessionId: "warn-old", + sessionKey: "agent:main:warn-history", + updatedAt: 1, + }); + const before = await measureSessionPhysicalDiskUsage(storePath); + + const inspected = await inspectSqliteSessionHistoryDiskBudget({ + storePath, + mode: "warn", + maintenance: { maxDiskBytes: before.totalBytes - 1, highWaterBytes: 0 }, + }); + const result = await enforceSqliteSessionHistoryDiskBudget({ + storePath, + mode: "warn", + maintenance: { maxDiskBytes: before.totalBytes - 1, highWaterBytes: 0 }, + }); + + expect(inspected.diskBudget?.totalBytesBefore).toBe(before.totalBytes); + expect(inspected.wouldMutate).toBe(false); + expect(result).toMatchObject({ overBudget: true, removedEntries: 0, removedFiles: 0 }); + expect(sessionExists("warn-old")).toBe(true); + expect(readArchiveNames("warn-old")).toHaveLength(0); + }); + + async function createHistoricalTranscript(params: { + content: string; + nextSessionId: string; + sessionId: string; + sessionKey: string; + updatedAt: number; + }): Promise { + await replaceSessionEntry( + { sessionKey: params.sessionKey, storePath }, + { sessionId: params.sessionId, updatedAt: params.updatedAt }, + ); + await appendTranscriptMessage( + { sessionId: params.sessionId, sessionKey: params.sessionKey, storePath }, + { message: { role: "user", content: params.content } }, + ); + await resetSessionEntryLifecycle({ + storePath, + target: { canonicalKey: params.sessionKey, storeKeys: [params.sessionKey] }, + buildNextEntry: () => ({ sessionId: params.nextSessionId, updatedAt: params.updatedAt + 1 }), + }); + setSessionUpdatedAt(params.sessionId, params.updatedAt); + } + + function database() { + const target = resolveSqliteTargetFromSessionStorePath(storePath); + if (!target.path) { + throw new Error("expected SQLite database path"); + } + return openOpenClawAgentDatabase({ agentId: target.agentId ?? "main", path: target.path }); + } + + function settlePhysicalUsage(): void { + const owner = database(); + owner.walMaintenance.checkpoint(); + const row = owner.db.prepare("PRAGMA freelist_count").get() as + | { freelist_count?: unknown } + | undefined; + const freePages = Number(row?.freelist_count ?? 0); + if (Number.isSafeInteger(freePages) && freePages > 0) { + owner.db.exec(`PRAGMA incremental_vacuum(${freePages});`); + } + owner.walMaintenance.checkpoint(); + } + + function setSessionUpdatedAt(sessionId: string, updatedAt: number): void { + const owner = database(); + const db = getSessionKysely(owner.db); + executeSqliteQuerySync( + owner.db, + db.updateTable("sessions").set({ updated_at: updatedAt }).where("session_id", "=", sessionId), + ); + } + + function addRouteReference(sessionKey: string, sessionId: string): void { + const owner = database(); + const db = getSessionKysely(owner.db); + executeSqliteQuerySync( + owner.db, + db + .insertInto("session_routes") + .values({ session_key: sessionKey, session_id: sessionId, updated_at: Date.now() }), + ); + } + + function sessionExists(sessionId: string): boolean { + const owner = database(); + const db = getSessionKysely(owner.db); + return ( + executeSqliteQuerySync( + owner.db, + db.selectFrom("sessions").select("session_id").where("session_id", "=", sessionId), + ).rows.length === 1 + ); + } + + function readArchiveNames(sessionId: string): string[] { + return fs.readdirSync(tempDir).filter((name) => name.startsWith(`${sessionId}.jsonl.deleted.`)); + } +}); + +function createTrajectoryEvent(sessionId: string, sessionKey: string): TrajectoryEvent { + return { + traceSchema: "openclaw-trajectory", + schemaVersion: 1, + traceId: sessionId, + source: "runtime", + type: "history.test", + ts: "2026-07-18T00:00:00.000Z", + seq: 1, + sessionId, + sessionKey, + }; +} + +describe("kickSessionHistoryDiskBudgetMaintenance", () => { + it("throttles repeat kicks and skips warn mode entirely", async () => { + const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-session-history-kick-")); + const storePath = path.join(tempDir, "sessions.json"); + const maintenance = { + mode: "warn", + maxDiskBytes: 1, + highWaterBytes: 0, + } as never; + // Warn mode must not schedule background enforcement at all. + kickSessionHistoryDiskBudgetMaintenance({ + storePath, + maintenanceConfig: maintenance, + }); + const enforceMaintenance = { + mode: "enforce", + maxDiskBytes: Number.MAX_SAFE_INTEGER, + highWaterBytes: Number.MAX_SAFE_INTEGER - 1, + } as never; + const first = Date.now(); + kickSessionHistoryDiskBudgetMaintenance({ + storePath, + maintenanceConfig: enforceMaintenance, + now: first, + }); + // Second kick inside the throttle window is a no-op (single-slot state). + kickSessionHistoryDiskBudgetMaintenance({ + storePath, + maintenanceConfig: enforceMaintenance, + now: first + 1_000, + }); + // Give the fire-and-forget pass a tick to settle; an under-budget store + // must leave every session untouched. + await new Promise((resolve) => { + setTimeout(resolve, 50); + }); + closeOpenClawAgentDatabasesForTest(); + fs.rmSync(tempDir, { recursive: true, force: true }); + }); +}); diff --git a/src/config/sessions/session-history-eviction.ts b/src/config/sessions/session-history-eviction.ts new file mode 100644 index 000000000000..59ea32a02b14 --- /dev/null +++ b/src/config/sessions/session-history-eviction.ts @@ -0,0 +1,444 @@ +import { executeSqliteQuerySync } from "../../infra/kysely-sync.js"; +import { + collectActiveSessionWorkAdmissionIdentities, + runExclusiveSessionLifecycleMutation, +} from "../../sessions/session-lifecycle-admission.js"; +import { runQueuedStoreWrite, type StoreWriterQueue } from "../../shared/store-writer-queue.js"; +import { + openOpenClawAgentDatabase, + runOpenClawAgentWriteTransaction, + type OpenClawAgentDatabase, +} from "../../state/openclaw-agent-db.js"; +import { + hasRetainedSessionTranscriptArchives, + measureSessionPhysicalDiskUsage, + pruneSessionTranscriptArchivesToHighWater, + type SessionDiskBudgetSweepResult, + type SessionPhysicalDiskUsage, +} from "./disk-budget.js"; +import { materializeSqliteSessionStateDeletePlans } from "./session-accessor.sqlite-archive.js"; +import { emitArchivedSqliteTranscriptUpdates } from "./session-accessor.sqlite-events.js"; +import { + collectSqliteSessionStateIdsForEntry, + deleteMaterializedSqliteSessionStatePlans, + planSqliteSessionStateDeleteIfUnreferenced, + readReferencedSqliteSessionIds, +} from "./session-accessor.sqlite-lifecycle-state.js"; +import { + getSessionKysely, + resolveSqliteScope, + resolveSqliteTranscriptArchiveDirectory, + runExclusiveSqliteSessionWrite, + toDatabaseOptions, +} from "./session-accessor.sqlite-scope.js"; +import { parseSqliteSessionEntryJson } from "./session-accessor.sqlite-status.js"; +import { normalizeStoreSessionKey } from "./store-entry.js"; +import { resolveMaintenanceConfig } from "./store-maintenance-runtime.js"; +import type { ResolvedSessionMaintenanceConfig } from "./store-maintenance.js"; + +type SessionHistoryDiskBudgetParams = { + agentId?: string; + mode: ResolvedSessionMaintenanceConfig["mode"]; + storePath: string; + maintenance: Pick; +}; + +function createPhysicalBudgetResult(params: { + totalBytesBefore: number; + totalBytesAfter?: number; + removedEntries?: number; + removedFiles?: number; + maxBytes: number; + highWaterBytes: number; +}): SessionDiskBudgetSweepResult { + const totalBytesAfter = params.totalBytesAfter ?? params.totalBytesBefore; + return { + totalBytesBefore: params.totalBytesBefore, + totalBytesAfter, + removedFiles: params.removedFiles ?? 0, + removedEntries: params.removedEntries ?? 0, + freedBytes: Math.max(0, params.totalBytesBefore - totalBytesAfter), + maxBytes: params.maxBytes, + highWaterBytes: params.highWaterBytes, + overBudget: params.totalBytesBefore > params.maxBytes, + }; +} + +/** Reports the same physical total enforce mode compares, without projecting logical row bytes. */ +export async function inspectSqliteSessionHistoryDiskBudget( + params: SessionHistoryDiskBudgetParams, +): Promise<{ diskBudget: SessionDiskBudgetSweepResult | null; wouldMutate: boolean }> { + const { highWaterBytes, maxDiskBytes } = params.maintenance; + if (maxDiskBytes == null || highWaterBytes == null) { + return { diskBudget: null, wouldMutate: false }; + } + const usage = await measureSessionPhysicalDiskUsage(params.storePath); + const diskBudget = createPhysicalBudgetResult({ + totalBytesBefore: usage.totalBytes, + maxBytes: maxDiskBytes, + highWaterBytes, + }); + if (!diskBudget.overBudget || params.mode !== "enforce") { + return { diskBudget, wouldMutate: false }; + } + // Predict only definite reclamation: prunable archives or unprotected + // historical generations. Checkpoint-only byte reclamation stays out of the + // preview; applied summaries report it via their byte-decrease predicate. + if (await hasRetainedSessionTranscriptArchives(params.storePath)) { + return { diskBudget, wouldMutate: true }; + } + const resolved = resolveSqliteScope({ + ...(params.agentId ? { agentId: params.agentId } : {}), + sessionKey: "", + storePath: params.storePath, + }); + const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); + const candidates = readHistoricalSessionIds({ + database, + protectedSessionIds: collectProtectedHistoricalSessionIds({ + database, + storePath: params.storePath, + }), + }); + return { diskBudget, wouldMutate: candidates.length > 0 }; +} + +function collectProtectedHistoricalSessionIds(params: { + database: OpenClawAgentDatabase; + storePath: string; +}): Set { + const protectedSessionIds = readReferencedSqliteSessionIds(params.database); + for (const sessionId of collectAdmissionProtectedSessionIds(params)) { + protectedSessionIds.add(sessionId); + } + return protectedSessionIds; +} + +/** Session ids owned by in-flight work admissions, without live-reference protection. */ +export function collectAdmissionProtectedSessionIds(params: { + database: OpenClawAgentDatabase; + storePath: string; +}): Set { + const protectedSessionIds = new Set(); + const admissionIdentities = collectActiveSessionWorkAdmissionIdentities(params.storePath); + if (admissionIdentities.size === 0) { + return protectedSessionIds; + } + + // Admissions may carry either the backing session id or its live session key. Protect both, + // then resolve admitted keys through their entries so cleanup cannot reclaim active work. + for (const identity of admissionIdentities) { + protectedSessionIds.add(identity); + } + const normalizedAdmissionKeys = new Set( + [...admissionIdentities].map((identity) => normalizeStoreSessionKey(identity)), + ); + const db = getSessionKysely(params.database.db); + const rows = executeSqliteQuerySync( + params.database.db, + db.selectFrom("session_entries").select(["entry_json", "session_id", "session_key"]), + ).rows; + for (const row of rows) { + if (!normalizedAdmissionKeys.has(normalizeStoreSessionKey(row.session_key))) { + continue; + } + protectedSessionIds.add(row.session_id); + const entry = parseSqliteSessionEntryJson(row); + if (entry) { + for (const sessionId of collectSqliteSessionStateIdsForEntry(entry)) { + protectedSessionIds.add(sessionId); + } + } + } + // Key-scoped admissions must survive rollover: an in-flight run admitted by + // key may still write to a generation the entry no longer references, so + // every generation of an admitted key stays off-limits. + const generationRows = executeSqliteQuerySync( + params.database.db, + db.selectFrom("sessions").select(["session_id", "session_key"]), + ).rows; + for (const row of generationRows) { + if (normalizedAdmissionKeys.has(normalizeStoreSessionKey(row.session_key))) { + protectedSessionIds.add(row.session_id); + } + } + return protectedSessionIds; +} + +function readHistoricalSessionIds(params: { + database: OpenClawAgentDatabase; + protectedSessionIds: ReadonlySet; +}): string[] { + const db = getSessionKysely(params.database.db); + return executeSqliteQuerySync( + params.database.db, + db + .selectFrom("sessions") + .select("session_id") + .orderBy("updated_at", "asc") + .orderBy("session_id", "asc"), + ).rows.flatMap((row) => (params.protectedSessionIds.has(row.session_id) ? [] : [row.session_id])); +} + +function reclaimSqliteFreePages(database: OpenClawAgentDatabase): void { + // 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(); + const row = database.db.prepare("PRAGMA freelist_count").get() as + | { freelist_count?: unknown } + | undefined; + const freePages = Number(row?.freelist_count ?? 0); + if (Number.isSafeInteger(freePages) && freePages > 0) { + database.db.exec(`PRAGMA incremental_vacuum(${freePages});`); + } + database.walMaintenance.checkpoint(); +} + +const PHYSICAL_BUDGET_CHECK_INTERVAL_MS = 30 * 60 * 1000; +// Single-slot per store: ordinary entry writes kick a throttled background +// budget pass so an over-budget database self-heals without waiting for a +// manual `sessions cleanup` invocation. +const budgetKickStateByStore = new Map< + string, + { lastCheckAt: number; running: boolean; pendingForce: boolean } +>(); + +/** Fire-and-forget budget pass from the ordinary entry-write maintenance seam. */ +export function kickSessionHistoryDiskBudgetMaintenance(params: { + agentId?: string; + storePath: string; + maintenanceConfig?: ResolvedSessionMaintenanceConfig; + now?: number; + /** Bypass the throttle interval; still single-slot per store. Used after + explicit deletion, which can double usage via freshly written archives. */ + force?: boolean; +}): void { + const maintenance = params.maintenanceConfig ?? resolveMaintenanceConfig(); + if ( + maintenance.mode !== "enforce" || + maintenance.maxDiskBytes == null || + maintenance.highWaterBytes == null + ) { + return; + } + const now = params.now ?? Date.now(); + const state = budgetKickStateByStore.get(params.storePath) ?? { + lastCheckAt: 0, + running: false, + pendingForce: false, + }; + if (state.running) { + // A running pass may already have taken its last measurement; a forced + // kick (post-delete spike) must not be dropped or the store could stay + // over budget until the next unrelated write. + state.pendingForce = state.pendingForce || params.force === true; + budgetKickStateByStore.set(params.storePath, state); + return; + } + if (!params.force && now - state.lastCheckAt < PHYSICAL_BUDGET_CHECK_INTERVAL_MS) { + // Dropped, not deferred: every entry write (including heartbeats) re-kicks, + // so a store that goes over budget is rechecked on the next activity. + // Reset/delete use force and bypass this window entirely. + return; + } + state.lastCheckAt = now; + state.running = true; + budgetKickStateByStore.set(params.storePath, state); + void enforceSqliteSessionHistoryDiskBudget({ + ...(params.agentId ? { agentId: params.agentId } : {}), + storePath: params.storePath, + mode: maintenance.mode, + maintenance, + }) + .catch(() => { + // Best-effort: budget pressure is retried on the next throttled kick. + }) + .finally(() => { + state.running = false; + if (state.pendingForce) { + state.pendingForce = false; + kickSessionHistoryDiskBudgetMaintenance({ ...params, force: true }); + } + }); +} + +// One enforcement pass per store at a time: overlapping passes (background +// kick vs `sessions cleanup`) would evict on stale usage measurements and +// prune each other's freshly extracted archives. +const SESSION_HISTORY_MAINTENANCE_QUEUES = new Map(); + +/** Extracts historical sessions durably before reclaiming their SQLite rows. */ +export async function enforceSqliteSessionHistoryDiskBudget( + params: SessionHistoryDiskBudgetParams, +): Promise { + return await runQueuedStoreWrite({ + queues: SESSION_HISTORY_MAINTENANCE_QUEUES, + storePath: params.storePath, + label: "enforceSqliteSessionHistoryDiskBudget", + fn: async () => await enforceSessionHistoryMaintenanceSerialized(params), + }); +} + +// Reclaims checkpointable pages, retained archives, then historical SQLite +// rows. Unreferenced session-dir artifacts (orphan transcripts, stale blobs) +// are owned by per-save store maintenance and `sessions cleanup`, not by this +// supplementary pressure pass. +async function enforceSessionHistoryMaintenanceSerialized( + params: SessionHistoryDiskBudgetParams, +): Promise { + const { highWaterBytes, maxDiskBytes } = params.maintenance; + if (maxDiskBytes == null || highWaterBytes == null) { + return null; + } + const initialUsage = await measureSessionPhysicalDiskUsage(params.storePath); + if (initialUsage.totalBytes <= maxDiskBytes || params.mode === "warn") { + return createPhysicalBudgetResult({ + totalBytesBefore: initialUsage.totalBytes, + maxBytes: maxDiskBytes, + highWaterBytes, + }); + } + + const resolved = resolveSqliteScope({ + ...(params.agentId ? { agentId: params.agentId } : {}), + sessionKey: "", + storePath: params.storePath, + }); + const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); + const archiveDirectory = resolveSqliteTranscriptArchiveDirectory(resolved); + let usage: SessionPhysicalDiskUsage = await runExclusiveSqliteSessionWrite(resolved, async () => { + reclaimSqliteFreePages(database); + return await measureSessionPhysicalDiskUsage(params.storePath); + }); + let removedEntries = 0; + let removedFiles = 0; + if (usage.totalBytes > highWaterBytes) { + // Archive pruning shares the session write lock so a concurrent + // reset/delete cannot race its archive file against the unlink pass. + const archiveSweep = await runExclusiveSqliteSessionWrite(resolved, async () => + pruneSessionTranscriptArchivesToHighWater({ + highWaterBytes, + storePath: params.storePath, + }), + ); + removedFiles = archiveSweep.removedFiles; + usage = archiveSweep.usage; + } + const candidates = readHistoricalSessionIds({ + database, + protectedSessionIds: collectProtectedHistoricalSessionIds({ + database, + storePath: params.storePath, + }), + }); + + for (const sessionId of candidates) { + if (usage.totalBytes <= highWaterBytes) { + break; + } + const eviction = await runExclusiveSessionLifecycleMutation({ + scope: params.storePath, + identities: [sessionId], + run: async () => + await runExclusiveSqliteSessionWrite(resolved, async () => { + const protectedBeforeArchive = collectProtectedHistoricalSessionIds({ + database, + storePath: params.storePath, + }); + const plan = planSqliteSessionStateDeleteIfUnreferenced({ + archiveDirectory, + archiveTranscript: true, + database, + reason: "deleted", + referencedSessionIds: protectedBeforeArchive, + sessionId, + }); + if (!plan) { + return null; + } + + // Extract-before-delete is the retention invariant. Admission is fenced across archive + // creation, then rechecked inside the write transaction before any rows are reclaimed. + const materialized = materializeSqliteSessionStateDeletePlans([plan]); + let deleted = false; + let archivedTranscripts: ReturnType = + []; + runOpenClawAgentWriteTransaction((transactionDb) => { + const protectedAtDelete = collectProtectedHistoricalSessionIds({ + database: transactionDb, + storePath: params.storePath, + }); + archivedTranscripts = deleteMaterializedSqliteSessionStatePlans( + transactionDb, + materialized, + protectedAtDelete, + ); + const db = getSessionKysely(transactionDb.db); + deleted = + executeSqliteQuerySync( + transactionDb.db, + db.selectFrom("sessions").select("session_id").where("session_id", "=", sessionId), + ).rows.length === 0; + }, toDatabaseOptions(resolved)); + if (!deleted) { + return null; + } + try { + // 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); + } catch { + // Best-effort reclamation only. + } + return { + archivedTranscripts, + usage: await measureSessionPhysicalDiskUsage(params.storePath), + }; + }), + }); + if (!eviction) { + continue; + } + removedEntries += 1; + emitArchivedSqliteTranscriptUpdates(eviction.archivedTranscripts); + usage = eviction.usage; + if (usage.totalBytes > highWaterBytes) { + // Reclaim archives (oldest first, including ones this pass committed) + // before spending another session's rows: each session's data should be + // destroyed at most once, and pruning an extracted copy beats evicting + // additional searchable history. No prune runs between an archive write + // and its row-deletion commit, so a sole copy is never mid-flight here. + const repruned = await runExclusiveSqliteSessionWrite(resolved, async () => + pruneSessionTranscriptArchivesToHighWater({ + highWaterBytes, + storePath: params.storePath, + }), + ); + removedFiles += repruned.removedFiles; + usage = repruned.usage; + } + } + + if (usage.totalBytes > highWaterBytes) { + // Candidates are exhausted but archives may remain; finish the pass at the + // target instead of returning over budget with removable artifacts. + const finalPrune = await runExclusiveSqliteSessionWrite(resolved, async () => + pruneSessionTranscriptArchivesToHighWater({ + highWaterBytes, + storePath: params.storePath, + }), + ); + removedFiles += finalPrune.removedFiles; + usage = finalPrune.usage; + } + + return createPhysicalBudgetResult({ + totalBytesBefore: initialUsage.totalBytes, + totalBytesAfter: usage.totalBytes, + removedEntries, + removedFiles, + maxBytes: maxDiskBytes, + highWaterBytes, + }); +} diff --git a/src/config/sessions/store-maintenance.ts b/src/config/sessions/store-maintenance.ts index b32b27a81dcd..a5652479859a 100644 --- a/src/config/sessions/store-maintenance.ts +++ b/src/config/sessions/store-maintenance.ts @@ -23,9 +23,8 @@ const DEFAULT_MODEL_RUN_PRUNE_AFTER_MS = 24 * 60 * 60 * 1000; const DEFAULT_SESSION_MAX_ENTRIES = 500; const DEFAULT_SESSION_MAINTENANCE_MODE: SessionMaintenanceMode = "enforce"; const DEFAULT_SESSION_DISK_BUDGET_HIGH_WATER_RATIO = 0.8; -// Archived transcripts are the user's conversation history: retention keeps -// them until the disk budget evicts oldest-first, never on a wall-clock timer. -// The budget default below is what makes "keep archives" still bounded. +// Conversation history stays in SQLite until physical main-file + WAL + artifact usage crosses +// this budget. Cleanup then extracts historical sessions oldest-first before reclaiming rows. const DEFAULT_SESSION_MAX_DISK_BYTES = 10 * 1024 * 1024 * 1024; const STRICT_ENTRY_MAINTENANCE_MAX_ENTRIES = 49; const MIN_BATCHED_ENTRY_MAINTENANCE_SLACK = 25; @@ -73,8 +72,8 @@ function resolvePruneAfterMs(maintenance?: SessionMaintenanceConfig): number { function resolveResetArchiveRetentionMs( maintenance: SessionMaintenanceConfig | undefined, ): number | null { - // null = keep archived transcripts indefinitely (disk budget still evicts - // oldest-first under pressure). An explicit duration opts back into + // null = keep extracted transcripts indefinitely (the disk budget still removes + // old archive artifacts under pressure). An explicit duration opts back into // wall-clock deletion; parse failures stay on the keep side because losing // history is the worse failure mode. const raw = maintenance?.resetArchiveRetention; diff --git a/src/config/sessions/store.pruning.integration.test.ts b/src/config/sessions/store.pruning.integration.test.ts index 7c78ae4d654b..157c8a4c461b 100644 --- a/src/config/sessions/store.pruning.integration.test.ts +++ b/src/config/sessions/store.pruning.integration.test.ts @@ -25,6 +25,7 @@ vi.mock("../config.js", async () => ({ import { getRuntimeConfig } from "../config.js"; import { runSessionsCleanup } from "./cleanup-service.js"; +import { measureSessionPhysicalDiskUsage } from "./disk-budget.js"; import { appendTranscriptMessage, listSessionEntries, @@ -32,6 +33,7 @@ import { loadTranscriptEventsSync, patchSessionEntry, replaceSessionEntry, + resetSessionEntryLifecycle, } from "./session-accessor.js"; import { registerSessionMaintenancePreserveKeysProvider } from "./store-maintenance-preserve.js"; import { @@ -879,7 +881,7 @@ describe("Integration: saveSessionStore with pruning", () => { await expectPathExists(oldOrphanTranscript); }); - it("sessions cleanup dry-run reports SQLite transcript row bytes for disk budget", async () => { + it("sessions cleanup dry-run reports physical SQLite usage without projecting row eviction", async () => { mockLoadConfig.mockReturnValue({ session: { maintenance: { @@ -924,11 +926,12 @@ describe("Integration: saveSessionStore with pruning", () => { throw new Error("expected SQLite row-byte disk budget summary"); } expect(diskBudgetSummary.totalBytesBefore).toBeGreaterThan(2_400); - expect(diskBudgetSummary.totalBytesAfter).toBeLessThan(diskBudgetSummary.totalBytesBefore); - expect(diskBudgetSummary.removedEntries).toBe(1); + expect(diskBudgetSummary.totalBytesAfter).toBe(diskBudgetSummary.totalBytesBefore); + expect(diskBudgetSummary.removedEntries).toBe(0); expect(diskBudgetSummary.removedFiles).toBe(0); - expect(preview?.summary.afterCount).toBe(1); - expect(preview?.budgetEvictedKeys.has(oldKey)).toBe(true); + expect(preview?.summary.afterCount).toBe(2); + expect(preview?.summary.wouldMutate).toBe(false); + expect(preview?.budgetEvictedKeys.has(oldKey)).toBe(false); expect(preview?.budgetEvictedKeys.has(freshKey)).toBe(false); expect(loadSessionEntry({ storePath, sessionKey: oldKey })).toBeDefined(); expect( @@ -940,7 +943,7 @@ describe("Integration: saveSessionStore with pruning", () => { ).toBeGreaterThan(0); }); - it("sessions cleanup apply reports SQLite disk-budget row eviction", async () => { + it("sessions cleanup preserves live SQLite entries even when their physical store is over budget", async () => { mockLoadConfig.mockReturnValue({ session: { maintenance: { @@ -984,10 +987,10 @@ describe("Integration: saveSessionStore with pruning", () => { if (diskBudgetSummary === null || diskBudgetSummary === undefined) { throw new Error("expected applied SQLite row-byte disk budget summary"); } - expect(diskBudgetSummary.removedEntries).toBe(1); + expect(diskBudgetSummary.removedEntries).toBe(0); expect(diskBudgetSummary.removedFiles).toBe(0); - expect(summary?.appliedCount).toBe(1); - expect(loadSessionEntry({ storePath, sessionKey: oldKey })).toBeUndefined(); + expect(summary?.appliedCount).toBe(2); + expect(loadSessionEntry({ storePath, sessionKey: oldKey })).toBeDefined(); expect(loadSessionEntry({ storePath, sessionKey: freshKey })).toBeDefined(); expect( loadTranscriptEventsSync({ @@ -995,7 +998,7 @@ describe("Integration: saveSessionStore with pruning", () => { sessionKey: oldKey, storePath, }), - ).toEqual([]); + ).not.toEqual([]); expect( loadTranscriptEventsSync({ sessionId: "fresh-apply-budget-session", @@ -1005,7 +1008,7 @@ describe("Integration: saveSessionStore with pruning", () => { ).toBeGreaterThan(0); }); - it("sessions cleanup dry-run accounts SQLite trajectory rows in disk budget", async () => { + it("sessions cleanup dry-run physically accounts for SQLite trajectory storage", async () => { mockLoadConfig.mockReturnValue({ session: { maintenance: { @@ -1043,9 +1046,9 @@ describe("Integration: saveSessionStore with pruning", () => { throw new Error("expected SQLite trajectory row-byte disk budget summary"); } expect(diskBudgetSummary.totalBytesBefore).toBeGreaterThan(1_200); - expect(diskBudgetSummary.totalBytesAfter).toBeLessThan(diskBudgetSummary.totalBytesBefore); - expect(diskBudgetSummary.removedEntries).toBe(1); - expect(dryRun.previewResults[0]?.budgetEvictedKeys.has(oldKey)).toBe(true); + expect(diskBudgetSummary.totalBytesAfter).toBe(diskBudgetSummary.totalBytesBefore); + expect(diskBudgetSummary.removedEntries).toBe(0); + expect(dryRun.previewResults[0]?.budgetEvictedKeys.has(oldKey)).toBe(false); expect(loadSessionEntry({ storePath, sessionKey: oldKey })).toBeDefined(); await expect( loadSqliteTrajectoryRuntimeEvents({ @@ -1055,7 +1058,7 @@ describe("Integration: saveSessionStore with pruning", () => { ).resolves.toHaveLength(1); }); - it("sessions cleanup apply removes budget-evicted SQLite trajectory rows", async () => { + it("sessions cleanup apply keeps trajectory rows for live entries", async () => { mockLoadConfig.mockReturnValue({ session: { maintenance: { @@ -1098,15 +1101,15 @@ describe("Integration: saveSessionStore with pruning", () => { if (diskBudgetSummary === null || diskBudgetSummary === undefined) { throw new Error("expected applied SQLite trajectory row-byte disk budget summary"); } - expect(diskBudgetSummary.removedEntries).toBe(1); - expect(loadSessionEntry({ storePath, sessionKey: oldKey })).toBeUndefined(); + expect(diskBudgetSummary.removedEntries).toBe(0); + expect(loadSessionEntry({ storePath, sessionKey: oldKey })).toBeDefined(); expect(loadSessionEntry({ storePath, sessionKey: freshKey })).toBeDefined(); await expect( loadSqliteTrajectoryRuntimeEvents({ sessionId: "old-trajectory-apply-budget-session", storePath, }), - ).resolves.toEqual([]); + ).resolves.toHaveLength(1); await expect( loadSqliteTrajectoryRuntimeEvents({ sessionId: "fresh-trajectory-apply-budget-session", @@ -1115,6 +1118,62 @@ describe("Integration: saveSessionStore with pruning", () => { ).resolves.toHaveLength(1); }); + it("sessions cleanup extracts and evicts historical SQLite rows under physical pressure", async () => { + const sessionKey = "agent:main:historical-budget"; + await seedSqliteSessionStore(storePath, { + [sessionKey]: { sessionId: "historical-budget-session", updatedAt: 1 }, + }); + await seedSqliteTranscriptMessage({ + content: "historical " + "x".repeat(128 * 1024), + sessionId: "historical-budget-session", + sessionKey, + storePath, + }); + await resetSessionEntryLifecycle({ + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + buildNextEntry: () => ({ sessionId: "live-budget-session", updatedAt: Date.now() }), + }); + const before = await measureSessionPhysicalDiskUsage(storePath); + mockLoadConfig.mockReturnValue({ + session: { + maintenance: { + mode: "enforce", + pruneAfter: "365d", + maxEntries: 500, + maxDiskBytes: before.totalBytes - 1, + highWaterBytes: 0, + }, + }, + }); + + const applied = await runSessionsCleanup({ + cfg: {}, + opts: { store: storePath, enforce: true }, + targets: [{ agentId: "main", storePath }], + }); + + expect(applied.appliedSummaries[0]?.diskBudget?.removedEntries).toBe(1); + expect(loadSessionEntry({ storePath, sessionKey })?.sessionId).toBe("live-budget-session"); + expect( + loadTranscriptEventsSync({ + sessionId: "historical-budget-session", + sessionKey, + storePath, + }), + ).toEqual([]); + // highWaterBytes 0 demands zero disk: after the extraction commits, the + // final unprotected prune reclaims even the fresh archive (counted in + // removedFiles). Archive survival under realistic budgets is covered by + // the session-history-eviction unit tests. + expect(applied.appliedSummaries[0]?.diskBudget?.removedFiles).toBe(1); + expect( + (await fs.readdir(testDir)).some((name) => + name.startsWith("historical-budget-session.jsonl.deleted."), + ), + ).toBe(false); + }); + it("sessions cleanup dry-run excludes stale and capped entry transcripts from orphan counts", async () => { mockLoadConfig.mockReturnValue({ session: { diff --git a/src/config/sessions/store.session-lifecycle-mutation.test.ts b/src/config/sessions/store.session-lifecycle-mutation.test.ts index a9a64755bd5e..47627a44a6b6 100644 --- a/src/config/sessions/store.session-lifecycle-mutation.test.ts +++ b/src/config/sessions/store.session-lifecycle-mutation.test.ts @@ -8,6 +8,7 @@ import { executeSqliteQueryTakeFirstSync, getNodeSqliteKysely, } from "../../infra/kysely-sync.js"; +import { beginSessionWorkAdmission } from "../../sessions/session-lifecycle-admission.js"; import { onInternalSessionTranscriptUpdate } from "../../sessions/transcript-events.js"; import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js"; import { openOpenClawAgentDatabase } from "../../state/openclaw-agent-db.js"; @@ -17,6 +18,7 @@ import { applySessionEntryLifecycleMutation, cleanupSessionLifecycleArtifacts, deleteSessionEntryLifecycle, + listSessionEntries, loadTranscriptEvents, loadSessionEntry, replaceSessionEntry, @@ -24,6 +26,7 @@ import { } from "./session-accessor.js"; import { replaceSqliteTranscriptEvents } from "./session-accessor.sqlite.js"; import { resolveSqliteTargetFromSessionStorePath } from "./session-sqlite-target.js"; +import { searchSessionTranscripts } from "./session-transcript-search.js"; import type { SessionEntry } from "./types.js"; type TestTranscriptEvent = Parameters[1][number]; @@ -42,7 +45,7 @@ describe("session store lifecycle mutations", () => { fs.rmSync(tempDir, { recursive: true, force: true }); }); - it("resets an entry in SQLite while archiving the previous transcript rows", async () => { + it("resets the live entry while keeping previous SQLite history searchable", async () => { const now = Date.now(); await replaceSessionEntry( { sessionKey: "agent:main:room", storePath }, @@ -71,7 +74,9 @@ describe("session store lifecycle mutations", () => { "post-compaction-session", ]) { await replaceSqliteTranscriptEvents({ sessionKey: "agent:main:room", sessionId, storePath }, [ - createTranscriptEvent(sessionId, `before reset ${sessionId}`), + sessionId === "old-session" + ? createSearchableTranscriptEvent(sessionId, "foreverneedle before reset") + : createTranscriptEvent(sessionId, `before reset ${sessionId}`), ]); } const transcriptUpdates = recordTranscriptUpdateFiles(); @@ -102,53 +107,51 @@ describe("session store lifecycle mutations", () => { const stored = loadSessionEntry({ sessionKey: "agent:main:room", storePath }); expect(stored?.sessionId).toBe("next-session"); expect(result.previousSessionId).toBe("old-session"); - expect(result.archivedTranscripts).toHaveLength(5); - expect(result.archivedTranscripts.map((transcript) => transcript.archivedPath)).toEqual( - expect.arrayContaining([ - expect.stringContaining("old-session.jsonl.reset."), - expect.stringContaining("usage-family-session.jsonl.reset."), - expect.stringContaining("checkpoint-session.jsonl.reset."), - expect.stringContaining("pre-compaction-session.jsonl.reset."), - expect.stringContaining("post-compaction-session.jsonl.reset."), - ]), - ); - expect(transcriptUpdates.files).toContain(result.archivedTranscripts[0]?.archivedPath); - expect(callbackTranscriptEvents).toEqual([]); - expect(readArchiveLinesForSession(result, "old-session")).toEqual([ - createTranscriptEventLine("old-session", "before reset old-session"), + expect(result.archivedTranscripts).toEqual([]); + expect(transcriptUpdates.files).toEqual([]); + expect(callbackTranscriptEvents).toEqual([ + createSearchableTranscriptEvent("old-session", "foreverneedle before reset"), ]); - expect(readArchiveLinesForSession(result, "usage-family-session")).toEqual([ - createTranscriptEventLine("usage-family-session", "before reset usage-family-session"), + expect(listSessionEntries({ storePath })).toEqual([ + { + sessionKey: "agent:main:room", + entry: expect.objectContaining({ sessionId: "next-session" }), + }, ]); - expect(readArchiveLinesForSession(result, "checkpoint-session")).toEqual([ - createTranscriptEventLine("checkpoint-session", "before reset checkpoint-session"), - ]); - expect(readArchiveLinesForSession(result, "pre-compaction-session")).toEqual([ - createTranscriptEventLine("pre-compaction-session", "before reset pre-compaction-session"), - ]); - expect(readArchiveLinesForSession(result, "post-compaction-session")).toEqual([ - createTranscriptEventLine("post-compaction-session", "before reset post-compaction-session"), + expect(readArchiveNames(path.dirname(storePath), "old-session.jsonl.reset.")).toEqual([]); + expect( + searchSessionTranscripts({ + agentId: "main", + env: { ...process.env, OPENCLAW_STATE_DIR: tempDir }, + query: "foreverneedle", + sessionKeys: ["agent:main:room"], + }).hits, + ).toEqual([ + expect.objectContaining({ + sessionId: "old-session", + sessionKey: "agent:main:room", + }), ]); await expect( loadTranscriptEvents({ sessionKey: "agent:main:room", sessionId: "old-session", storePath }), - ).resolves.toEqual([]); + ).resolves.toHaveLength(1); await expect( loadTranscriptEvents({ sessionKey: "agent:main:room", sessionId: "usage-family-session", storePath, }), - ).resolves.toEqual([]); + ).resolves.toHaveLength(1); await expect( loadTranscriptEvents({ sessionKey: "agent:main:room", sessionId: "pre-compaction-session", storePath, }), - ).resolves.toEqual([]); + ).resolves.toHaveLength(1); }); - it("archives old SQLite transcript rows before reset callbacks can fail", async () => { + it("keeps old SQLite rows when a post-reset callback fails", async () => { const now = Date.now(); await replaceSessionEntry( { sessionKey: "agent:main:callback-failure", storePath }, @@ -189,12 +192,108 @@ describe("session store lifecycle mutations", () => { sessionId: "callback-old-session", storePath, }), - ).resolves.toEqual([]); + ).resolves.toHaveLength(1); expect( - fs - .readdirSync(path.dirname(storePath)) - .filter((file) => file.startsWith("callback-old-session.jsonl.reset.")), - ).toHaveLength(1); + readArchiveNames(path.dirname(storePath), "callback-old-session.jsonl.reset."), + ).toHaveLength(0); + }); + + it("explicit delete archives and removes every retained reset generation", async () => { + const sessionKey = "agent:main:delete-history"; + const sessionIds = ["delete-history-one", "delete-history-two", "delete-history-three"]; + await replaceSessionEntry( + { sessionKey, storePath }, + { sessionId: "delete-history-one", updatedAt: 1 }, + ); + for (const [index, sessionId] of sessionIds.entries()) { + await replaceSqliteTranscriptEvents({ sessionId, sessionKey, storePath }, [ + createSearchableTranscriptEvent(sessionId, `deleteforever generation ${index + 1}`), + ]); + const nextSessionId = sessionIds[index + 1]; + if (nextSessionId) { + await resetSessionEntryLifecycle({ + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + buildNextEntry: () => ({ sessionId: nextSessionId, updatedAt: index + 2 }), + }); + } + } + expect( + searchSessionTranscripts({ + agentId: "main", + env: { ...process.env, OPENCLAW_STATE_DIR: tempDir }, + query: "deleteforever", + sessionKeys: [sessionKey], + }).hits, + ).toHaveLength(3); + + const result = await deleteSessionEntryLifecycle({ + archiveTranscript: true, + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + }); + + expect(result.deleted).toBe(true); + expect(result.archivedTranscripts).toHaveLength(3); + expect(listSessionEntries({ storePath })).toEqual([]); + for (const sessionId of sessionIds) { + await expect(loadTranscriptEvents({ sessionId, sessionKey, storePath })).resolves.toEqual([]); + expect(readArchiveNames(path.dirname(storePath), `${sessionId}.jsonl.deleted.`)).toHaveLength( + 1, + ); + } + expect( + searchSessionTranscripts({ + agentId: "main", + env: { ...process.env, OPENCLAW_STATE_DIR: tempDir }, + query: "deleteforever", + sessionKeys: [sessionKey], + }).hits, + ).toEqual([]); + }); + + it("explicit delete aborts while admitted work owns a retained generation", async () => { + const sessionKey = "agent:main:delete-admitted-history"; + await replaceSessionEntry({ sessionKey, storePath }, { sessionId: "admit-old", updatedAt: 1 }); + await replaceSqliteTranscriptEvents({ sessionId: "admit-old", sessionKey, storePath }, [ + createSearchableTranscriptEvent("admit-old", "admitted generation"), + ]); + await resetSessionEntryLifecycle({ + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + buildNextEntry: () => ({ sessionId: "admit-live", updatedAt: 2 }), + }); + const admission = await beginSessionWorkAdmission({ + scope: storePath, + identities: ["admit-old"], + assertAllowed: () => {}, + }); + try { + await expect( + deleteSessionEntryLifecycle({ + archiveTranscript: true, + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + }), + ).rejects.toThrow(/work is in flight/); + expect(listSessionEntries({ storePath })).toHaveLength(1); + await expect( + loadTranscriptEvents({ sessionId: "admit-old", sessionKey, storePath }), + ).resolves.toHaveLength(1); + } finally { + admission.release(); + } + + const result = await deleteSessionEntryLifecycle({ + archiveTranscript: true, + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + }); + expect(result.deleted).toBe(true); + // Only admit-old carries transcript events; admit-live never wrote any. + expect(result.archivedTranscripts).toHaveLength(1); + expect(readArchiveNames(path.dirname(storePath), "admit-old.jsonl.deleted.")).toHaveLength(1); + expect(listSessionEntries({ storePath })).toEqual([]); }); it("deletes an entry from SQLite while archiving unreferenced transcript rows", async () => { @@ -362,7 +461,9 @@ describe("session store lifecycle mutations", () => { expect(result.archivedTranscripts).toHaveLength(1); expect(entryObservedDuringArchiveRename).toEqual([true]); const archiveTempOpenIndexes = openSpy.mock.calls.flatMap((args, index) => - String(args[0]).includes("durable-delete-session.jsonl.deleted.") ? [index] : [], + String(args[0]).includes("durable-delete-session.jsonl.deleted.") && args[1] === "wx" + ? [index] + : [], ); expect(archiveTempOpenIndexes).toHaveLength(1); const archiveTempOpenIndex = archiveTempOpenIndexes[0] ?? -1; @@ -741,6 +842,15 @@ function createTranscriptEvent(sessionId: string, content: string): TestTranscri return JSON.parse(createTranscriptEventLine(sessionId, content)) as TestTranscriptEvent; } +function createSearchableTranscriptEvent(sessionId: string, content: string): TestTranscriptEvent { + return { + id: `message-${sessionId}`, + message: { role: "user", content }, + timestamp: "2026-07-18T00:00:00.000Z", + type: "message", + } as TestTranscriptEvent; +} + function createTranscriptEventLine(sessionId: string, content: string): string { return JSON.stringify({ type: "session", diff --git a/src/gateway/server.sessions.reset-hooks.test.ts b/src/gateway/server.sessions.reset-hooks.test.ts index 2fc3a72e6333..4ab52be0e4b8 100644 --- a/src/gateway/server.sessions.reset-hooks.test.ts +++ b/src/gateway/server.sessions.reset-hooks.test.ts @@ -554,13 +554,10 @@ test("sessions.reset emits enriched session_end and session_start hooks", async expect(endEvent.sessionId).toBe("sess-main"); expect(endEvent.sessionKey).toBe("agent:main:main"); expect(endEvent.reason).toBe("new"); - expect(endEvent.transcriptArchived).toBe(true); - const archivedSessionFile = expectStringWithPrefix( - path.basename(expectStringValue(endEvent.sessionFile, "archived session file")), - "sess-main.jsonl.reset.", - "archived session file", - ); - expect(archivedSessionFile).toContain(".jsonl.reset."); + // Retained history: reset keeps the SQLite transcript searchable under the + // same key, so nothing is archived and no reset artifact file exists. + expect(endEvent.transcriptArchived).toBeUndefined(); + expect(endEvent.sessionFile).toBeUndefined(); expect(endEvent.nextSessionId).toBe(startEvent.sessionId); expectMainHookContext(endContext, "sess-main"); expect(startEvent.sessionKey).toBe("agent:main:main"); diff --git a/src/gateway/server.sessions.store-rpc.test.ts b/src/gateway/server.sessions.store-rpc.test.ts index 22b2ba093241..30e86705f105 100644 --- a/src/gateway/server.sessions.store-rpc.test.ts +++ b/src/gateway/server.sessions.store-rpc.test.ts @@ -592,13 +592,15 @@ test("lists and patches session store via sessions.* RPC", async () => { const entryAfterReset = loadSessionEntry({ sessionKey: "agent:main:main", storePath }); expect(entryAfterReset?.lastAccountId).toBe("work"); expect(entryAfterReset?.lastThreadId).toBe("1737500000.123456"); + // Retained history: reset rotates the live session id but keeps the old + // generation's transcript rows in SQLite. await expect( loadTranscriptRows({ sessionId: "sess-main", sessionKey: "agent:main:main", storePath, }), - ).resolves.toEqual([]); + ).resolves.toHaveLength(3); const badThinking = await directSessionReq("sessions.patch", { key: "agent:main:main", diff --git a/test/helpers/sqlite-sessions-transcripts-flip-proof.ts b/test/helpers/sqlite-sessions-transcripts-flip-proof.ts index bda3dfd46f11..5bfb3a747c68 100644 --- a/test/helpers/sqlite-sessions-transcripts-flip-proof.ts +++ b/test/helpers/sqlite-sessions-transcripts-flip-proof.ts @@ -2492,12 +2492,14 @@ function validateCheckpointInvariants( }); } if (checkpoint.label === "after-sessions-reset") { - requireArchiveText(checkpoint, failures, { - description: "reset transcript archive", - includes: ["legacy hello", "sqlite user-facing send before reset"], - reason: "reset", - sessionId: context.legacySessionId, - }); + // Retained history: reset rotates the live session id but keeps the old + // generation's SQLite rows searchable; no reset archive is produced. + if (findArchiveArtifact(checkpoint, { reason: "reset", sessionId: context.legacySessionId })) { + failures.push(`${checkpoint.label}: unexpected reset transcript archive`); + } + if (checkpoint.sqlite.transcriptEvents === 0) { + failures.push(`${checkpoint.label}: retained transcript rows missing after reset`); + } } if (checkpoint.label === "after-sessions-delete") { requireArchiveText(checkpoint, failures, { diff --git a/test/scripts/sqlite-sessions-transcripts-flip-proof.built-cli.e2e.test.ts b/test/scripts/sqlite-sessions-transcripts-flip-proof.built-cli.e2e.test.ts index 900d07339906..2ddafdb89b87 100644 --- a/test/scripts/sqlite-sessions-transcripts-flip-proof.built-cli.e2e.test.ts +++ b/test/scripts/sqlite-sessions-transcripts-flip-proof.built-cli.e2e.test.ts @@ -119,12 +119,14 @@ describe("SQLite sessions/transcripts flip built CLI proof", () => { const resetCheckpoint = report.checkpoints.find( (checkpoint) => checkpoint.label === "after-sessions-reset", ); + // Retained history: reset keeps the old generation's SQLite rows and + // writes no reset archive artifact. const resetArchive = resetCheckpoint?.archiveArtifacts.find( (artifact) => artifact.archiveReason === "reset" && artifact.archiveSessionId === report.legacySessionId, ); - expect(resetArchive?.messageTexts).toContain("legacy hello"); - expect(resetArchive?.messageTexts).toContain("sqlite user-facing send before reset"); + expect(resetArchive).toBeUndefined(); + expect(resetCheckpoint?.sqlite.transcriptEvents ?? 0).toBeGreaterThan(0); const sharedFirstCheckpoint = report.checkpoints.find( (checkpoint) => checkpoint.label === "after-shared-first-delete", ); diff --git a/test/scripts/sqlite-sessions-transcripts-flip-proof.e2e.test.ts b/test/scripts/sqlite-sessions-transcripts-flip-proof.e2e.test.ts index b3fe7fb0e8cb..883ee9997841 100644 --- a/test/scripts/sqlite-sessions-transcripts-flip-proof.e2e.test.ts +++ b/test/scripts/sqlite-sessions-transcripts-flip-proof.e2e.test.ts @@ -118,12 +118,14 @@ describe("SQLite sessions/transcripts flip proof harness", () => { const resetCheckpoint = report.checkpoints.find( (checkpoint) => checkpoint.label === "after-sessions-reset", ); + // Retained history: reset keeps the old generation's SQLite rows and + // writes no reset archive artifact. const resetArchive = resetCheckpoint?.archiveArtifacts.find( (artifact) => artifact.archiveReason === "reset" && artifact.archiveSessionId === report.legacySessionId, ); - expect(resetArchive?.messageTexts).toContain("legacy hello"); - expect(resetArchive?.messageTexts).toContain("sqlite user-facing send before reset"); + expect(resetArchive).toBeUndefined(); + expect(resetCheckpoint?.sqlite.transcriptEvents ?? 0).toBeGreaterThan(0); expect( report.checkpoints.some( (checkpoint) =>