diff --git a/src/config/sessions/disk-budget.test.ts b/src/config/sessions/disk-budget.test.ts index d1e2e559fd14..a25ad7c2856a 100644 --- a/src/config/sessions/disk-budget.test.ts +++ b/src/config/sessions/disk-budget.test.ts @@ -378,6 +378,9 @@ describe("enforceSessionDiskBudget", () => { highWaterBytes: 1, }, warnOnly: false, + commitEvictedIndex: async () => { + await fs.writeFile(storePath, JSON.stringify(store, null, 2), "utf-8"); + }, }); expectBudgetResult(result); @@ -641,6 +644,117 @@ describe("enforceSessionDiskBudget", () => { expect(result.removedEntries).toBe(1); }); }); + + it("commits the reduced session index before deleting an evicted transcript", async () => { + await withTempDir({ prefix: "openclaw-disk-budget-commit-order-" }, async (dir) => { + const storePath = path.join(dir, "sessions.json"); + const oldKey = "agent:main:subagent:old-worker"; + const activeKey = "agent:main:main"; + const oldTranscript = path.join(dir, "old.jsonl"); + const activeTranscript = path.join(dir, "active.jsonl"); + const store: Record = { + [oldKey]: { sessionId: "old", updatedAt: 1 }, + [activeKey]: { sessionId: "active", updatedAt: 2 }, + }; + await fs.writeFile(storePath, JSON.stringify(store, null, 2), "utf-8"); + await fs.writeFile(oldTranscript, "t".repeat(10 * 1024), "utf-8"); + await fs.writeFile(activeTranscript, "a".repeat(64), "utf-8"); + + let commitCalls = 0; + let transcriptPresentAtCommit: boolean | null = null; + let indexPresentActiveOnlyAtCommit: boolean | null = null; + const result = await enforceSessionDiskBudget({ + store, + storePath, + activeSessionKey: activeKey, + maintenance: { maxDiskBytes: 100, highWaterBytes: 100 }, + warnOnly: false, + commitEvictedIndex: async () => { + commitCalls += 1; + transcriptPresentAtCommit = nodeFs.existsSync(oldTranscript); + await fs.writeFile(storePath, JSON.stringify({ [activeKey]: store[activeKey] }, null, 2)); + const persisted = JSON.parse(await fs.readFile(storePath, "utf-8")) as Record< + string, + SessionEntry + >; + indexPresentActiveOnlyAtCommit = + persisted[activeKey] !== undefined && persisted[oldKey] === undefined; + }, + }); + + expectBudgetResult(result); + expect(commitCalls).toBe(1); + expect(transcriptPresentAtCommit).toBe(true); + expect(indexPresentActiveOnlyAtCommit).toBe(true); + expect(result.removedEntries).toBe(1); + expect(result.removedFiles).toBeGreaterThanOrEqual(1); + expect(store[oldKey]).toBeUndefined(); + expect(store).toHaveProperty(activeKey); + await expectPathMissing(oldTranscript); + await expectPathExists(activeTranscript); + }); + }); + + it("retains the evicted transcript when the index commit fails", async () => { + await withTempDir({ prefix: "openclaw-disk-budget-commit-fail-" }, async (dir) => { + const storePath = path.join(dir, "sessions.json"); + const oldKey = "agent:main:subagent:old-worker"; + const activeKey = "agent:main:main"; + const oldTranscript = path.join(dir, "old.jsonl"); + const store: Record = { + [oldKey]: { sessionId: "old", updatedAt: 1 }, + [activeKey]: { sessionId: "active", updatedAt: 2 }, + }; + await fs.writeFile(storePath, JSON.stringify(store, null, 2), "utf-8"); + await fs.writeFile(oldTranscript, "t".repeat(10 * 1024), "utf-8"); + + const commitFailure = new Error("simulated store-write failure"); + await expect( + enforceSessionDiskBudget({ + store, + storePath, + activeSessionKey: activeKey, + maintenance: { maxDiskBytes: 100, highWaterBytes: 100 }, + warnOnly: false, + commitEvictedIndex: async () => { + throw commitFailure; + }, + }), + ).rejects.toBe(commitFailure); + + await expectPathExists(oldTranscript); + }); + }); + + it("retains evicted artifacts when no durable index commit is available", async () => { + await withTempDir({ prefix: "openclaw-disk-budget-missing-commit-" }, async (dir) => { + const storePath = path.join(dir, "sessions.json"); + const oldKey = "agent:main:subagent:old-worker"; + const activeKey = "agent:main:main"; + const oldTranscript = path.join(dir, "old.jsonl"); + const store: Record = { + [oldKey]: { sessionId: "old", updatedAt: 1 }, + [activeKey]: { sessionId: "active", updatedAt: 2 }, + }; + await fs.writeFile(storePath, JSON.stringify(store, null, 2), "utf-8"); + await fs.writeFile(oldTranscript, "t".repeat(10 * 1024), "utf-8"); + + const result = await enforceSessionDiskBudget({ + store, + storePath, + activeSessionKey: activeKey, + maintenance: { maxDiskBytes: 100, highWaterBytes: 100 }, + warnOnly: false, + }); + + expectBudgetResult(result); + expect(result.removedEntries).toBe(1); + expect(result.removedFiles).toBe(0); + expect(result.totalBytesAfter).toBeGreaterThan(result.highWaterBytes); + expect(store[oldKey]).toBeUndefined(); + await expectPathExists(oldTranscript); + }); + }); }); describe("pruneUnreferencedSessionArtifacts", () => { diff --git a/src/config/sessions/disk-budget.ts b/src/config/sessions/disk-budget.ts index ffb1f6130e86..efc0d33e2966 100644 --- a/src/config/sessions/disk-budget.ts +++ b/src/config/sessions/disk-budget.ts @@ -553,6 +553,7 @@ export async function enforceSessionDiskBudget(params: { dryRun?: boolean; log?: SessionDiskBudgetLogger; onRemoveFile?: (canonicalPath: string) => void; + commitEvictedIndex?: () => Promise; }): Promise { const maxBytes = params.maintenance.maxDiskBytes; const highWaterBytes = params.maintenance.highWaterBytes; @@ -637,6 +638,7 @@ export async function enforceSessionDiskBudget(params: { let removedFiles = 0; let removedEntries = 0; let freedBytes = 0; + const commitEvictedIndex = params.commitEvictedIndex; const referencedPaths = resolveReferencedSessionArtifactPaths({ sessionsDir, @@ -704,6 +706,27 @@ export async function enforceSessionDiskBudget(params: { removedFiles += 1; } + const deferredEvictedArtifactPaths: string[] = []; + const planEvictedArtifactRemoval = (rawPath: string, canonicalPathHint?: string): number => { + // An evicted artifact may only be unlinked after its reduced index is durable. + // Callers without that boundary retain the artifact as a reclaimable orphan. + if (!dryRun && !commitEvictedIndex) { + return 0; + } + const resolvedPath = path.resolve(rawPath); + const canonicalPath = canonicalPathHint ?? canonicalizePathForComparison(resolvedPath); + if (simulatedRemovedPaths.has(canonicalPath)) { + return 0; + } + const size = fileSizesByPath.get(canonicalPath) ?? 0; + if (size <= 0) { + return 0; + } + simulatedRemovedPaths.add(canonicalPath); + deferredEvictedArtifactPaths.push(resolvedPath); + return size; + }; + if (total > highWaterBytes) { const activeSessionKey = normalizeOptionalLowercaseString(params.activeSessionKey); const sessionIdRefCounts = buildSessionIdRefCounts(params.store); @@ -762,20 +785,16 @@ export async function enforceSessionDiskBudget(params: { tempStaleCutoffMs, ) ) { - const deletedBytes = await removePromptBlobFileForBudget({ - file: blobFile, - projectedPromptBlobRefCounts, - promptBlobCutoffMs: promptBlobOrphanCutoffMs, - tempCutoffMs: tempStaleCutoffMs, - dryRun, - fileSizesByPath, - simulatedRemovedPaths, - onRemovedPath: params.onRemoveFile, - }); - if (deletedBytes > 0) { - total -= deletedBytes; - freedBytes += deletedBytes; - removedFiles += 1; + const plannedBytes = planEvictedArtifactRemoval( + blobFile.path, + blobFile.canonicalPath, + ); + if (plannedBytes > 0) { + total -= plannedBytes; + if (dryRun) { + freedBytes += plannedBytes; + removedFiles += 1; + } } } } @@ -794,23 +813,37 @@ export async function enforceSessionDiskBudget(params: { } sessionIdRefCounts.delete(sessionId); for (const artifactPath of resolveSessionArtifactPathsForEntry({ sessionsDir, entry })) { - const deletedBytes = await removeFileForBudget({ - filePath: artifactPath, - dryRun, - fileSizesByPath, - simulatedRemovedPaths, - onRemovedPath: params.onRemoveFile, - }); - if (deletedBytes <= 0) { + const plannedBytes = planEvictedArtifactRemoval(artifactPath); + if (plannedBytes <= 0) { continue; } - total -= deletedBytes; - freedBytes += deletedBytes; - removedFiles += 1; + total -= plannedBytes; + if (dryRun) { + freedBytes += plannedBytes; + removedFiles += 1; + } } } } + if (!dryRun && commitEvictedIndex && deferredEvictedArtifactPaths.length > 0) { + await commitEvictedIndex(); + for (const filePath of deferredEvictedArtifactPaths) { + const deletedBytes = await removeFileForBudget({ + filePath, + dryRun: false, + fileSizesByPath, + simulatedRemovedPaths, + onRemovedPath: params.onRemoveFile, + }); + if (deletedBytes <= 0) { + continue; + } + freedBytes += deletedBytes; + removedFiles += 1; + } + } + if (!dryRun) { if (total > highWaterBytes) { log.warn("session disk budget still above high-water target after cleanup", { diff --git a/src/config/sessions/sessions.test.ts b/src/config/sessions/sessions.test.ts index 3a66b83fea95..53fdad332cfa 100644 --- a/src/config/sessions/sessions.test.ts +++ b/src/config/sessions/sessions.test.ts @@ -25,6 +25,7 @@ import { readSessionStoreCache, writeSessionStoreCache } from "./store-cache.js" import { clearSessionStoreCacheForTest, loadSessionStore, + saveSessionStore, updateSessionStore, updateSessionStoreEntry, } from "./store.js"; @@ -1000,6 +1001,38 @@ describe("session store writer queue", () => { writeSpy.mockRestore(); }); + it("uses a durable index write before disk-budget eviction deletes transcripts", async () => { + const oldKey = "agent:main:subagent:old-worker"; + const activeKey = "agent:main:main"; + const now = Date.now(); + const store: Record = { + [oldKey]: { sessionId: "old", updatedAt: now - 1_000 }, + [activeKey]: { sessionId: "active", updatedAt: now }, + }; + const { dir, storePath } = await makeTmpStore(store); + await fsPromises.writeFile(path.join(dir, "old.jsonl"), "t".repeat(10 * 1024), "utf-8"); + await fsPromises.writeFile(path.join(dir, "active.jsonl"), "a".repeat(64), "utf-8"); + + const writeSpy = vi.spyOn(jsonFiles, "writeTextAtomic"); + try { + await saveSessionStore(storePath, store, { + activeSessionKey: activeKey, + maintenanceOverride: { mode: "enforce", maxDiskBytes: 100, highWaterBytes: 100 }, + }); + + expect(writeSpy).toHaveBeenCalledTimes(1); + const [writtenPath, , writeOptions] = requireWriteTextAtomicCall(writeSpy); + expect(writtenPath).toBe(storePath); + expect(writeOptions?.durable).toBe(true); + expect(store[oldKey]).toBeUndefined(); + await expect(fsPromises.access(path.join(dir, "old.jsonl"))).rejects.toMatchObject({ + code: "ENOENT", + }); + } finally { + writeSpy.mockRestore(); + } + }); + it("can persist a known single entry without touching hydrated prompts from other sessions", async () => { const key = "agent:main:single-entry"; const otherKey = "agent:main:other-entry"; diff --git a/src/config/sessions/store-maintenance-operations.ts b/src/config/sessions/store-maintenance-operations.ts index 7ef4db41ca87..729dea3f660b 100644 --- a/src/config/sessions/store-maintenance-operations.ts +++ b/src/config/sessions/store-maintenance-operations.ts @@ -62,6 +62,7 @@ type FileBackedSessionStoreMaintenanceParams = { maintenanceConfig?: ResolvedSessionMaintenanceConfig; log: SessionMaintenanceLogger; artifacts: RemovedSessionArtifactCleanup; + commitReducedStore?: () => Promise; }; type FileBackedSessionStoreMaintenanceResult = { @@ -251,6 +252,7 @@ async function applyEnforcedMaintenance(params: { maintenance: params.maintenance, warnOnly: false, log: params.operation.log, + commitEvictedIndex: params.operation.commitReducedStore, }); await params.operation.onMaintenanceApplied?.({ mode: params.maintenance.mode, diff --git a/src/config/sessions/store.ts b/src/config/sessions/store.ts index b572a58aed2c..14184e499b99 100644 --- a/src/config/sessions/store.ts +++ b/src/config/sessions/store.ts @@ -746,6 +746,17 @@ async function saveSessionStoreUnlocked( let maintenanceChangedStore = false; if (!opts?.skipMaintenance) { + const commitReducedStore = async (): Promise => { + const projected = projectSessionStoreForPersistence({ storePath, store }); + await writeSessionStoreAtomic({ + storePath, + store, + serialized: JSON.stringify(projected.store, null, 2), + serializedPromptRefs: collectStorePromptRefs(projected.store), + promptBlobs: [...projected.promptBlobs.values()], + durable: true, + }); + }; const maintenance = await applyFileBackedSessionStoreMaintenance({ storePath, store, @@ -755,6 +766,7 @@ async function saveSessionStoreUnlocked( maintenanceOverride: opts?.maintenanceOverride, maintenanceConfig: opts?.maintenanceConfig, log, + commitReducedStore, artifacts: { archiveRemovedSessionTranscripts, removeRemovedSessionTrajectoryArtifacts: async (params) => { @@ -1211,12 +1223,13 @@ async function writeSessionStoreAtomic(params: { cloneSerialized?: string; promptBlobs: Iterable; takeOwnership?: boolean; + durable?: boolean; }): Promise { // Stage the temp as `sessions.json...tmp` (not the generic // `.fs-safe-replace.*`) so a temp orphaned by a crash between write and rename // is identifiable as a session-store temp and reclaimable by cleanup (#56827). await writeTextAtomic(params.storePath, params.serialized, { - durable: false, + durable: params.durable ?? false, mode: 0o600, tempPrefix: path.basename(params.storePath), beforeRename: async () => {