diff --git a/docs/.generated/sqlite-session-transcript-schema-baseline.sha256 b/docs/.generated/sqlite-session-transcript-schema-baseline.sha256 index 1d13df5ded6f..833b2f7270d3 100644 --- a/docs/.generated/sqlite-session-transcript-schema-baseline.sha256 +++ b/docs/.generated/sqlite-session-transcript-schema-baseline.sha256 @@ -1 +1 @@ -b7ce196c35d975dfefee75416c03573f94da00ec35495db8d3c22ab4a5bc72b7 sqlite-session-transcript-schema-baseline.sql +371735fe46e20b33bc5e942083c56040233cf5c9676b4f738f5beaf4c194f872 sqlite-session-transcript-schema-baseline.sql diff --git a/scripts/lib/sqlite-session-schema-baseline.ts b/scripts/lib/sqlite-session-schema-baseline.ts index 651119c37352..a79afa0c5137 100644 --- a/scripts/lib/sqlite-session-schema-baseline.ts +++ b/scripts/lib/sqlite-session-schema-baseline.ts @@ -37,6 +37,7 @@ const TARGET_TABLES = new Set([ "transcript_event_identities", "session_transcript_index_state", "session_transcript_active_events", + "session_transcript_archives", ]); function sha256(value: string): string { diff --git a/src/commands/doctor-session-canonical-keys.ts b/src/commands/doctor-session-canonical-keys.ts index 4ea00273c1d5..e8b6fff634cc 100644 --- a/src/commands/doctor-session-canonical-keys.ts +++ b/src/commands/doctor-session-canonical-keys.ts @@ -4,6 +4,7 @@ import { resolveSessionStorePathCore } from "../config/sessions/paths.js"; import { applySessionEntryLifecycleMutation, copySessionOwnedStateForCanonicalRepair, + ensureTranscriptGenerationsForCanonicalRepair, listSessionGenerationIdsForCanonicalRepair, loadTranscriptEvents, rehomeSessionDeliveryReferencesForCanonicalRepair, @@ -455,6 +456,7 @@ async function repairCanonicalSessionGroupsInSingleDatabase( if (!first) { return []; } + await ensureTranscriptGenerationsForCanonicalRepair(groups.flatMap((group) => group.candidates)); const destination = first.selected.destination; const result = await applySessionEntryLifecycleMutation({ agentId: destination.agentId, @@ -498,6 +500,7 @@ async function repairCanonicalSessionGroup( if (!selected) { return []; } + await ensureTranscriptGenerationsForCanonicalRepair(candidates); const winner = selected.winner; const destination = selected.destination; const byDatabase = new Map(); diff --git a/src/config/sessions/artifacts.test.ts b/src/config/sessions/artifacts.test.ts index ad324fa8ac4d..f4fab67802d2 100644 --- a/src/config/sessions/artifacts.test.ts +++ b/src/config/sessions/artifacts.test.ts @@ -16,6 +16,9 @@ import { describe("session artifact helpers", () => { it("classifies archived artifact file names", () => { expect(isSessionArchiveArtifactName("abc.jsonl.deleted.2026-01-01T00-00-00.000Z")).toBe(true); + expect( + isSessionArchiveArtifactName(`abc.jsonl.deleted.2026-01-01T00-00-00.000Z.${"a".repeat(32)}`), + ).toBe(true); expect(isSessionArchiveArtifactName("abc.jsonl.reset.2026-01-01T00-00-00.000Z")).toBe(true); expect(isSessionArchiveArtifactName("abc.jsonl.bak.2026-01-01T00-00-00.000Z")).toBe(true); expect(isSessionArchiveArtifactName("sessions.json.bak.1737420882")).toBe(true); @@ -93,6 +96,11 @@ describe("session artifact helpers", () => { expect( isUsageCountedSessionTranscriptFileName("abc.jsonl.deleted.2026-01-01T00-00-00.000Z"), ).toBe(true); + expect( + isUsageCountedSessionTranscriptFileName( + `abc.jsonl.deleted.2026-01-01T00-00-00.000Z.${"a".repeat(32)}`, + ), + ).toBe(true); expect(isUsageCountedSessionTranscriptFileName("abc.jsonl.bak.2026-01-01T00-00-00.000Z")).toBe( false, ); @@ -112,6 +120,11 @@ describe("session artifact helpers", () => { expect( parseUsageCountedSessionIdFromFileName("abc.jsonl.deleted.2026-01-01T00-00-00.000Z"), ).toBe("abc"); + expect( + parseUsageCountedSessionIdFromFileName( + `abc.jsonl.deleted.2026-01-01T00-00-00.000Z.${"a".repeat(32)}`, + ), + ).toBe("abc"); expect(parseUsageCountedSessionIdFromFileName("abc.jsonl.bak.2026-01-01T00-00-00.000Z")).toBe( null, ); @@ -130,6 +143,7 @@ describe("session artifact helpers", () => { const file = `abc.jsonl.deleted.${stamp}`; expect(parseSessionArchiveTimestamp(file, "deleted")).toBe(now); + expect(parseSessionArchiveTimestamp(`${file}.${"a".repeat(32)}`, "deleted")).toBe(now); expect(parseSessionArchiveTimestamp(file, "reset")).toBeNull(); expect(parseSessionArchiveTimestamp("keep.deleted.keep.jsonl", "deleted")).toBeNull(); }); diff --git a/src/config/sessions/artifacts.ts b/src/config/sessions/artifacts.ts index 2a5d5769d886..3a6753884eda 100644 --- a/src/config/sessions/artifacts.ts +++ b/src/config/sessions/artifacts.ts @@ -7,7 +7,8 @@ import { stripSessionArchiveCompressionSuffix } from "./archive-compression.js"; export type SessionArchiveReason = "bak" | "reset" | "deleted"; -const ARCHIVE_TIMESTAMP_RE = /^\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2}(?:\.\d{3})?Z$/; +const ARCHIVE_SUFFIX_RE = + /^(\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2}(?:\.\d{3})?Z)(?:\.([0-9a-f]{32}))?$/; const LEGACY_STORE_BACKUP_RE = /^sessions\.json\.bak\.\d+$/; const MIGRATION_ARCHIVE_RE = /\.migrated(?:\.\d+)?$/u; const COMPACTION_CHECKPOINT_TRANSCRIPT_RE = @@ -15,7 +16,7 @@ const COMPACTION_CHECKPOINT_TRANSCRIPT_RE = function hasArchiveSuffix(fileName: string, reason: SessionArchiveReason): boolean { // Compressed archives carry a trailing .zst; strip it so every classifier - // sees one canonical `.jsonl..` shape. + // sees one canonical `.jsonl..[.]` shape. const marker = `.${reason}.`; const normalized = stripSessionArchiveCompressionSuffix(fileName); const index = normalized.lastIndexOf(marker); @@ -23,7 +24,7 @@ function hasArchiveSuffix(fileName: string, reason: SessionArchiveReason): boole return false; } const raw = normalized.slice(index + marker.length); - return ARCHIVE_TIMESTAMP_RE.test(raw); + return ARCHIVE_SUFFIX_RE.test(raw); } /** Returns true for archived session artifacts and legacy store backup names. */ @@ -181,9 +182,10 @@ export function parseSessionArchiveTimestamp( if (!raw) { return null; } - if (!ARCHIVE_TIMESTAMP_RE.test(raw)) { + const timestampRaw = ARCHIVE_SUFFIX_RE.exec(raw)?.[1]; + if (!timestampRaw) { return null; } - const timestamp = Date.parse(restoreSessionArchiveTimestamp(raw)); + const timestamp = Date.parse(restoreSessionArchiveTimestamp(timestampRaw)); return Number.isNaN(timestamp) ? null : timestamp; } diff --git a/src/config/sessions/cleanup-service.fix-missing.test.ts b/src/config/sessions/cleanup-service.fix-missing.test.ts new file mode 100644 index 000000000000..665d23b25c96 --- /dev/null +++ b/src/config/sessions/cleanup-service.fix-missing.test.ts @@ -0,0 +1,362 @@ +import fs from "node:fs"; +import path from "node:path"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; +import { + closeOpenClawAgentDatabasesForTest, + openOpenClawAgentDatabase, +} from "../../state/openclaw-agent-db.js"; +import { readSessionArchiveContentSync } from "./archive-compression.js"; +import { isRetainedSessionTranscriptArchiveName } from "./artifacts.js"; +import { runSessionsCleanup } from "./cleanup-service.js"; +import { + appendTranscriptEventSync, + appendTranscriptMessageSync, + applySessionEntryLifecycleMutation, + inspectTranscriptEventsSync, + loadSessionEntry, + replaceSessionEntry, +} from "./session-accessor.js"; +import { prunePublishedSessionArchivesByRetention } from "./session-accessor.sqlite-archive-store.js"; +import { resolveSqliteTargetFromSessionStorePath } from "./session-sqlite-target.js"; + +const tempDirs = useAutoCleanupTempDirTracker(afterEach); + +function listDeletedArchives(directory: string): string[] { + if (!fs.existsSync(directory)) { + return []; + } + return fs + .readdirSync(directory) + .filter(isRetainedSessionTranscriptArchiveName) + .map((entry) => path.join(directory, entry)); +} + +describe("sessions cleanup --fix-missing", () => { + let storePath: string; + + beforeEach(() => { + const tempDir = tempDirs.make("openclaw-cleanup-fix-missing-"); + storePath = path.join(tempDir, "agents", "main", "sessions", "sessions.json"); + }); + + afterEach(() => { + closeOpenClawAgentDatabasesForTest(); + }); + + it("preserves readable session state when a later transcript row is malformed", async () => { + const sessionKey = "agent:main:malformed-after-message"; + const sessionId = "malformed-after-message"; + const scope = { sessionKey, sessionId, storePath }; + await replaceSessionEntry(scope, { sessionId, updatedAt: Date.now() }); + appendTranscriptMessageSync(scope, { + eventId: "readable-user-message", + message: { role: "user", content: [{ type: "text", text: "keep this conversation" }] }, + }); + appendTranscriptEventSync(scope, { type: "proof", id: "row-to-corrupt" }); + + const sqlitePath = resolveSqliteTargetFromSessionStorePath(storePath, { agentId: "main" }).path; + if (!sqlitePath) { + throw new Error("expected SQLite session store"); + } + const database = openOpenClawAgentDatabase({ agentId: "main", path: sqlitePath }); + database.db + .prepare( + `UPDATE transcript_events + SET event_json = '{malformed' + WHERE session_id = ? AND seq = ( + SELECT MAX(seq) FROM transcript_events WHERE session_id = ? + )`, + ) + .run(sessionId, sessionId); + + const result = await runSessionsCleanup({ + cfg: {}, + opts: { enforce: true, fixMissing: true }, + targets: [{ agentId: "main", storePath }], + }); + + expect(result.appliedSummaries[0]?.missing).toBe(0); + expect(loadSessionEntry({ sessionKey, storePath })).toMatchObject({ sessionId }); + expect(listDeletedArchives(path.dirname(storePath))).toEqual([]); + }); + + it("archives raw non-message rows before removing a confirmed missing session", async () => { + const sessionKey = "agent:main:message-free"; + const sessionId = "message-free"; + const scope = { sessionKey, sessionId, storePath }; + await replaceSessionEntry(scope, { sessionId, updatedAt: Date.now() }); + appendTranscriptEventSync(scope, { + type: "proof", + id: "raw-event", + content: "recoverable non-message state", + }); + const rawEventJson = + '{ "content": "recoverable non-message state", "id": "raw-event", "type": "proof" }'; + const sqlitePath = resolveSqliteTargetFromSessionStorePath(storePath, { agentId: "main" }).path; + if (!sqlitePath) { + throw new Error("expected SQLite session store"); + } + openOpenClawAgentDatabase({ agentId: "main", path: sqlitePath }) + .db.prepare("UPDATE transcript_events SET event_json = ? WHERE session_id = ?") + .run(rawEventJson, sessionId); + + const result = await runSessionsCleanup({ + cfg: {}, + opts: { enforce: true, fixMissing: true }, + targets: [{ agentId: "main", storePath }], + }); + + expect(result.appliedSummaries[0]?.missing).toBe(1); + expect(loadSessionEntry({ sessionKey, storePath })).toBeUndefined(); + const archives = listDeletedArchives(path.dirname(storePath)); + expect(archives).toHaveLength(1); + expect(readSessionArchiveContentSync(archives[0] ?? "")).toBe(`${rawEventJson}\n`); + expect( + openOpenClawAgentDatabase({ agentId: "main", path: sqlitePath }) + .db.prepare( + `SELECT session_key, reason, published_at + FROM session_transcript_archives WHERE session_id = ?`, + ) + .get(sessionId), + ).toMatchObject({ + published_at: expect.any(Number), + reason: "deleted", + session_key: sessionKey, + }); + }); + + it("recreates every derived file from pending canonical archives after commit", async () => { + const sessionIds = Array.from({ length: 6 }, (_, index) => `pending-export-${index}`); + for (const sessionId of sessionIds) { + const scope = { + sessionId, + sessionKey: `agent:main:${sessionId}`, + storePath, + }; + await replaceSessionEntry(scope, { sessionId, updatedAt: Date.now() }); + appendTranscriptEventSync(scope, { + type: "proof", + content: `recover after commit ${sessionId}`, + }); + } + + await runSessionsCleanup({ + cfg: {}, + opts: { enforce: true, fixMissing: true }, + targets: [{ agentId: "main", storePath }], + }); + + const archives = listDeletedArchives(path.dirname(storePath)); + expect(archives).toHaveLength(sessionIds.length); + for (const archivePath of archives) { + fs.rmSync(archivePath); + } + const sqlitePath = resolveSqliteTargetFromSessionStorePath(storePath, { agentId: "main" }).path; + if (!sqlitePath) { + throw new Error("expected SQLite session store"); + } + openOpenClawAgentDatabase({ agentId: "main", path: sqlitePath }) + .db.prepare( + `UPDATE session_transcript_archives + SET published_at = NULL, last_publish_error = 'simulated crash'`, + ) + .run(); + + await runSessionsCleanup({ + cfg: {}, + opts: { enforce: true, fixMissing: true }, + targets: [{ agentId: "main", storePath }], + }); + + const recovered = listDeletedArchives(path.dirname(storePath)); + expect(recovered).toHaveLength(sessionIds.length); + for (const sessionId of sessionIds) { + const archivePath = recovered.find((candidate) => + path.basename(candidate).startsWith(`${sessionId}.jsonl.deleted.`), + ); + expect(archivePath).toBeTruthy(); + expect(readSessionArchiveContentSync(archivePath ?? "")).toContain( + `recover after commit ${sessionId}`, + ); + } + const statuses = openOpenClawAgentDatabase({ agentId: "main", path: sqlitePath }) + .db.prepare( + `SELECT session_id, published_at, publish_attempts, last_publish_error + FROM session_transcript_archives ORDER BY session_id`, + ) + .all(); + expect(statuses).toHaveLength(sessionIds.length); + for (const status of statuses) { + expect(status).toMatchObject({ + last_publish_error: null, + publish_attempts: 2, + published_at: expect.any(Number), + }); + } + }); + + it("never overwrites a different derived-file collision", async () => { + const sessionKey = "agent:main:collision"; + const sessionId = "collision"; + const scope = { sessionKey, sessionId, storePath }; + await replaceSessionEntry(scope, { sessionId, updatedAt: Date.now() }); + appendTranscriptEventSync(scope, { type: "proof", content: "canonical bytes" }); + await runSessionsCleanup({ + cfg: {}, + opts: { enforce: true, fixMissing: true }, + targets: [{ agentId: "main", storePath }], + }); + const archivePath = listDeletedArchives(path.dirname(storePath))[0]; + expect(archivePath).toBeTruthy(); + fs.writeFileSync(archivePath ?? "", "different bytes", "utf8"); + const sqlitePath = resolveSqliteTargetFromSessionStorePath(storePath, { agentId: "main" }).path; + if (!sqlitePath) { + throw new Error("expected SQLite session store"); + } + const database = openOpenClawAgentDatabase({ agentId: "main", path: sqlitePath }); + database.db + .prepare("UPDATE session_transcript_archives SET published_at = NULL WHERE session_id = ?") + .run(sessionId); + + await expect( + runSessionsCleanup({ + cfg: {}, + opts: { enforce: true, fixMissing: true }, + targets: [{ agentId: "main", storePath }], + }), + ).rejects.toThrow("remain pending in SQLite"); + + expect(fs.readFileSync(archivePath ?? "", "utf8")).toBe("different bytes"); + expect( + database.db + .prepare( + "SELECT published_at, last_publish_error FROM session_transcript_archives WHERE session_id = ?", + ) + .get(sessionId), + ).toMatchObject({ + last_publish_error: expect.stringContaining("collision"), + published_at: null, + }); + }); + + it("rolls back the canonical archive when lifecycle deletion fails", async () => { + const sessionKey = "agent:main:rollback-delete"; + const sessionId = "rollback-delete"; + const scope = { sessionKey, sessionId, storePath }; + await replaceSessionEntry(scope, { sessionId, updatedAt: Date.now() }); + appendTranscriptEventSync(scope, { type: "proof", content: "must remain live" }); + const sqlitePath = resolveSqliteTargetFromSessionStorePath(storePath, { agentId: "main" }).path; + if (!sqlitePath) { + throw new Error("expected SQLite session store"); + } + const database = openOpenClawAgentDatabase({ agentId: "main", path: sqlitePath }); + database.db.exec(` + CREATE TRIGGER fail_session_window_delete + BEFORE DELETE ON session_windows + WHEN OLD.session_id = '${sessionId}' + BEGIN + SELECT RAISE(ABORT, 'injected lifecycle delete failure'); + END; + `); + + await expect( + runSessionsCleanup({ + cfg: {}, + opts: { enforce: true, fixMissing: true }, + targets: [{ agentId: "main", storePath }], + }), + ).rejects.toThrow("injected lifecycle delete failure"); + + expect(loadSessionEntry({ sessionKey, storePath })).toMatchObject({ sessionId }); + expect( + database.db.prepare("SELECT 1 FROM transcript_events WHERE session_id = ?").get(sessionId), + ).toEqual({ 1: 1 }); + expect( + database.db + .prepare("SELECT 1 FROM session_transcript_archives WHERE session_id = ?") + .get(sessionId), + ).toBeUndefined(); + expect(listDeletedArchives(path.dirname(storePath))).toEqual([]); + }); + + it("omits a cleanup removal whose transcript classification became stale", async () => { + const sessionKey = "agent:main:stale-classification"; + const sessionId = "stale-classification"; + const scope = { sessionKey, sessionId, storePath }; + const entry = { sessionId, updatedAt: Date.now() }; + await replaceSessionEntry(scope, entry); + const observation = inspectTranscriptEventsSync(scope).snapshot; + appendTranscriptMessageSync(scope, { + eventId: "message-after-classification", + message: { role: "user", content: [{ type: "text", text: "now live" }] }, + }); + + const result = await applySessionEntryLifecycleMutation({ + storePath, + removals: [ + { + sessionKey, + expectedEntry: entry, + expectedTranscriptSnapshot: observation, + archiveRemovedTranscript: true, + }, + ], + skipMaintenance: true, + }); + + expect(result.removedSessionKeys).toEqual([]); + expect(loadSessionEntry({ sessionKey, storePath })).toMatchObject({ sessionId }); + expect(listDeletedArchives(path.dirname(storePath))).toEqual([]); + }); + + it("drops a retained canonical row only after retention removes its derived file", async () => { + const sessionKey = "agent:main:retention"; + const sessionId = "retention"; + const scope = { sessionKey, sessionId, storePath }; + await replaceSessionEntry(scope, { sessionId, updatedAt: Date.now() }); + appendTranscriptEventSync(scope, { type: "proof", content: "expire together" }); + await runSessionsCleanup({ + cfg: {}, + opts: { enforce: true, fixMissing: true }, + targets: [{ agentId: "main", storePath }], + }); + const archivePath = listDeletedArchives(path.dirname(storePath))[0]; + expect(archivePath).toBeTruthy(); + const sqlitePath = resolveSqliteTargetFromSessionStorePath(storePath, { agentId: "main" }).path; + if (!sqlitePath) { + throw new Error("expected SQLite session store"); + } + const database = openOpenClawAgentDatabase({ agentId: "main", path: sqlitePath }); + database.db + .prepare("UPDATE session_transcript_archives SET created_at = 1 WHERE session_id = ?") + .run(sessionId); + + expect( + await prunePublishedSessionArchivesByRetention({ + scope: { agentId: "main", path: sqlitePath }, + rules: [{ reason: "deleted", olderThanMs: 10 }], + nowMs: 100, + }), + ).toBe(0); + expect( + database.db + .prepare("SELECT 1 FROM session_transcript_archives WHERE session_id = ?") + .get(sessionId), + ).toEqual({ 1: 1 }); + + fs.rmSync(archivePath ?? ""); + expect( + await prunePublishedSessionArchivesByRetention({ + scope: { agentId: "main", path: sqlitePath }, + rules: [{ reason: "deleted", olderThanMs: 10 }], + nowMs: 100, + }), + ).toBe(1); + expect( + database.db + .prepare("SELECT 1 FROM session_transcript_archives WHERE session_id = ?") + .get(sessionId), + ).toBeUndefined(); + }); +}); diff --git a/src/config/sessions/cleanup-service.ts b/src/config/sessions/cleanup-service.ts index e035fbf2ea1f..b53d4f24aca8 100644 --- a/src/config/sessions/cleanup-service.ts +++ b/src/config/sessions/cleanup-service.ts @@ -16,8 +16,8 @@ import { import { resolveSessionStorePathCore } from "./paths.js"; import { applySessionEntryLifecycleMutation, + inspectTranscriptEventsSync, listSessionEntriesCore, - loadTranscriptEventsSync, purgeDeletedAgentSessionEntries, type SessionEntryLifecycleRemoval, } from "./session-accessor.js"; @@ -156,15 +156,16 @@ function isTranscriptMessageRecord(entry: unknown): boolean { return record.type === undefined && isTranscriptMessageRole(record.role); } -function sqliteTranscriptHasMessageRecords(params: { +function inspectConfirmedMessageFreeTranscript(params: { sessionId: string; sessionKey: string; storePath: string; -}): boolean { +}) { try { - return loadTranscriptEventsSync(params).some(isTranscriptMessageRecord); + const inspection = inspectTranscriptEventsSync(params); + return inspection.events.some(isTranscriptMessageRecord) ? undefined : inspection; } catch { - return false; + return undefined; } } @@ -285,7 +286,11 @@ export function serializeSessionCleanupResult(params: { function pruneMissingTranscriptEntries(params: { store: Record; storePath: string; - onPruned?: (key: string, entry: SessionEntry) => void; + onPruned?: ( + key: string, + entry: SessionEntry, + inspection?: ReturnType, + ) => void; }): number { let removed = 0; for (const [key, entry] of Object.entries(params.store)) { @@ -316,16 +321,15 @@ function pruneMissingTranscriptEntries(params: { params.onPruned?.(key, entry); continue; } - if ( - !sqliteTranscriptHasMessageRecords({ - sessionId: entry.sessionId, - sessionKey: key, - storePath: params.storePath, - }) - ) { + const inspection = inspectConfirmedMessageFreeTranscript({ + sessionId: entry.sessionId, + sessionKey: key, + storePath: params.storePath, + }); + if (inspection) { delete params.store[key]; removed += 1; - params.onPruned?.(key, entry); + params.onPruned?.(key, entry, inspection); } } return removed; @@ -555,10 +559,12 @@ export async function runSessionsCleanup(params: { pruneMissingTranscriptEntries({ store: applyStore, storePath: target.storePath, - onPruned: (sessionKey, entry) => { + onPruned: (sessionKey, entry, inspection) => { missingRemovals.push({ sessionKey, expectedEntry: structuredClone(entry), + archiveRemovedTranscript: true, + ...(inspection ? { expectedTranscriptSnapshot: inspection.snapshot } : {}), }); }, }); diff --git a/src/config/sessions/disk-budget.ts b/src/config/sessions/disk-budget.ts index 0a3893bd3970..0162cecd2a95 100644 --- a/src/config/sessions/disk-budget.ts +++ b/src/config/sessions/disk-budget.ts @@ -318,6 +318,7 @@ export async function hasRetainedSessionTranscriptArchives(storePath: string): P /** Removes oldest retained reset/delete archives, remeasuring physical usage after each file. */ export async function pruneSessionTranscriptArchivesToHighWater(params: { + excludeNames?: ReadonlySet; highWaterBytes: number; storePath: string; }): Promise<{ removedFiles: number; usage: SessionPhysicalDiskUsage }> { @@ -325,7 +326,10 @@ export async function pruneSessionTranscriptArchivesToHighWater(params: { // 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)) + .filter( + (file) => + isRetainedSessionTranscriptArchiveName(file.name) && !params.excludeNames?.has(file.name), + ) .toSorted((left, right) => left.mtimeMs - right.mtimeMs); let usage = await measureSessionPhysicalDiskUsage(params.storePath); let removedFiles = 0; diff --git a/src/config/sessions/session-accessor.entry.ts b/src/config/sessions/session-accessor.entry.ts index 69fbad8279d4..d2151d9e3a8f 100644 --- a/src/config/sessions/session-accessor.entry.ts +++ b/src/config/sessions/session-accessor.entry.ts @@ -11,6 +11,7 @@ import { resolveSessionStorePathCore } from "./paths.js"; import { clearPluginOwnedSessionState } from "./plugin-host-cleanup.js"; import { copySqliteSessionOwnedStateForCanonicalRepair as copySessionOwnedStateForCanonicalRepair, + ensureSqliteTranscriptGenerationsForCanonicalRepair as ensureTranscriptGenerationsForCanonicalRepair, listSqliteSessionEntriesForCanonicalRepair as listSessionEntriesForCanonicalRepair, listSqliteSessionGenerationIdsForCanonicalRepair as listSessionGenerationIdsForCanonicalRepair, rehomeSqliteSessionDeliveryReferencesForCanonicalRepair as rehomeSessionDeliveryReferencesForCanonicalRepair, @@ -67,6 +68,7 @@ export { clearPluginOwnedSessionState }; export { countSessionEntryRowsReadOnly, copySessionOwnedStateForCanonicalRepair, + ensureTranscriptGenerationsForCanonicalRepair, ensureSessionEntrySync, hasSessionEntriesByStatusReadOnly, listSessionGenerationIdsForCanonicalRepair, diff --git a/src/config/sessions/session-accessor.lifecycle-types.ts b/src/config/sessions/session-accessor.lifecycle-types.ts index 5ba3dd58d56f..08f751f04995 100644 --- a/src/config/sessions/session-accessor.lifecycle-types.ts +++ b/src/config/sessions/session-accessor.lifecycle-types.ts @@ -1,5 +1,6 @@ import type { OpenClawConfig } from "../types.openclaw.js"; import type { SessionUnreferencedArtifactSweepResult } from "./disk-budget.js"; +import type { SessionStateDeleteSnapshot } from "./session-accessor.sqlite-delete-snapshot.types.js"; import type { SessionResetBoundaryReason } from "./session-reset-boundary-event.js"; import type { SessionMaintenanceApplyReport } from "./store-maintenance-operations.js"; import type { SessionEntry } from "./types.js"; @@ -27,6 +28,9 @@ export type SessionLifecycleStoreTarget = { }; export type SessionLifecycleArchivedTranscript = { + /** Canonical SQLite archive identity used for idempotent derived-file publication. */ + generation: string; + sessionId: string; sourcePath: string; archivedPath: string; }; @@ -107,6 +111,8 @@ type SessionEntryLifecycleRemovalBase = { /** Doctor cross-store repair only: delivery aliases copied under the canonical destination key. */ deliveryCleanupKeys?: readonly string[]; archiveRemovedTranscript?: boolean; + /** Omit removal when the transcript changed after the caller's positive classification. */ + expectedTranscriptSnapshot?: SessionStateDeleteSnapshot; expectedSessionId?: string; expectedLifecycleRevision?: string; expectedUpdatedAt?: number; diff --git a/src/config/sessions/session-accessor.sqlite-archive-store.ts b/src/config/sessions/session-accessor.sqlite-archive-store.ts new file mode 100644 index 000000000000..49807fef9dc3 --- /dev/null +++ b/src/config/sessions/session-accessor.sqlite-archive-store.ts @@ -0,0 +1,315 @@ +import fs from "node:fs"; +import path from "node:path"; +import { + executeSqliteQuerySync, + executeSqliteQueryTakeFirstSync, +} from "../../infra/kysely-sync.js"; +import { + openOpenClawAgentDatabase, + runOpenClawAgentWriteTransaction, + type OpenClawAgentDatabase, +} from "../../state/openclaw-agent-db.js"; +import { ensureSessionTranscriptArchiveSchema } from "../../state/openclaw-agent-session-transcript-archive-schema.js"; +import { + runSqliteTranscriptArchivePublishWorker, + type MaterializedSessionStateDeletePlan, +} from "./session-accessor.sqlite-archive.js"; +import type { SessionLifecycleArchivedTranscript } from "./session-accessor.sqlite-contract.js"; +import { emitArchivedTranscriptUpdates } from "./session-accessor.sqlite-events.js"; +import { + getSessionKysely, + resolveSqliteTranscriptArchiveDirectory, + runExclusiveSqliteSessionWrite, + toDatabaseOptions, + type ResolvedSqliteReadScope, +} from "./session-accessor.sqlite-scope.js"; + +/** Inserts the canonical archive row inside the lifecycle deletion transaction. */ +export function persistSessionTranscriptArchive( + database: OpenClawAgentDatabase, + plan: MaterializedSessionStateDeletePlan, +): void { + const archive = plan.archive; + const generation = plan.snapshot.generation; + const sessionKey = plan.snapshot.sessionKey; + if (!archive || !generation || !sessionKey) { + throw new Error( + `Cannot persist SQLite transcript archive without an owner generation for ${plan.sessionId}`, + ); + } + ensureSessionTranscriptArchiveSchema(database.db); + const db = getSessionKysely(database.db); + executeSqliteQuerySync( + database.db, + db + .insertInto("session_transcript_archives") + .values({ + archive_blob: archive.bytes, + archive_name: archive.archiveName, + archive_sha256: archive.sha256, + created_at: archive.createdAt, + encoding: archive.encoding, + generation, + last_publish_attempt_at: null, + last_publish_error: null, + published_at: null, + reason: plan.reason, + session_id: plan.sessionId, + session_key: sessionKey, + }) + .onConflict((conflict) => conflict.columns(["session_id", "generation"]).doNothing()), + ); + const persisted = executeSqliteQueryTakeFirstSync( + database.db, + db + .selectFrom("session_transcript_archives") + .select([ + "archive_blob", + "archive_name", + "archive_sha256", + "created_at", + "encoding", + "reason", + "session_key", + ]) + .where("session_id", "=", plan.sessionId) + .where("generation", "=", generation), + ); + if ( + !persisted || + persisted.archive_name !== archive.archiveName || + persisted.archive_sha256 !== archive.sha256 || + persisted.created_at !== archive.createdAt || + persisted.encoding !== archive.encoding || + persisted.reason !== plan.reason || + persisted.session_key !== sessionKey || + !Buffer.from(persisted.archive_blob).equals(Buffer.from(archive.bytes)) + ) { + throw new Error(`Conflicting SQLite transcript archive for ${plan.sessionId}`); + } +} + +const PENDING_ARCHIVE_PUBLISH_BATCH_SIZE = 4; + +// Composite map keys keep repeated physical IDs distinct across transcript rewrites. +function transcriptArchiveIdentityKey(sessionId: string, generation: string): string { + return `${sessionId}\u0000${generation}`; +} + +// Retain one publication plan per immutable archive identity. +function uniqueTranscriptArchives( + archives: readonly T[], +): T[] { + return [ + ...new Map( + archives.map((archive) => [ + transcriptArchiveIdentityKey(archive.sessionId, archive.generation), + archive, + ]), + ).values(), + ]; +} + +/** Publishes derived archive files after their canonical rows and deletions commit. */ +export async function publishSessionStateArchives( + scope: Pick, + requested: readonly SessionLifecycleArchivedTranscript[], +): Promise { + const requestedArchives = uniqueTranscriptArchives(requested); + const requestedIdentitySet = new Set( + requestedArchives.map((archive) => + transcriptArchiveIdentityKey(archive.sessionId, archive.generation), + ), + ); + let includeRequested = true; + while (true) { + const plans = await runExclusiveSqliteSessionWrite(scope, async () => { + const database = openOpenClawAgentDatabase(toDatabaseOptions(scope)); + const db = getSessionKysely(database.db); + if (includeRequested && requestedArchives.length > 0) { + ensureSessionTranscriptArchiveSchema(database.db); + } else { + const exists = executeSqliteQueryTakeFirstSync( + database.db, + db + .selectFrom("sqlite_schema") + .select("name") + .where("type", "=", "table") + .where("name", "=", "session_transcript_archives"), + ); + if (!exists) { + return []; + } + } + const pendingArchives = executeSqliteQuerySync( + database.db, + db + .selectFrom("session_transcript_archives") + .select(["generation", "session_id"]) + .where("published_at", "is", null) + .orderBy("created_at", "asc") + .orderBy("session_id", "asc") + .orderBy("generation", "asc") + .limit(PENDING_ARCHIVE_PUBLISH_BATCH_SIZE), + ).rows.map((row) => ({ generation: row.generation, sessionId: row.session_id })); + const archives = uniqueTranscriptArchives([ + ...(includeRequested ? requestedArchives : []), + ...pendingArchives, + ]); + const archiveDirectory = resolveSqliteTranscriptArchiveDirectory(scope); + return archives.map((archive) => ({ + agentId: database.agentId, + archiveDirectory, + databasePath: database.path, + generation: archive.generation, + sessionId: archive.sessionId, + })); + }); + includeRequested = false; + if (plans.length === 0) { + break; + } + + const results = await runSqliteTranscriptArchivePublishWorker(plans); + await runExclusiveSqliteSessionWrite(scope, async () => { + const now = Date.now(); + runOpenClawAgentWriteTransaction((transactionDb) => { + ensureSessionTranscriptArchiveSchema(transactionDb.db); + const db = getSessionKysely(transactionDb.db); + for (const result of results) { + executeSqliteQuerySync( + transactionDb.db, + db + .updateTable("session_transcript_archives") + .set((eb) => ({ + last_publish_attempt_at: now, + last_publish_error: result.error?.slice(0, 1024) ?? null, + publish_attempts: eb("publish_attempts", "+", 1), + ...(result.archivedPath ? { published_at: now } : {}), + })) + .where("session_id", "=", result.sessionId) + .where("generation", "=", result.generation), + ); + } + }, toDatabaseOptions(scope)); + }); + + const planByIdentity = new Map( + plans.map((plan) => [transcriptArchiveIdentityKey(plan.sessionId, plan.generation), plan]), + ); + emitArchivedTranscriptUpdates( + results.flatMap((result) => { + const identity = transcriptArchiveIdentityKey(result.sessionId, result.generation); + if (!result.archivedPath || requestedIdentitySet.has(identity)) { + return []; + } + const plan = planByIdentity.get(identity); + return plan + ? [ + { + archivedPath: result.archivedPath, + generation: result.generation, + sessionId: result.sessionId, + sourcePath: path.join(plan.archiveDirectory, `${result.sessionId}.jsonl`), + }, + ] + : []; + }), + ); + const failedIds = results.flatMap((result) => (result.archivedPath ? [] : [result.sessionId])); + if (failedIds.length > 0) { + throw new Error( + `Session deletion committed, but ${failedIds.length} transcript archive file export(s) remain pending in SQLite; retry the operation to publish them.`, + ); + } + } + return [...requested]; +} + +const ARCHIVE_RETENTION_BATCH_SIZE = 256; + +/** Removes canonical rows only after retention has removed their derived files. */ +export async function prunePublishedSessionArchivesByRetention(params: { + nowMs?: number; + rules: readonly { olderThanMs: number; reason: "deleted" | "reset" }[]; + scope: Pick; +}): Promise { + const rules = new Map( + params.rules + .filter((rule) => Number.isFinite(rule.olderThanMs) && rule.olderThanMs >= 0) + .map((rule) => [rule.reason, rule.olderThanMs] as const), + ); + if (rules.size === 0) { + return 0; + } + const candidates = await runExclusiveSqliteSessionWrite(params.scope, async () => { + const database = openOpenClawAgentDatabase(toDatabaseOptions(params.scope)); + const db = getSessionKysely(database.db); + const exists = executeSqliteQueryTakeFirstSync( + database.db, + db + .selectFrom("sqlite_schema") + .select("name") + .where("type", "=", "table") + .where("name", "=", "session_transcript_archives"), + ); + if (!exists) { + return []; + } + return executeSqliteQuerySync( + database.db, + db + .selectFrom("session_transcript_archives") + .select([ + "archive_name", + "created_at", + "generation", + "published_at", + "reason", + "session_id", + ]) + .where("published_at", "is not", null) + .orderBy("created_at", "asc") + .orderBy("session_id", "asc") + .orderBy("generation", "asc") + .limit(ARCHIVE_RETENTION_BATCH_SIZE), + ).rows; + }); + const now = params.nowMs ?? Date.now(); + const archiveDirectory = resolveSqliteTranscriptArchiveDirectory(params.scope); + const removable = candidates.filter((row) => { + const olderThanMs = rules.get(row.reason as "deleted" | "reset"); + if (olderThanMs === undefined || now - row.created_at <= olderThanMs) { + return false; + } + const archivePath = path.resolve(archiveDirectory, row.archive_name); + return ( + path.dirname(archivePath) === path.resolve(archiveDirectory) && + path.basename(archivePath) === row.archive_name && + !fs.existsSync(archivePath) + ); + }); + if (removable.length === 0) { + return 0; + } + return await runExclusiveSqliteSessionWrite(params.scope, async () => { + let removed = 0; + runOpenClawAgentWriteTransaction((transactionDb) => { + const db = getSessionKysely(transactionDb.db); + for (const row of removable) { + const result = executeSqliteQuerySync( + transactionDb.db, + db + .deleteFrom("session_transcript_archives") + .where("session_id", "=", row.session_id) + .where("generation", "=", row.generation) + .where("archive_name", "=", row.archive_name) + .where("created_at", "=", row.created_at) + .where("published_at", "=", row.published_at), + ); + removed += Number(result.numAffectedRows ?? 0n); + } + }, toDatabaseOptions(params.scope)); + return removed; + }); +} diff --git a/src/config/sessions/session-accessor.sqlite-archive.ts b/src/config/sessions/session-accessor.sqlite-archive.ts index 72f8ecb4aa39..358d6b19a913 100644 --- a/src/config/sessions/session-accessor.sqlite-archive.ts +++ b/src/config/sessions/session-accessor.sqlite-archive.ts @@ -1,4 +1,4 @@ -import { randomUUID } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import fs from "node:fs"; import path from "node:path"; import { fileURLToPath, pathToFileURL } from "node:url"; @@ -6,6 +6,7 @@ import { Worker } from "node:worker_threads"; import { toStringifiedError } from "@openclaw/normalization-core/error-coercion"; import { syncDirectoryBestEffortSync } from "../../infra/directory-durability.js"; import { KeyedAsyncQueue } from "../../plugin-sdk/keyed-async-queue.js"; +import { withOpenClawAgentDatabaseReadOnly } from "../../state/openclaw-agent-db-readonly.js"; import { encodeSessionArchiveContent, readSessionArchiveContentSync, @@ -13,15 +14,11 @@ import { } from "./archive-compression.js"; import { formatSessionArchiveTimestamp, type SessionArchiveReason } from "./artifacts.js"; import type { SessionLifecycleArchivedTranscript } from "./session-accessor.sqlite-contract.js"; - -export type SessionStateDeleteSnapshot = { - acpParentStreamEventCount: number; - generation: string | null; - lastSeq: number | null; - sessionUpdatedAt: number | null; - trajectoryLastSeq: number | null; - transcriptUpdatedAt: number | null; -}; +import { + readSessionStateDeleteSnapshot, + sqliteSessionStateDeleteSnapshotsEqual, +} from "./session-accessor.sqlite-delete-snapshot.js"; +import type { SessionStateDeleteSnapshot } from "./session-accessor.sqlite-delete-snapshot.types.js"; export type SessionStateDeletePlan = { agentId: string; @@ -34,16 +31,25 @@ export type SessionStateDeletePlan = { }; export type MaterializedSessionStateDeletePlan = SessionStateDeletePlan & { + archive: MaterializedSessionTranscriptArchive | null; archivedTranscript: SessionLifecycleArchivedTranscript | null; }; +type MaterializedSessionTranscriptArchive = { + archiveName: string; + bytes: Uint8Array; + createdAt: number; + encoding: "identity" | "zstd"; + sha256: string; +}; + export type TranscriptArchiveWorkerPlan = Pick< SessionStateDeletePlan, "agentId" | "archiveDirectory" | "databasePath" | "reason" | "sessionId" | "snapshot" >; export type TranscriptArchiveWorkerResult = { - archivedPath: string | null; + archive: MaterializedSessionTranscriptArchive | null; sessionId: string; }; @@ -52,30 +58,38 @@ export type TranscriptArchiveWorkerMessage = { results: TranscriptArchiveWorkerResult[]; }; -export function sqliteSessionStateDeleteSnapshotsEqual( - left: SessionStateDeleteSnapshot, - right: SessionStateDeleteSnapshot, -): boolean { - return ( - left.acpParentStreamEventCount === right.acpParentStreamEventCount && - left.generation === right.generation && - left.lastSeq === right.lastSeq && - left.sessionUpdatedAt === right.sessionUpdatedAt && - left.trajectoryLastSeq === right.trajectoryLastSeq && - left.transcriptUpdatedAt === right.transcriptUpdatedAt - ); -} +export type TranscriptArchivePublishPlan = { + agentId: string; + archiveDirectory: string; + databasePath: string; + generation: string; + sessionId: string; +}; + +export type TranscriptArchivePublishResult = { + archivedPath?: string; + error?: string; + generation: string; + sessionId: string; +}; + +export type TranscriptArchivePublishWorkerMessage = { + type: "published"; + results: TranscriptArchivePublishResult[]; +}; function resolveSqliteTranscriptArchivePath(params: { archiveDirectory: string; + generation?: string; reason: SessionArchiveReason; sessionId: string; nowMs?: number; }): string { const archiveDirectory = path.resolve(params.archiveDirectory); + const generationSuffix = params.generation ? `.${params.generation}` : ""; const archivePath = path.resolve( archiveDirectory, - `${params.sessionId}.jsonl.${params.reason}.${formatSessionArchiveTimestamp(params.nowMs)}`, + `${params.sessionId}.jsonl.${params.reason}.${formatSessionArchiveTimestamp(params.nowMs)}${generationSuffix}`, ); if (path.dirname(archivePath) !== archiveDirectory) { throw new Error(`Cannot archive SQLite transcript outside ${archiveDirectory}`); @@ -83,6 +97,32 @@ function resolveSqliteTranscriptArchivePath(params: { return archivePath; } +export function encodeMaterializedSessionTranscriptArchive(params: { + archiveDirectory: string; + content: string; + generation: string; + reason: SessionArchiveReason; + sessionId: string; + nowMs?: number; +}): MaterializedSessionTranscriptArchive { + const encoded = encodeSessionArchiveContent(params.content); + const createdAt = params.nowMs ?? Date.now(); + const archivedPath = `${resolveSqliteTranscriptArchivePath({ + archiveDirectory: params.archiveDirectory, + generation: params.generation, + reason: params.reason, + sessionId: params.sessionId, + nowMs: createdAt, + })}${encoded.suffix}`; + return { + archiveName: path.basename(archivedPath), + bytes: encoded.bytes, + createdAt, + encoding: encoded.suffix ? "zstd" : "identity", + sha256: createHash("sha256").update(encoded.bytes).digest("hex"), + }; +} + function findMatchingSqliteTranscriptArchive(params: { archiveDirectory: string; content: string; @@ -181,6 +221,51 @@ function writeDurableFileExclusive(filePath: string, content: Buffer): void { } } +export function hashSessionArchiveBytes(bytes: Uint8Array): string { + return createHash("sha256").update(bytes).digest("hex"); +} + +/** Publishes one exact canonical archive without directory scans or replacement. */ +export function publishEncodedSessionTranscriptArchive(params: { + archiveDirectory: string; + archiveName: string; + bytes: Uint8Array; + sha256: string; +}): string { + const archiveDirectory = path.resolve(params.archiveDirectory); + const archivePath = path.resolve(archiveDirectory, params.archiveName); + if ( + path.dirname(archivePath) !== archiveDirectory || + path.basename(archivePath) !== params.archiveName + ) { + throw new Error(`Cannot publish SQLite transcript archive outside ${archiveDirectory}`); + } + fs.mkdirSync(archiveDirectory, { recursive: true, mode: 0o700 }); + if (fs.existsSync(archivePath)) { + if (hashSessionArchiveBytes(fs.readFileSync(archivePath)) !== params.sha256) { + throw new Error(`SQLite transcript archive collision for ${params.archiveName}`); + } + return archivePath; + } + + const tempPath = `${archivePath}.${randomUUID()}.tmp`; + writeDurableFileExclusive(tempPath, Buffer.from(params.bytes)); + try { + fs.linkSync(tempPath, archivePath); + } catch (error) { + if ((error as { code?: unknown }).code !== "EEXIST") { + throw error; + } + } finally { + fs.rmSync(tempPath, { force: true }); + } + syncDirectoryBestEffortSync(archiveDirectory); + if (hashSessionArchiveBytes(fs.readFileSync(archivePath)) !== params.sha256) { + throw new Error(`SQLite transcript archive verification failed for ${params.archiveName}`); + } + return archivePath; +} + function resolveSqliteTranscriptArchiveWorkerUrl(currentModuleUrl = import.meta.url): URL { const currentPath = fileURLToPath(currentModuleUrl); const normalized = currentPath.replaceAll(path.sep, "/"); @@ -206,9 +291,10 @@ function resolveSourceWorkerExecArgv(): string[] { return ["--import", `data:text/javascript,${encodeURIComponent(registerTsx)}`]; } -function spawnSqliteTranscriptArchiveWorker( - plans: readonly TranscriptArchiveWorkerPlan[], -): Promise { +function spawnSqliteTranscriptArchiveWorker(params: { + expectedMessageType: "done" | "published"; + workerData: object; +}): Promise { const workerUrl = resolveSqliteTranscriptArchiveWorkerUrl(); let worker: Worker; try { @@ -216,7 +302,7 @@ function spawnSqliteTranscriptArchiveWorker( ? resolveSourceWorkerExecArgv() : undefined; worker = new Worker(workerUrl, { - workerData: { type: "sqlite-transcript-archive-v1", plans }, + workerData: params.workerData, execArgv: sourceWorkerExecArgv, }); } catch (error) { @@ -224,11 +310,16 @@ function spawnSqliteTranscriptArchiveWorker( } return new Promise((resolve, reject) => { - let results: TranscriptArchiveWorkerResult[] | undefined; + let results: Result[] | undefined; let workerError: Error | undefined; - worker.once("message", (message: TranscriptArchiveWorkerMessage) => { - results = message.results; - }); + worker.once( + "message", + (message: TranscriptArchiveWorkerMessage | TranscriptArchivePublishWorkerMessage) => { + if (message.type === params.expectedMessageType) { + results = message.results as Result[]; + } + }, + ); worker.once("error", (error) => { // An uncaught Worker error is followed by exit. Wait for that event so // callers never race the Worker's SQLite/file handles on Windows. @@ -263,41 +354,108 @@ function runSqliteTranscriptArchiveWorker( ): Promise { return sqliteTranscriptArchiveWorkerQueue.enqueue( SQLITE_TRANSCRIPT_ARCHIVE_WORKER_QUEUE_KEY, - () => spawnSqliteTranscriptArchiveWorker(plans), + () => + spawnSqliteTranscriptArchiveWorker({ + expectedMessageType: "done", + workerData: { operation: "materialize", type: "sqlite-transcript-archive-v2", plans }, + }), ); } -// Runs duplicate probing, archive write, rename, fsync, and readback outside -// SQLite write transactions and off the gateway event loop. The lifecycle -// Worker queue and per-call dedupe prevent concurrent whole-buffer spikes -// within this path. +export function runSqliteTranscriptArchivePublishWorker( + plans: readonly TranscriptArchivePublishPlan[], +): Promise { + return sqliteTranscriptArchiveWorkerQueue.enqueue( + SQLITE_TRANSCRIPT_ARCHIVE_WORKER_QUEUE_KEY, + () => + spawnSqliteTranscriptArchiveWorker({ + expectedMessageType: "published", + workerData: { operation: "publish", type: "sqlite-transcript-archive-v2", plans }, + }), + ); +} + +function validateEmptyTranscriptArchivePlan(plan: TranscriptArchiveWorkerPlan): void { + const opened = withOpenClawAgentDatabaseReadOnly( + (database) => readSessionStateDeleteSnapshot(database.db, plan.sessionId), + { agentId: plan.agentId, path: plan.databasePath }, + ); + if (!opened.found) { + throw new Error( + `Cannot archive SQLite transcript ${plan.sessionId}: ${opened.reason.replaceAll("-", " ")}`, + ); + } + if (!sqliteSessionStateDeleteSnapshotsEqual(opened.value, plan.snapshot)) { + throw new Error( + `SQLite session state changed before archive materialization for ${plan.sessionId}`, + ); + } +} + +// Reads and encodes one consistent generation outside SQLite write transactions +// and off the gateway event loop. The lifecycle Worker queue and per-call +// dedupe prevent concurrent whole-buffer spikes within this path. export async function materializeSessionStateDeletePlans( plans: readonly SessionStateDeletePlan[], ): Promise { const deduped = dedupeSqliteSessionStateDeletePlans(plans); const archivePlans = deduped.filter((plan) => plan.archiveTranscript); - const workerResults = - archivePlans.length > 0 ? await runSqliteTranscriptArchiveWorker(archivePlans) : []; + const workerResults: TranscriptArchiveWorkerResult[] = []; + let materializedBytes = 0; + // Archive bytes must never accumulate for an unbounded cleanup batch. One + // generation crosses the Worker boundary at a time, then becomes transaction input. + for (const archivePlan of archivePlans) { + if (archivePlan.snapshot.lastSeq === null) { + // Empty transcripts still need a fresh snapshot fence, but have no bytes + // to encode off-thread and should not pay Worker startup latency. + validateEmptyTranscriptArchivePlan(archivePlan); + workerResults.push({ archive: null, sessionId: archivePlan.sessionId }); + continue; + } + const [result] = await runSqliteTranscriptArchiveWorker([archivePlan]); + if (result) { + materializedBytes += result.archive?.bytes.byteLength ?? 0; + if (materializedBytes > MAX_MATERIALIZED_ARCHIVE_BATCH_BYTES) { + throw new Error( + `SQLite transcript archive batch exceeds ${MAX_MATERIALIZED_ARCHIVE_BATCH_BYTES} bytes; retry with fewer sessions.`, + ); + } + workerResults.push(result); + } + } const resultBySessionId = new Map(workerResults.map((result) => [result.sessionId, result])); return deduped.map((plan) => { if (!plan.archiveTranscript) { - return Object.assign({}, plan, { archivedTranscript: null }); + return Object.assign({}, plan, { archive: null, archivedTranscript: null }); } const result = resultBySessionId.get(plan.sessionId); if (!result) { throw new Error(`SQLite transcript archive worker omitted ${plan.sessionId}`); } - const archivedTranscript = result.archivedPath - ? { - archivedPath: result.archivedPath, - sourcePath: path.join(plan.archiveDirectory, `${plan.sessionId}.jsonl`), - } - : null; - return Object.assign({}, plan, { archivedTranscript }); + const generation = plan.snapshot.generation; + if (result.archive && !generation) { + throw new Error( + `Cannot archive SQLite transcript without a generation for ${plan.sessionId}`, + ); + } + const archivedTranscript = + result.archive && generation + ? { + generation, + sessionId: plan.sessionId, + archivedPath: path.join(plan.archiveDirectory, result.archive.archiveName), + sourcePath: path.join(plan.archiveDirectory, `${plan.sessionId}.jsonl`), + } + : null; + return Object.assign({}, plan, { archive: result.archive, archivedTranscript }); }); } +// Bulk cleanup plans retain encoded bytes until one atomic lifecycle commit. +// Bound that retained input while still allowing exceptionally large single transcripts. +const MAX_MATERIALIZED_ARCHIVE_BATCH_BYTES = 256 * 1024 * 1024; + // Multiple removed entries can point at one transcript session. If any owner // asked to keep an archive, the shared row gets exported once. function dedupeSqliteSessionStateDeletePlans( diff --git a/src/config/sessions/session-accessor.sqlite-archive.worker.test.ts b/src/config/sessions/session-accessor.sqlite-archive.worker.test.ts index 7b1d5467c2e4..f80f4554da4d 100644 --- a/src/config/sessions/session-accessor.sqlite-archive.worker.test.ts +++ b/src/config/sessions/session-accessor.sqlite-archive.worker.test.ts @@ -3,7 +3,7 @@ import { createHash, randomBytes } from "node:crypto"; import fs from "node:fs"; import os from "node:os"; import path from "node:path"; -import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { recordAcpParentStreamEvents } from "../../agents/subagents/spawn/acp-parent-stream-store.sqlite.js"; import { executeSqliteQuerySync, getNodeSqliteKysely } from "../../infra/kysely-sync.js"; import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js"; @@ -14,15 +14,15 @@ import { } from "../../state/openclaw-agent-db.js"; import { appendSqliteTrajectoryRuntimeEvents } from "../../trajectory/runtime-store.sqlite.js"; import type { TrajectoryEvent } from "../../trajectory/types.js"; -import { readSessionArchiveContentSync } from "./archive-compression.js"; +import { decodeSessionArchiveBytes, readSessionArchiveContentSync } from "./archive-compression.js"; import { + applySessionEntryLifecycleMutation, deleteSessionEntryLifecycle, loadSessionEntry, loadTranscriptEvents, replaceSessionEntry, } from "./session-accessor.js"; import { materializeSessionStateDeletePlans } from "./session-accessor.sqlite-archive.js"; -import { materializeTranscriptArchiveInWorker } from "./session-accessor.sqlite-archive.worker.js"; import { deleteMaterializedSessionStatePlans, planSessionStateDeleteIfUnreferenced, @@ -81,20 +81,24 @@ describe("SQLite transcript archive worker", () => { expect(heartbeatCount).toBeGreaterThan(5); expect(materialized).toHaveLength(1); - const archivedPath = materialized[0]?.archivedTranscript?.archivedPath; - expect(archivedPath).toBeTruthy(); + const archive = materialized[0]?.archive; + expect(archive).toBeTruthy(); + expect(fs.existsSync(materialized[0]?.archivedTranscript?.archivedPath ?? "")).toBe(false); const expectedContent = `${events.map((event) => JSON.stringify(event)).join("\n")}\n`; - const archivedContent = readSessionArchiveContentSync(archivedPath ?? ""); + const archivedContent = decodeSessionArchiveBytes( + archive?.bytes ?? new Uint8Array(), + archive?.encoding === "zstd", + ); expect(Buffer.byteLength(archivedContent)).toBe(Buffer.byteLength(expectedContent)); expect(sha256(archivedContent)).toBe(sha256(expectedContent)); - const archiveLines = readArchiveLines(archivedPath); + const archiveLines = archivedContent.trim().split("\n"); expect(archiveLines).toHaveLength(events.length); expect(archiveLines.map((line) => (JSON.parse(line) as { id: string }).id)).toEqual( events.map((event) => event.id), ); }); - it("publishes a durable archive before lifecycle deletion", async () => { + it("commits a canonical archive before publishing its derived file", async () => { const sessionId = "durable-delete-session"; const sessionKey = "agent:main:durable-delete"; await replaceSessionEntry( @@ -108,49 +112,6 @@ describe("SQLite transcript archive worker", () => { createTranscriptEvent(sessionId, "durable archive first"), ]); - const originalLinkSync = fs.linkSync; - const originalRenameSync = fs.renameSync; - const entryObservedDuringArchivePublish: boolean[] = []; - const observeArchivePublish = (archivePath: unknown) => { - if (String(archivePath).includes(`${sessionId}.jsonl.deleted.`)) { - entryObservedDuringArchivePublish.push( - loadSessionEntry({ sessionKey, storePath })?.sessionId === sessionId, - ); - } - }; - const openSpy = vi.spyOn(fs, "openSync"); - const fsyncSpy = vi.spyOn(fs, "fsyncSync"); - const linkSpy = vi.spyOn(fs, "linkSync").mockImplementation((...args) => { - observeArchivePublish(args[1]); - return originalLinkSync(...args); - }); - const renameSpy = vi.spyOn(fs, "renameSync").mockImplementation((...args) => { - observeArchivePublish(args[1]); - return originalRenameSync(...args); - }); - - let archivedPath: string | null = null; - try { - const database = openLifecycleTestDatabase(storePath); - const workerResult = materializeTranscriptArchiveInWorker( - planArchiveWorker(database, path.dirname(storePath), sessionId), - ); - archivedPath = workerResult.archivedPath; - expect(archivedPath).not.toBeNull(); - expect(entryObservedDuringArchivePublish).toEqual([true]); - const archiveTempOpenIndexes = openSpy.mock.calls.flatMap((args, index) => - String(args[0]).includes(`${sessionId}.jsonl.deleted.`) && args[1] === "wx" ? [index] : [], - ); - expect(archiveTempOpenIndexes).toHaveLength(1); - const archiveTempOpenIndex = archiveTempOpenIndexes[0] ?? -1; - expect(fsyncSpy).toHaveBeenCalledWith(openSpy.mock.results[archiveTempOpenIndex]?.value); - } finally { - renameSpy.mockRestore(); - linkSpy.mockRestore(); - fsyncSpy.mockRestore(); - openSpy.mockRestore(); - } - const result = await deleteSessionEntryLifecycle({ archiveTranscript: true, storePath, @@ -160,9 +121,104 @@ describe("SQLite transcript archive worker", () => { }, }); expect(result.deleted).toBe(true); - expect(result.archivedTranscripts.map((archive) => archive.archivedPath)).toEqual([ - archivedPath, + const archivedPath = result.archivedTranscripts[0]?.archivedPath; + expect(archivedPath).toBeTruthy(); + expect(readArchiveLines(archivedPath)).toEqual([ + JSON.stringify(createTranscriptEvent(sessionId, "durable archive first")), ]); + const database = openLifecycleTestDatabase(storePath); + expect( + database.db + .prepare( + "SELECT session_key, published_at FROM session_transcript_archives WHERE session_id = ?", + ) + .get(sessionId), + ).toMatchObject({ published_at: expect.any(Number), session_key: sessionKey }); + }); + + it("retains distinct transcript generations after a physical session id is restored", async () => { + const sessionId = "restored-archive-session"; + const sessionKey = "agent:main:restored-archive"; + const scope = { sessionId, sessionKey, storePath }; + await replaceSessionEntry(scope, { sessionId, updatedAt: 1 }); + await replaceTranscriptEvents(scope, [createTranscriptEvent(sessionId, "first generation")]); + const first = await deleteSessionEntryLifecycle({ + archiveTranscript: true, + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + }); + const firstArchive = first.archivedTranscripts[0]; + if (!firstArchive) { + throw new Error("expected first transcript archive"); + } + fs.rmSync(firstArchive.archivedPath); + openLifecycleTestDatabase(storePath) + .db.prepare( + `UPDATE session_transcript_archives + SET published_at = NULL + WHERE session_id = ? AND generation = ?`, + ) + .run(sessionId, firstArchive.generation); + + await replaceSessionEntry(scope, { sessionId, updatedAt: 2 }); + await replaceTranscriptEvents(scope, [createTranscriptEvent(sessionId, "second generation")]); + const second = await deleteSessionEntryLifecycle({ + archiveTranscript: true, + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + }); + + expect(second.archivedTranscripts).toHaveLength(1); + expect(second.archivedTranscripts[0]?.archivedPath).not.toBe(firstArchive.archivedPath); + expect(readArchiveLines(firstArchive.archivedPath)).toEqual([ + JSON.stringify(createTranscriptEvent(sessionId, "first generation")), + ]); + expect(readArchiveLines(second.archivedTranscripts[0]?.archivedPath)).toEqual([ + JSON.stringify(createTranscriptEvent(sessionId, "second generation")), + ]); + expect( + openLifecycleTestDatabase(storePath) + .db.prepare( + "SELECT generation FROM session_transcript_archives WHERE session_id = ? ORDER BY generation", + ) + .all(sessionId), + ).toHaveLength(2); + }); + + it("retries a pending archive export when deletion is already committed", async () => { + const sessionId = "retry-committed-delete"; + const sessionKey = "agent:main:retry-committed-delete"; + await replaceSessionEntry({ sessionKey, storePath }, { sessionId, updatedAt: Date.now() }); + await replaceTranscriptEvents({ sessionKey, sessionId, storePath }, [ + createTranscriptEvent(sessionId, "retry pending export"), + ]); + const first = await deleteSessionEntryLifecycle({ + archiveTranscript: true, + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + }); + const archivePath = first.archivedTranscripts[0]?.archivedPath; + fs.rmSync(archivePath ?? ""); + const database = openLifecycleTestDatabase(storePath); + database.db + .prepare("UPDATE session_transcript_archives SET published_at = NULL WHERE session_id = ?") + .run(sessionId); + + const retry = await deleteSessionEntryLifecycle({ + archiveTranscript: true, + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + }); + + expect(retry).toMatchObject({ archivedTranscripts: [], deleted: false }); + expect(readArchiveLines(archivePath)).toEqual([ + JSON.stringify(createTranscriptEvent(sessionId, "retry pending export")), + ]); + expect( + database.db + .prepare("SELECT published_at FROM session_transcript_archives WHERE session_id = ?") + .get(sessionId), + ).toMatchObject({ published_at: expect.any(Number) }); }); it("archives a logical agent transcript through the exact database's physical owner", async () => { @@ -206,17 +262,15 @@ describe("SQLite transcript archive worker", () => { expect(database.agentId).toBe("main"); expect(database.agentId).not.toBe(opsScope.agentId); - const plan = planArchiveWorker(database, tempDir, opsSessionId); - expect(plan).toMatchObject({ - agentId: database.agentId, - databasePath: database.path, - sessionId: opsSessionId, + const deleted = await deleteSessionEntryLifecycle({ + agentId: opsScope.agentId, + archiveTranscript: true, + storePath: sharedDatabasePath, + target: { canonicalKey: opsSessionKey, storeKeys: [opsSessionKey] }, }); - const materialized = await materializeSessionStateDeletePlans([plan]); - const archivedPath = materialized[0]?.archivedTranscript?.archivedPath; - expect(readArchiveLines(archivedPath ?? undefined)).toEqual([JSON.stringify(opsEvent)]); - - deleteMaterializedPlans(database, materialized, opsSessionKey); + expect(readArchiveLines(deleted.archivedTranscripts[0]?.archivedPath)).toEqual([ + JSON.stringify(opsEvent), + ]); await expect(loadTranscriptEvents(opsScope)).resolves.toEqual([]); await expect(loadTranscriptEvents(mainScope)).resolves.toEqual([mainEvent]); @@ -294,47 +348,6 @@ describe("SQLite transcript archive worker", () => { await expect(loadTranscriptEvents(scope)).resolves.toHaveLength(1); }); - it("recovers the lifecycle archive queue after a worker file failure", async () => { - const sessionId = "archive-file-failure-session"; - const scope = { - sessionKey: "agent:main:archive-file-failure", - sessionId, - storePath, - }; - await replaceTranscriptEvents(scope, [ - createTranscriptEvent(sessionId, "preserve after file failure"), - ]); - const blockedArchiveDirectory = path.join(tempDir, "archive-path-is-a-file"); - fs.writeFileSync(blockedArchiveDirectory, "not a directory", "utf8"); - const database = openLifecycleTestDatabase(storePath); - const plan = planArchiveWorker(database, blockedArchiveDirectory, sessionId); - const recoverySessionId = "archive-after-file-failure-session"; - const recoveryScope = { - sessionKey: "agent:main:archive-after-file-failure", - sessionId: recoverySessionId, - storePath, - }; - await replaceTranscriptEvents(recoveryScope, [ - createTranscriptEvent(recoverySessionId, "archive after queued failure"), - ]); - const recoveryPlan = planArchiveWorker(database, path.dirname(storePath), recoverySessionId); - - const failedArchive = materializeSessionStateDeletePlans([plan]); - const recoveredArchive = materializeSessionStateDeletePlans([recoveryPlan]); - - await expect(failedArchive).rejects.toThrow(); - await expect(recoveredArchive).resolves.toMatchObject([ - { - archivedTranscript: { - archivedPath: expect.stringContaining(`${recoverySessionId}.jsonl.deleted.`), - }, - sessionId: recoverySessionId, - }, - ]); - await expect(loadTranscriptEvents(scope)).resolves.toHaveLength(1); - expect(fs.readFileSync(blockedArchiveDirectory, "utf8")).toBe("not a directory"); - }); - it("preserves all lifecycle state when the archive worker rejects publication", async () => { const sessionId = "nested/archive-worker-lifecycle-failure"; const sessionKey = "agent:main:archive-worker-lifecycle-failure"; @@ -438,6 +451,26 @@ describe("SQLite transcript archive worker", () => { }); }); + it("captures archive materialization failure without deleting the requested entry", async () => { + const sessionId = "nested/captured-archive-failure"; + const sessionKey = "agent:main:captured-archive-failure"; + const scope = { sessionKey, sessionId, storePath }; + await replaceSessionEntry(scope, { sessionId, updatedAt: Date.now() }); + await replaceTranscriptEvents(scope, [createTranscriptEvent(sessionId, "retain on failure")]); + + const result = await applySessionEntryLifecycleMutation({ + captureArtifactCleanupError: true, + removals: [{ archiveRemovedTranscript: true, sessionKey }], + skipMaintenance: true, + storePath, + }); + + expect(result.removedEntries).toBe(0); + expect(result.artifactCleanupError).toBeInstanceOf(Error); + expect(loadSessionEntry(scope)).toMatchObject({ sessionId }); + await expect(loadTranscriptEvents(scope)).resolves.toHaveLength(1); + }); + it("keeps rows when a transcript changes after its archive snapshot", async () => { const sessionId = "stale-archive-snapshot-session"; const sessionKey = "agent:main:stale-archive-snapshot"; @@ -619,85 +652,6 @@ describe("SQLite transcript archive worker", () => { expect(rows).toHaveLength(1); }, ); - - it("does not reuse a matching in-flight temp file as an archive", async () => { - const sessionId = "in-flight-temp-archive-session"; - const line = createTranscriptEventLine(sessionId, "in-flight temp archive"); - await replaceTranscriptEvents( - { sessionKey: "agent:main:in-flight-temp-archive", sessionId, storePath }, - [JSON.parse(line) as TestTranscriptEvent], - ); - const archiveDirectory = path.dirname(storePath); - const tempPath = path.join( - archiveDirectory, - `${sessionId}.jsonl.deleted.2026-01-01T00-00-00.000Z.writer.tmp`, - ); - fs.mkdirSync(archiveDirectory, { recursive: true }); - fs.writeFileSync(tempPath, `${line}\n`, "utf8"); - - const database = openLifecycleTestDatabase(storePath); - const result = materializeTranscriptArchiveInWorker( - planArchiveWorker(database, archiveDirectory, sessionId), - ); - - expect(result.archivedPath).not.toBe(tempPath); - expect(fs.existsSync(tempPath)).toBe(true); - expect(readArchiveLines(result.archivedPath ?? undefined)).toEqual([line]); - }); - - it("reuses a matching archive before deleting entry rows", async () => { - const sessionId = "duplicate-archive-session"; - const sessionKey = "agent:main:duplicate-archive"; - await replaceSessionEntry({ sessionKey, storePath }, { sessionId, updatedAt: Date.now() }); - await replaceTranscriptEvents({ sessionKey, sessionId, storePath }, [ - createTranscriptEvent(sessionId, "reuse archive"), - ]); - const archivePath = path.join( - path.dirname(storePath), - `${sessionId}.jsonl.deleted.2026-01-01T00-00-00.000Z`, - ); - fs.mkdirSync(path.dirname(storePath), { recursive: true }); - fs.writeFileSync( - archivePath, - `${createTranscriptEventLine(sessionId, "reuse archive")}\n`, - "utf-8", - ); - - const originalReaddirSync = fs.readdirSync; - const entryObservedDuringDuplicateProbe: boolean[] = []; - const readdirSpy = vi.spyOn(fs, "readdirSync").mockImplementation((...args) => { - if (String(args[0]) === path.dirname(storePath)) { - entryObservedDuringDuplicateProbe.push( - loadSessionEntry({ sessionKey, storePath })?.sessionId === sessionId, - ); - } - return originalReaddirSync(...args); - }); - - try { - const database = openLifecycleTestDatabase(storePath); - const workerResult = materializeTranscriptArchiveInWorker( - planArchiveWorker(database, path.dirname(storePath), sessionId), - ); - expect(workerResult.archivedPath).toBe(archivePath); - expect(entryObservedDuringDuplicateProbe).toEqual([true]); - } finally { - readdirSpy.mockRestore(); - } - - const result = await deleteSessionEntryLifecycle({ - archiveTranscript: true, - storePath, - target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, - }); - expect(result.deleted).toBe(true); - expect(result.archivedTranscripts).toEqual([ - { - archivedPath: archivePath, - sourcePath: path.join(path.dirname(storePath), `${sessionId}.jsonl`), - }, - ]); - }); }); function createTranscriptEvent(sessionId: string, content: string): TestTranscriptEvent { diff --git a/src/config/sessions/session-accessor.sqlite-archive.worker.ts b/src/config/sessions/session-accessor.sqlite-archive.worker.ts index 604fb8a22a12..03312ab179ac 100644 --- a/src/config/sessions/session-accessor.sqlite-archive.worker.ts +++ b/src/config/sessions/session-accessor.sqlite-archive.worker.ts @@ -4,27 +4,71 @@ import { executeSqliteQuerySync, getNodeSqliteKysely } from "../../infra/kysely- import { withOpenClawAgentDatabaseReadOnly } from "../../state/openclaw-agent-db-readonly.js"; import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js"; import { - sqliteSessionStateDeleteSnapshotsEqual, - type SessionStateDeleteSnapshot, + encodeMaterializedSessionTranscriptArchive, + hashSessionArchiveBytes, + publishEncodedSessionTranscriptArchive, + type TranscriptArchivePublishPlan, + type TranscriptArchivePublishResult, + type TranscriptArchivePublishWorkerMessage, type TranscriptArchiveWorkerMessage, type TranscriptArchiveWorkerPlan, type TranscriptArchiveWorkerResult, - writeTranscriptArchive, } from "./session-accessor.sqlite-archive.js"; -import { readSessionStateDeleteSnapshot } from "./session-accessor.sqlite-delete-snapshot.js"; +import { + readSessionStateDeleteSnapshot, + sqliteSessionStateDeleteSnapshotsEqual, +} from "./session-accessor.sqlite-delete-snapshot.js"; +import type { SessionStateDeleteSnapshot } from "./session-accessor.sqlite-delete-snapshot.types.js"; import { serializeJsonlLines } from "./transcript-jsonl.js"; -type TranscriptArchiveDatabase = Pick; +type TranscriptArchiveDatabase = Pick< + OpenClawAgentKyselyDatabase, + "session_transcript_archives" | "transcript_events" +>; function isSqliteTranscriptArchiveWorkerData(value: unknown): boolean { return ( Boolean(value) && typeof value === "object" && !Array.isArray(value) && - (value as { type?: unknown }).type === "sqlite-transcript-archive-v1" + (value as { type?: unknown }).type === "sqlite-transcript-archive-v2" ); } +function parsePublishWorkerPlans(value: unknown): TranscriptArchivePublishPlan[] | undefined { + if (!value || typeof value !== "object" || Array.isArray(value)) { + return undefined; + } + const plans = (value as { plans?: unknown }).plans; + if (!Array.isArray(plans)) { + return undefined; + } + const parsed: TranscriptArchivePublishPlan[] = []; + for (const planValue of plans) { + if (!planValue || typeof planValue !== "object" || Array.isArray(planValue)) { + return undefined; + } + const plan = planValue as Record; + if ( + typeof plan.agentId !== "string" || + typeof plan.archiveDirectory !== "string" || + typeof plan.databasePath !== "string" || + typeof plan.generation !== "string" || + typeof plan.sessionId !== "string" + ) { + return undefined; + } + parsed.push({ + agentId: plan.agentId, + archiveDirectory: plan.archiveDirectory, + databasePath: plan.databasePath, + generation: plan.generation, + sessionId: plan.sessionId, + }); + } + return parsed; +} + function parseSessionStateDeleteSnapshot(value: unknown): SessionStateDeleteSnapshot | null { if (!value || typeof value !== "object" || Array.isArray(value)) { return null; @@ -34,6 +78,7 @@ function parseSessionStateDeleteSnapshot(value: unknown): SessionStateDeleteSnap typeof snapshot.acpParentStreamEventCount !== "number" || (snapshot.generation !== null && typeof snapshot.generation !== "string") || (snapshot.lastSeq !== null && typeof snapshot.lastSeq !== "number") || + (snapshot.sessionKey !== null && typeof snapshot.sessionKey !== "string") || (snapshot.sessionUpdatedAt !== null && typeof snapshot.sessionUpdatedAt !== "number") || (snapshot.trajectoryLastSeq !== null && typeof snapshot.trajectoryLastSeq !== "number") || (snapshot.transcriptUpdatedAt !== null && typeof snapshot.transcriptUpdatedAt !== "number") @@ -44,6 +89,7 @@ function parseSessionStateDeleteSnapshot(value: unknown): SessionStateDeleteSnap acpParentStreamEventCount: snapshot.acpParentStreamEventCount, generation: snapshot.generation, lastSeq: snapshot.lastSeq, + sessionKey: snapshot.sessionKey, sessionUpdatedAt: snapshot.sessionUpdatedAt, trajectoryLastSeq: snapshot.trajectoryLastSeq, transcriptUpdatedAt: snapshot.transcriptUpdatedAt, @@ -138,16 +184,64 @@ export function materializeTranscriptArchiveInWorker( ); } const { content } = opened.value; - const archivedPath = - content.length > 0 - ? writeTranscriptArchive({ + const generation = plan.snapshot.generation; + if (content.length > 0 && !generation) { + throw new Error(`Cannot archive SQLite transcript without a generation for ${plan.sessionId}`); + } + const archive = + content.length > 0 && generation + ? encodeMaterializedSessionTranscriptArchive({ archiveDirectory: plan.archiveDirectory, content, + generation, reason: plan.reason, sessionId: plan.sessionId, }) : null; - return { archivedPath, sessionId: plan.sessionId }; + return { archive, sessionId: plan.sessionId }; +} + +export function publishTranscriptArchiveInWorker( + plan: TranscriptArchivePublishPlan, +): TranscriptArchivePublishResult { + try { + const opened = withOpenClawAgentDatabaseReadOnly( + (database) => { + const db = getNodeSqliteKysely(database.db); + return executeSqliteQuerySync( + database.db, + db + .selectFrom("session_transcript_archives") + .select(["archive_blob", "archive_name", "archive_sha256"]) + .where("session_id", "=", plan.sessionId) + .where("generation", "=", plan.generation), + ).rows[0]; + }, + { agentId: plan.agentId, path: plan.databasePath }, + ); + if (!opened.found || !opened.value) { + throw new Error(`Canonical SQLite transcript archive is missing for ${plan.sessionId}`); + } + if (hashSessionArchiveBytes(opened.value.archive_blob) !== opened.value.archive_sha256) { + throw new Error(`Canonical SQLite transcript archive is corrupt for ${plan.sessionId}`); + } + return { + archivedPath: publishEncodedSessionTranscriptArchive({ + archiveDirectory: plan.archiveDirectory, + archiveName: opened.value.archive_name, + bytes: opened.value.archive_blob, + sha256: opened.value.archive_sha256, + }), + generation: plan.generation, + sessionId: plan.sessionId, + }; + } catch (error) { + return { + error: error instanceof Error ? error.message : String(error), + generation: plan.generation, + sessionId: plan.sessionId, + }; + } } function runWorkerPort( @@ -159,13 +253,33 @@ function runWorkerPort( port.close(); } +function runPublishWorkerPort( + port: NonNullable, + plans: readonly TranscriptArchivePublishPlan[], +): void { + const results = plans.map((plan) => publishTranscriptArchiveInWorker(plan)); + port.postMessage({ type: "published", results } satisfies TranscriptArchivePublishWorkerMessage); + port.close(); +} + if (isSqliteTranscriptArchiveWorkerData(workerData)) { if (!parentPort) { throw new Error("SQLite transcript archive worker requires a parent port"); } - const plans = parseWorkerPlans(workerData); - if (!plans) { - throw new Error("SQLite transcript archive worker requires valid worker data"); + const operation = (workerData as { operation?: unknown }).operation; + if (operation === "materialize") { + const plans = parseWorkerPlans(workerData); + if (!plans) { + throw new Error("SQLite transcript archive worker requires valid materialization data"); + } + runWorkerPort(parentPort, plans); + } else if (operation === "publish") { + const plans = parsePublishWorkerPlans(workerData); + if (!plans) { + throw new Error("SQLite transcript archive worker requires valid publication data"); + } + runPublishWorkerPort(parentPort, plans); + } else { + throw new Error("SQLite transcript archive worker requires a supported operation"); } - runWorkerPort(parentPort, plans); } diff --git a/src/config/sessions/session-accessor.sqlite-canonical-repair.ts b/src/config/sessions/session-accessor.sqlite-canonical-repair.ts index e694c4aea4df..913194881edc 100644 --- a/src/config/sessions/session-accessor.sqlite-canonical-repair.ts +++ b/src/config/sessions/session-accessor.sqlite-canonical-repair.ts @@ -5,6 +5,7 @@ import { } from "../../infra/kysely-sync.js"; import { openOpenClawAgentDatabase, + runOpenClawAgentWriteTransaction, type OpenClawAgentDatabase, } from "../../state/openclaw-agent-db.js"; import { listSqliteSessionEntriesWithCanonicalOwnerEvidence } from "./session-accessor.sqlite-canonical-inventory.js"; @@ -19,10 +20,12 @@ import { collectSessionStateIdsForEntry } from "./session-accessor.sqlite-refere import { getSessionKysely, resolveSqliteStoreScope, + runExclusiveSqliteSessionWrite, toDatabaseOptions, } from "./session-accessor.sqlite-scope.js"; import { bindSessionWindowEntryProjection } from "./session-accessor.sqlite-session-row.js"; import { parseSessionEntryJson } from "./session-accessor.sqlite-status.js"; +import { ensureTranscriptGenerationInTransaction } from "./session-accessor.sqlite-transcript-state.js"; import type { SessionEntryListScope } from "./session-accessor.types.js"; import { canonicalSessionKeyMigrationRequiredError } from "./session-canonical-key.js"; import { @@ -135,6 +138,55 @@ export function listSqliteSessionGenerationIdsForCanonicalRepair(params: { }); } +/** Doctor-only normalization of imported transcript rows before copy or archival. */ +export async function ensureSqliteTranscriptGenerationsForCanonicalRepair( + sources: readonly { + agentId: string; + entry: SessionEntry; + sessionKey: string; + storePath: string; + }[], +): Promise { + const byDatabase = new Map< + string, + { resolved: ReturnType; sources: typeof sources } + >(); + for (const source of sources) { + const resolved = resolveSqliteStoreScope(source.storePath, { agentId: source.agentId }); + const key = `${resolved.path ?? source.storePath}\0${resolved.databaseAgentId ?? resolved.agentId}`; + const grouped = byDatabase.get(key) ?? { resolved, sources: [] }; + byDatabase.set(key, { ...grouped, sources: [...grouped.sources, source] }); + } + for (const group of byDatabase.values()) { + await runExclusiveSqliteSessionWrite(group.resolved, async () => { + runOpenClawAgentWriteTransaction((database) => { + // Inventory and generation creation share one snapshot so copied rows and later archive + // plans observe the same immutable identity for each imported transcript. + const sessionIds = uniqueStrings([ + ...group.sources.flatMap((source) => [...collectSessionStateIdsForEntry(source.entry)]), + ...readSessionGenerationIdsForKeys( + database, + group.sources.map((source) => source.sessionKey), + { exactStoredKeys: true }, + ), + ]); + const db = getSessionKysely(database.db); + const eventSessionIds = executeSqliteQuerySync( + database.db, + db + .selectFrom("transcript_events") + .select("session_id") + .where("session_id", "in", sessionIds) + .groupBy("session_id"), + ).rows; + for (const row of eventSessionIds) { + ensureTranscriptGenerationInTransaction(database, row.session_id); + } + }, toDatabaseOptions(group.resolved)); + }); + } +} + /** Doctor-only same-store rewrite for delivery attribution owned by removed aliases. */ export function rehomeSqliteSessionDeliveryReferencesForCanonicalRepair( database: OpenClawAgentDatabase, diff --git a/src/config/sessions/session-accessor.sqlite-delete-snapshot.ts b/src/config/sessions/session-accessor.sqlite-delete-snapshot.ts index 0b098c7f5d2e..073f1403dd2a 100644 --- a/src/config/sessions/session-accessor.sqlite-delete-snapshot.ts +++ b/src/config/sessions/session-accessor.sqlite-delete-snapshot.ts @@ -1,6 +1,21 @@ import { executeSqliteQueryTakeFirstSync, getNodeSqliteKysely } from "../../infra/kysely-sync.js"; import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js"; -import type { SessionStateDeleteSnapshot } from "./session-accessor.sqlite-archive.js"; +import type { SessionStateDeleteSnapshot } from "./session-accessor.sqlite-delete-snapshot.types.js"; + +export function sqliteSessionStateDeleteSnapshotsEqual( + left: SessionStateDeleteSnapshot, + right: SessionStateDeleteSnapshot, +): boolean { + return ( + left.acpParentStreamEventCount === right.acpParentStreamEventCount && + left.generation === right.generation && + left.lastSeq === right.lastSeq && + left.sessionKey === right.sessionKey && + left.sessionUpdatedAt === right.sessionUpdatedAt && + left.trajectoryLastSeq === right.trajectoryLastSeq && + left.transcriptUpdatedAt === right.transcriptUpdatedAt + ); +} type SessionStateDeleteSnapshotDatabase = Pick< OpenClawAgentKyselyDatabase, @@ -25,7 +40,7 @@ export function readSessionStateDeleteSnapshot( database, db .selectFrom("session_windows") - .select(["transcript_updated_at", "updated_at"]) + .select(["session_key", "transcript_updated_at", "updated_at"]) .where("session_id", "=", sessionId), ); const rewriteWatermark = executeSqliteQueryTakeFirstSync( @@ -64,6 +79,7 @@ export function readSessionStateDeleteSnapshot( acpParentStreamEventCount: normalizeOptionalSqliteNumber(acpParentStream?.event_count) ?? 0, generation: rewriteWatermark?.generation ?? null, lastSeq: lastEvent?.seq ?? null, + sessionKey: window?.session_key ?? null, sessionUpdatedAt: window?.updated_at ?? null, trajectoryLastSeq: lastTrajectory?.seq ?? null, transcriptUpdatedAt: window?.transcript_updated_at ?? null, diff --git a/src/config/sessions/session-accessor.sqlite-delete-snapshot.types.ts b/src/config/sessions/session-accessor.sqlite-delete-snapshot.types.ts new file mode 100644 index 000000000000..7c6c7c78d4d0 --- /dev/null +++ b/src/config/sessions/session-accessor.sqlite-delete-snapshot.types.ts @@ -0,0 +1,9 @@ +export type SessionStateDeleteSnapshot = { + acpParentStreamEventCount: number; + generation: string | null; + lastSeq: number | null; + sessionKey: string | null; + sessionUpdatedAt: number | null; + trajectoryLastSeq: number | null; + transcriptUpdatedAt: number | null; +}; diff --git a/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts b/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts index 4bbe0e92f2b7..13d70bce2eed 100644 --- a/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts +++ b/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts @@ -8,17 +8,20 @@ import { isIncognitoOpenClawAgentDatabase, type OpenClawAgentDatabase, } from "../../state/openclaw-agent-db.js"; -import { - sqliteSessionStateDeleteSnapshotsEqual, - type MaterializedSessionStateDeletePlan, - type SessionStateDeletePlan, +import { persistSessionTranscriptArchive } from "./session-accessor.sqlite-archive-store.js"; +import type { + MaterializedSessionStateDeletePlan, + SessionStateDeletePlan, } from "./session-accessor.sqlite-archive.js"; import type { SessionEntryLifecycleRemoval, SessionEntryLifecycleUpsert, SessionLifecycleArchivedTranscript, } from "./session-accessor.sqlite-contract.js"; -import { readSessionStateDeleteSnapshot } from "./session-accessor.sqlite-delete-snapshot.js"; +import { + readSessionStateDeleteSnapshot, + sqliteSessionStateDeleteSnapshotsEqual, +} from "./session-accessor.sqlite-delete-snapshot.js"; import { deleteSessionEntryRows, readExactSessionEntryJsonForCanonicalRepair, @@ -244,6 +247,9 @@ export function deleteMaterializedSessionStatePlans( if (!sqliteSessionStateDeleteSnapshotsEqual(currentSnapshot, plan.snapshot)) { throw new Error(`SQLite session state changed before deletion for ${plan.sessionId}`); } + if (plan.archive) { + persistSessionTranscriptArchive(database, plan); + } deleteSqliteSessionStateRows(database, plan.sessionId); if (plan.snapshot.lastSeq !== null && plan.archivedTranscript) { archivedTranscripts.push(plan.archivedTranscript); @@ -333,6 +339,20 @@ export async function projectSessionEntryLifecycleMutation( if (!shouldRemoveSessionEntry(entry, removal)) { continue; } + if (removal.expectedTranscriptSnapshot) { + const sessionId = entry.sessionId; + if ( + !sessionId || + !sqliteSessionStateDeleteSnapshotsEqual( + readSessionStateDeleteSnapshot(database.db, sessionId), + removal.expectedTranscriptSnapshot, + ) + ) { + // Classification happens before the lifecycle writer lane. A stale fact + // must become a no-op so newly live state is never archived and deleted. + continue; + } + } projectedRemovals.push({ expectedEntry: cloneSessionEntry(entry), removal, @@ -403,6 +423,21 @@ export async function projectSessionEntryLifecycleMutation( referencedSessionIds, }), ); + const observedSnapshotsBySessionId = new Map( + projectedRemovals.flatMap(({ expectedEntry, removal }) => + expectedEntry.sessionId && removal.expectedTranscriptSnapshot + ? [[expectedEntry.sessionId, removal.expectedTranscriptSnapshot] as const] + : [], + ), + ); + for (const plan of deletePlans) { + const observedSnapshot = observedSnapshotsBySessionId.get(plan.sessionId); + if (observedSnapshot) { + // Keep the delete plan bound to classification, even if another process + // changes the transcript after the initial projection comparison. + plan.snapshot = observedSnapshot; + } + } const plannedIds = new Set(deletePlans.map((plan) => plan.sessionId)); for (const sessionId of readSessionGenerationIdsForKeys(database, removedKeysToArchive)) { if (plannedIds.has(sessionId)) { diff --git a/src/config/sessions/session-accessor.sqlite-lifecycle.ts b/src/config/sessions/session-accessor.sqlite-lifecycle.ts index 5583c5d72dd2..46ff5259bd55 100644 --- a/src/config/sessions/session-accessor.sqlite-lifecycle.ts +++ b/src/config/sessions/session-accessor.sqlite-lifecycle.ts @@ -16,6 +16,7 @@ import { type OpenClawAgentDatabase, } from "../../state/openclaw-agent-db.js"; import type { ResetSessionEntryLifecycleMutation } from "./session-accessor.lifecycle-types.js"; +import { publishSessionStateArchives } from "./session-accessor.sqlite-archive-store.js"; import { materializeSessionStateDeletePlans } from "./session-accessor.sqlite-archive.js"; import type { SessionLifecycleArchivedTranscript, @@ -139,7 +140,7 @@ export async function cleanupSessionLifecycleArtifactsCore( }); }); const materializedPlans = await materializeSessionStateDeletePlans(cleanupPlan.deletePlans); - return await runExclusiveSqliteSessionWrite(resolved, async () => { + const committed = await runExclusiveSqliteSessionWrite(resolved, async () => { let removedEntries = 0; let archivedTranscripts: SessionLifecycleArchivedTranscript[] = []; runOpenClawAgentWriteTransaction((transactionDb) => { @@ -155,9 +156,17 @@ export async function cleanupSessionLifecycleArtifactsCore( emitCommittedSessionEntryRemovals(cleanupPlan.entries); return { removedEntries, - archivedTranscriptArtifacts: archivedTranscripts.length, + archivedTranscripts, }; }); + const archivedTranscripts = await publishSessionStateArchives( + resolved, + committed.archivedTranscripts, + ); + return { + removedEntries: committed.removedEntries, + archivedTranscriptArtifacts: archivedTranscripts.length, + }; } /** Resets one persisted session entry using SQLite session rows. */ @@ -371,6 +380,7 @@ async function deleteSqliteSessionEntryLifecycleLocked( return { archiveDirectory, current, entryPlans, historicalGenerationIds, targetSnapshot }; }); if (!prepared) { + await publishSessionStateArchives(resolved, []); return { archivedTranscripts: [], deleted: false }; } @@ -444,8 +454,9 @@ async function deleteSqliteSessionEntryLifecycleLocked( // 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). - emitArchivedTranscriptUpdates(archivedGeneration); - historicalArchivedTranscripts.push(...archivedGeneration); + const publishedGeneration = await publishSessionStateArchives(resolved, archivedGeneration); + emitArchivedTranscriptUpdates(publishedGeneration); + historicalArchivedTranscripts.push(...publishedGeneration); } // Archive materialization is the expensive phase. It must run between short @@ -505,6 +516,10 @@ async function deleteSqliteSessionEntryLifecycleLocked( }, }); } + result.archivedTranscripts = await publishSessionStateArchives( + resolved, + result.archivedTranscripts, + ); emitArchivedTranscriptUpdates(result.archivedTranscripts); // Historical generations were emitted per commit above; merge them into // the result after the final emit so callers still see every archive. diff --git a/src/config/sessions/session-accessor.sqlite-maintenance.ts b/src/config/sessions/session-accessor.sqlite-maintenance.ts index 9790e98d975e..88b275df732c 100644 --- a/src/config/sessions/session-accessor.sqlite-maintenance.ts +++ b/src/config/sessions/session-accessor.sqlite-maintenance.ts @@ -5,6 +5,7 @@ import { runOpenClawAgentWriteTransaction, type OpenClawAgentDatabase, } from "../../state/openclaw-agent-db.js"; +import { publishSessionStateArchives } from "./session-accessor.sqlite-archive-store.js"; import { materializeSessionStateDeletePlans, type SessionStateDeletePlan, @@ -292,7 +293,7 @@ async function finalizeSqliteSessionEntryMaintenancePlansWithCommit( return committed; }); emitCommittedSessionEntryRemovals(entryRemovals); - return archivedTranscripts; + return await publishSessionStateArchives(scope, archivedTranscripts); } catch (error) { getChildLogger({ subsystem: "session-sqlite" }).warn( "SQLite session maintenance cleanup failed", diff --git a/src/config/sessions/session-accessor.sqlite-projection.ts b/src/config/sessions/session-accessor.sqlite-projection.ts index 9f2dc216e8f1..56a2d48910e0 100644 --- a/src/config/sessions/session-accessor.sqlite-projection.ts +++ b/src/config/sessions/session-accessor.sqlite-projection.ts @@ -11,6 +11,10 @@ import { type OpenClawAgentDatabase, } from "../../state/openclaw-agent-db.js"; import type { SessionArchivedTranscriptCleanupRule } from "./session-accessor.lifecycle-types.js"; +import { + prunePublishedSessionArchivesByRetention, + publishSessionStateArchives, +} from "./session-accessor.sqlite-archive-store.js"; import { materializeSessionStateDeletePlans, type MaterializedSessionStateDeletePlan, @@ -260,9 +264,11 @@ export async function applySessionEntryLifecycleMutation(params: { }); }); let materializedRemovalPlans: MaterializedSessionStateDeletePlan[] = []; + let removalArchiveMaterializationFailed = false; try { materializedRemovalPlans = await materializeSessionStateDeletePlans(projected.deletePlans); } catch (error) { + removalArchiveMaterializationFailed = true; captureArtifactCleanupError(error); } const committed = await runExclusiveSqliteSessionWrite(resolved, async () => { @@ -272,6 +278,12 @@ export async function applySessionEntryLifecycleMutation(params: { runOpenClawAgentWriteTransaction((transactionDb) => { params.beforeCommitInTransaction?.(); const validatedRemovals = projected.removals.filter((removal) => { + if ( + removalArchiveMaterializationFailed && + removal.removal.archiveRemovedTranscript === true + ) { + return false; + } const entry = readProjectedRemovalEntry( transactionDb, removal, @@ -425,7 +437,16 @@ export async function applySessionEntryLifecycleMutation(params: { resolved, committed.maintenancePlans, ); - const archivedTranscripts = [...committed.archivedTranscripts, ...maintenanceArchivedTranscripts]; + let publishedRemovalTranscripts: SessionLifecycleArchivedTranscript[] = []; + try { + publishedRemovalTranscripts = await publishSessionStateArchives( + resolved, + committed.archivedTranscripts, + ); + } catch (error) { + captureArtifactCleanupError(error); + } + const archivedTranscripts = [...publishedRemovalTranscripts, ...maintenanceArchivedTranscripts]; const afterCount = readSessionEntryCount(openOpenClawAgentDatabase(toDatabaseOptions(resolved))); emitArchivedTranscriptUpdates(archivedTranscripts); const archivedTranscriptDirectories = uniqueStrings( @@ -439,6 +460,11 @@ export async function applySessionEntryLifecycleMutation(params: { rules: params.cleanupArchivedTranscripts.rules, nowMs: params.cleanupArchivedTranscripts.nowMs, }); + await prunePublishedSessionArchivesByRetention({ + scope: resolved, + rules: params.cleanupArchivedTranscripts.rules, + nowMs: params.cleanupArchivedTranscripts.nowMs, + }); } catch (error) { captureArtifactCleanupError(error); } @@ -540,7 +566,7 @@ export async function purgeDeletedAgentSessionEntries( return { archivedTranscripts, maintenancePlans, removedSessionKeys }; }); const archivedTranscripts = [ - ...committed.archivedTranscripts, + ...(await publishSessionStateArchives(resolved, committed.archivedTranscripts)), ...(await finalizeSessionEntryMaintenancePlansAfterWriterReleaseBestEffort( resolved, committed.maintenancePlans, diff --git a/src/config/sessions/session-accessor.sqlite-read.ts b/src/config/sessions/session-accessor.sqlite-read.ts index 01e56ecc6562..6d9076a4c770 100644 --- a/src/config/sessions/session-accessor.sqlite-read.ts +++ b/src/config/sessions/session-accessor.sqlite-read.ts @@ -19,6 +19,8 @@ import type { SessionTranscriptStats, TranscriptEvent, } from "./session-accessor.sqlite-contract.js"; +import { readSessionStateDeleteSnapshot } from "./session-accessor.sqlite-delete-snapshot.js"; +import type { SessionStateDeleteSnapshot } from "./session-accessor.sqlite-delete-snapshot.types.js"; import { coerceSqliteNumber } from "./session-accessor.sqlite-normalize.js"; import { getSessionKysely, @@ -60,6 +62,26 @@ export function loadTranscriptEventsSync(scope: SessionTranscriptReadScope): Tra ); } +/** Reads a complete transcript and its lifecycle snapshot from one SQLite read transaction. */ +export function inspectTranscriptEventsSync(scope: SessionTranscriptReadScope): { + events: TranscriptEvent[]; + snapshot: SessionStateDeleteSnapshot; +} { + const resolved = resolveSqliteTranscriptReadScope(scope); + const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); + return runSqliteDeferredTransactionSync( + database.db, + () => ({ + events: readTranscriptSnapshot(database, resolved.sessionId).events, + snapshot: readSessionStateDeleteSnapshot(database.db, resolved.sessionId), + }), + { + databaseLabel: database.path, + operationLabel: "session transcript inspection", + }, + ); +} + /** Loads only the first transcript row for header metadata hot paths. */ export function loadTranscriptHeaderSync(scope: SessionTranscriptReadScope): unknown { const resolved = resolveSqliteTranscriptReadScope(scope); diff --git a/src/config/sessions/session-accessor.sqlite-scope.ts b/src/config/sessions/session-accessor.sqlite-scope.ts index 0236d6b0a58c..47e9ccff58dc 100644 --- a/src/config/sessions/session-accessor.sqlite-scope.ts +++ b/src/config/sessions/session-accessor.sqlite-scope.ts @@ -39,6 +39,7 @@ type SessionSqliteDatabase = Pick< | "session_members" | "session_nodes" | "session_suggestions" + | "session_transcript_archives" | "session_transcript_active_events" | "session_transcript_index_state" | "session_windows" diff --git a/src/config/sessions/session-accessor.transcript.ts b/src/config/sessions/session-accessor.transcript.ts index ff12a477b74c..ae0d3383b837 100644 --- a/src/config/sessions/session-accessor.transcript.ts +++ b/src/config/sessions/session-accessor.transcript.ts @@ -4,6 +4,7 @@ import { resolveSessionKeyBySessionId as resolveTranscriptSessionKeyBySessionId import { publishTranscriptUpdate } from "./session-accessor.sqlite-events.js"; import { findTranscriptEvent, + inspectTranscriptEventsSync, loadLatestAssistantText as readLatestTranscriptAssistantText, loadTranscriptEventRowsAfterSeqSync, loadTranscriptEvents, @@ -42,6 +43,7 @@ export { appendTranscriptMessage, appendTranscriptMessageSync, findTranscriptEvent, + inspectTranscriptEventsSync, loadTranscriptEventRowsAfterSeqSync, loadTranscriptEvents, loadTranscriptEventsSync, diff --git a/src/config/sessions/session-accessor.ts b/src/config/sessions/session-accessor.ts index 895886699390..cddae52f5274 100644 --- a/src/config/sessions/session-accessor.ts +++ b/src/config/sessions/session-accessor.ts @@ -129,6 +129,7 @@ export { countSessionEntryRowsReadOnly, ensureSessionEntrySync, copySessionOwnedStateForCanonicalRepair, + ensureTranscriptGenerationsForCanonicalRepair, hasSessionEntriesByStatusReadOnly, listSessionGenerationIdsForCanonicalRepair, clearPluginOwnedSessionState, @@ -211,6 +212,7 @@ export { appendTranscriptMessage, appendTranscriptMessageSync, findTranscriptEvent, + inspectTranscriptEventsSync, loadTranscriptEventRowsAfterSeqSync, loadTranscriptEvents, loadTranscriptEventsSync, diff --git a/src/config/sessions/session-history-eviction.test.ts b/src/config/sessions/session-history-eviction.test.ts index 8b58811a91f8..b5bcd1a98e34 100644 --- a/src/config/sessions/session-history-eviction.test.ts +++ b/src/config/sessions/session-history-eviction.test.ts @@ -1,3 +1,4 @@ +import { randomBytes } from "node:crypto"; import fs from "node:fs"; import path from "node:path"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; @@ -16,6 +17,7 @@ import type { TrajectoryEvent } from "../../trajectory/types.js"; import { measureSessionPhysicalDiskUsage } from "./disk-budget.js"; import { appendTranscriptMessage, + deleteSessionEntryLifecycle, replaceSessionEntry, resetSessionEntryLifecycle, } from "./session-accessor.js"; @@ -84,6 +86,9 @@ describe("SQLite historical session disk budget", () => { expect(result?.removedEntries).toBe(1); expect(result?.totalBytesAfter).toBeLessThanOrEqual(before.totalBytes - 1); + expect(result?.totalBytesAfter).toBe( + (await measureSessionPhysicalDiskUsage(storePath)).totalBytes, + ); expect(sessionExists("oldest-history")).toBe(false); expect(sessionExists("newer-history")).toBe(true); expect(sessionExists("live-history")).toBe(true); @@ -91,6 +96,36 @@ describe("SQLite historical session disk budget", () => { expect(readArchiveNames("newer-history")).toHaveLength(0); }); + it("remeasures incompressible archive publication before declaring high water", async () => { + const sessionId = "incompressible-history"; + const sessionKey = "agent:main:incompressible-history"; + await createHistoricalTranscript({ + content: randomBytes(192 * 1024).toString("base64"), + nextSessionId: "incompressible-live", + sessionId, + sessionKey, + updatedAt: 1, + }); + settlePhysicalUsage(); + const before = await measureSessionPhysicalDiskUsage(storePath); + const highWaterBytes = before.totalBytes - 1; + + const result = await enforceSqliteSessionHistoryDiskBudget({ + storePath, + mode: "enforce", + maintenance: { + maxDiskBytes: highWaterBytes, + highWaterBytes, + }, + }); + const actualAfter = await measureSessionPhysicalDiskUsage(storePath); + + expect(result?.removedEntries).toBe(1); + expect(result?.totalBytesAfter).toBe(actualAfter.totalBytes); + expect(actualAfter.totalBytes).toBeLessThanOrEqual(highWaterBytes); + expect(sessionExists(sessionId)).toBe(false); + }); + it("removes counted archives before evicting searchable history", async () => { await createHistoricalTranscript({ content: "keep searchable history", @@ -121,6 +156,92 @@ describe("SQLite historical session disk budget", () => { expect(sessionExists("archive-history")).toBe(true); }); + it("prunes the canonical archive row and its derived file before searchable history", async () => { + const archivedSessionId = "canonical-archive"; + const archivedSessionKey = "agent:main:canonical-archive"; + await replaceSessionEntry( + { sessionKey: archivedSessionKey, storePath }, + { sessionId: archivedSessionId, updatedAt: 1 }, + ); + await appendTranscriptMessage( + { sessionId: archivedSessionId, sessionKey: archivedSessionKey, storePath }, + { message: { role: "user", content: "canonical archive pressure" } }, + ); + const deleted = await deleteSessionEntryLifecycle({ + archiveTranscript: true, + storePath, + target: { canonicalKey: archivedSessionKey, storeKeys: [archivedSessionKey] }, + }); + const archivePath = deleted.archivedTranscripts[0]?.archivedPath; + expect(archivePath).toBeTruthy(); + + await createHistoricalTranscript({ + content: "keep searchable history", + nextSessionId: "canonical-live", + sessionId: "canonical-history", + sessionKey: "agent:main:canonical-pressure", + updatedAt: 2, + }); + settlePhysicalUsage(); + const before = await measureSessionPhysicalDiskUsage(storePath); + + const result = await enforceSqliteSessionHistoryDiskBudget({ + storePath, + mode: "enforce", + maintenance: { + maxDiskBytes: before.totalBytes - 1, + highWaterBytes: before.totalBytes - 1, + }, + }); + + expect(result).toMatchObject({ removedEntries: 0, removedFiles: 1 }); + expect(fs.existsSync(archivePath ?? "")).toBe(false); + expect( + database() + .db.prepare("SELECT 1 FROM session_transcript_archives WHERE session_id = ?") + .get(archivedSessionId), + ).toBeUndefined(); + expect(sessionExists("canonical-history")).toBe(true); + }); + + it("never prunes an unpublished canonical archive under disk pressure", async () => { + const sessionId = "pending-pressure"; + const sessionKey = "agent:main:pending-pressure"; + await replaceSessionEntry({ sessionKey, storePath }, { sessionId, updatedAt: Date.now() }); + await appendTranscriptMessage( + { sessionId, sessionKey, storePath }, + { message: { role: "user", content: "sole crash-recovery copy" } }, + ); + const deleted = await deleteSessionEntryLifecycle({ + archiveTranscript: true, + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + }); + const pendingArchivePath = deleted.archivedTranscripts[0]?.archivedPath; + database() + .db.prepare("UPDATE session_transcript_archives SET published_at = NULL WHERE session_id = ?") + .run(sessionId); + settlePhysicalUsage(); + const before = await measureSessionPhysicalDiskUsage(storePath); + + const result = await enforceSqliteSessionHistoryDiskBudget({ + storePath, + mode: "enforce", + maintenance: { + maxDiskBytes: before.totalBytes - 1, + highWaterBytes: before.totalBytes - 1, + }, + }); + + expect(result).toMatchObject({ removedEntries: 0, removedFiles: 0 }); + expect( + database() + .db.prepare("SELECT published_at FROM session_transcript_archives WHERE session_id = ?") + .get(sessionId), + ).toEqual({ published_at: null }); + expect(fs.existsSync(pendingArchivePath ?? "")).toBe(true); + }); + it("excludes entry, route, and admitted ids while evicting trajectory-only history", async () => { const sessionKey = "agent:main:history-protection"; await replaceSessionEntry( diff --git a/src/config/sessions/session-history-eviction.ts b/src/config/sessions/session-history-eviction.ts index 799960f9a7cd..3f30778c513f 100644 --- a/src/config/sessions/session-history-eviction.ts +++ b/src/config/sessions/session-history-eviction.ts @@ -1,3 +1,5 @@ +import fs from "node:fs"; +import path from "node:path"; import { executeSqliteQuerySync } from "../../infra/kysely-sync.js"; import { collectActiveSessionWorkAdmissionIdentities, @@ -17,6 +19,7 @@ import { type SessionDiskBudgetSweepResult, type SessionPhysicalDiskUsage, } from "./disk-budget.js"; +import { publishSessionStateArchives } from "./session-accessor.sqlite-archive-store.js"; import { materializeSessionStateDeletePlans } from "./session-accessor.sqlite-archive.js"; import { emitArchivedTranscriptUpdates } from "./session-accessor.sqlite-events.js"; import { @@ -85,15 +88,18 @@ export async function inspectSqliteSessionHistoryDiskBudget( // 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)); + if ( + hasCanonicalSessionTranscriptArchives(database) || + (await hasRetainedSessionTranscriptArchives(params.storePath)) + ) { + return { diskBudget, wouldMutate: true }; + } const candidates = readHistoricalSessionIds({ database, protectedSessionIds: collectProtectedHistoricalSessionIds({ @@ -195,6 +201,144 @@ function reclaimSqliteFreePages(database: OpenClawAgentDatabase): void { database.walMaintenance.checkpoint(); } +function hasCanonicalSessionTranscriptArchives(database: OpenClawAgentDatabase): boolean { + const db = getSessionKysely(database.db); + const table = executeSqliteQuerySync( + database.db, + db + .selectFrom("sqlite_schema") + .select("name") + .where("type", "=", "table") + .where("name", "=", "session_transcript_archives"), + ).rows[0]; + if (!table) { + return false; + } + return ( + executeSqliteQuerySync( + database.db, + db + .selectFrom("session_transcript_archives") + .select("session_id") + .where("published_at", "is not", null) + .limit(1), + ).rows.length > 0 + ); +} + +function readUnpublishedSessionTranscriptArchiveNames( + database: OpenClawAgentDatabase, +): Set { + const db = getSessionKysely(database.db); + const table = executeSqliteQuerySync( + database.db, + db + .selectFrom("sqlite_schema") + .select("name") + .where("type", "=", "table") + .where("name", "=", "session_transcript_archives"), + ).rows[0]; + if (!table) { + return new Set(); + } + return new Set( + executeSqliteQuerySync( + database.db, + db + .selectFrom("session_transcript_archives") + .select("archive_name") + .where("published_at", "is", null), + ).rows.map((row) => row.archive_name), + ); +} + +async function pruneCanonicalSessionTranscriptArchivesToHighWater(params: { + archiveDirectory: string; + database: OpenClawAgentDatabase; + highWaterBytes: number; + storePath: string; +}): Promise<{ removedFiles: number; usage: SessionPhysicalDiskUsage }> { + let usage = await measureSessionPhysicalDiskUsage(params.storePath); + let removedFiles = 0; + while (usage.totalBytes > params.highWaterBytes) { + const db = getSessionKysely(params.database.db); + const row = executeSqliteQuerySync( + params.database.db, + db + .selectFrom("session_transcript_archives") + .select(["archive_name", "generation", "session_id"]) + .where("published_at", "is not", null) + .orderBy("created_at", "asc") + .orderBy("session_id", "asc") + .orderBy("generation", "asc") + .limit(1), + ).rows[0]; + if (!row) { + break; + } + const archivePath = path.resolve(params.archiveDirectory, row.archive_name); + if ( + path.dirname(archivePath) !== path.resolve(params.archiveDirectory) || + path.basename(archivePath) !== row.archive_name + ) { + throw new Error(`Invalid canonical session archive name for ${row.session_id}`); + } + try { + await fs.promises.rm(archivePath); + removedFiles += 1; + } catch (error) { + if ((error as NodeJS.ErrnoException).code !== "ENOENT") { + // The database is the recovery copy. Retain it unless its derived file + // is gone, otherwise retention could leave an undeletable orphan. + break; + } + } + runOpenClawAgentWriteTransaction( + (transactionDb) => { + const transactionKysely = getSessionKysely(transactionDb.db); + executeSqliteQuerySync( + transactionDb.db, + transactionKysely + .deleteFrom("session_transcript_archives") + .where("session_id", "=", row.session_id) + .where("generation", "=", row.generation), + ); + }, + { agentId: params.database.agentId, path: params.database.path }, + ); + reclaimSqliteFreePages(params.database); + usage = await measureSessionPhysicalDiskUsage(params.storePath); + } + return { removedFiles, usage }; +} + +async function pruneAllSessionTranscriptArchivesToHighWater(params: { + archiveDirectory: string; + database: OpenClawAgentDatabase; + highWaterBytes: number; + storePath: string; +}): Promise<{ removedFiles: number; usage: SessionPhysicalDiskUsage }> { + let canonical = { + removedFiles: 0, + usage: await measureSessionPhysicalDiskUsage(params.storePath), + }; + if (hasCanonicalSessionTranscriptArchives(params.database)) { + canonical = await pruneCanonicalSessionTranscriptArchivesToHighWater(params); + } + if (canonical.usage.totalBytes <= params.highWaterBytes) { + return canonical; + } + const legacy = await pruneSessionTranscriptArchivesToHighWater({ + excludeNames: readUnpublishedSessionTranscriptArchiveNames(params.database), + highWaterBytes: params.highWaterBytes, + storePath: params.storePath, + }); + return { + removedFiles: canonical.removedFiles + legacy.removedFiles, + usage: legacy.usage, + }; +} + 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 @@ -323,7 +467,9 @@ async function enforceSessionHistoryMaintenanceSerialized( // Archive pruning shares the SQLite writer queue so a concurrent // reset/delete cannot race its archive file against the unlink pass. const archiveSweep = await runExclusiveSqliteSessionWrite(resolved, async () => - pruneSessionTranscriptArchivesToHighWater({ + pruneAllSessionTranscriptArchivesToHighWater({ + archiveDirectory, + database, highWaterBytes, storePath: params.storePath, }), @@ -412,16 +558,23 @@ async function enforceSessionHistoryMaintenanceSerialized( } return { archivedTranscripts: committedArchives, - usage: await measureSessionPhysicalDiskUsage(params.storePath), }; }, }); if (!eviction) { continue; } + // The lifecycle and SQLite writer lanes are both released before file I/O; + // publication reacquires the writer only for its short status commit. + const publishedArchives = await publishSessionStateArchives( + resolved, + eviction.archivedTranscripts, + ); removedEntries += 1; - emitArchivedTranscriptUpdates(eviction.archivedTranscripts); - usage = eviction.usage; + emitArchivedTranscriptUpdates(publishedArchives); + // Publication adds both the derived file and SQLite status WAL after the + // deletion measurement. Re-read physical usage before declaring high water. + usage = await measureSessionPhysicalDiskUsage(params.storePath); 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 @@ -429,7 +582,9 @@ async function enforceSessionHistoryMaintenanceSerialized( // 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({ + pruneAllSessionTranscriptArchivesToHighWater({ + archiveDirectory, + database, highWaterBytes, storePath: params.storePath, }), @@ -443,7 +598,9 @@ async function enforceSessionHistoryMaintenanceSerialized( // 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({ + pruneAllSessionTranscriptArchivesToHighWater({ + archiveDirectory, + database, highWaterBytes, storePath: params.storePath, }), diff --git a/src/gateway/server.sessions.delete-lifecycle.test.ts b/src/gateway/server.sessions.delete-lifecycle.test.ts index 8b7fe8d0692a..e8dc6239f246 100644 --- a/src/gateway/server.sessions.delete-lifecycle.test.ts +++ b/src/gateway/server.sessions.delete-lifecycle.test.ts @@ -572,14 +572,33 @@ test("sessions.delete serializes a patch behind asynchronous runtime cleanup", a }); await runtimeCleanupStarted; let patchSettled = false; - const patch = directSessionReq("sessions.patch", { - key: sessionKey, - label: "updated during cleanup", - }).then((result) => { + let markPatchPreflight = () => {}; + const patchPreflight = new Promise((resolve) => { + markPatchPreflight = resolve; + }); + const patch = directSessionReq( + "sessions.patch", + { + key: sessionKey, + label: "updated during cleanup", + }, + { + context: { + workerSessionPlacementService: { + getMany(sessionIds: readonly string[]) { + if (sessionIds.includes(sessionId)) { + markPatchPreflight(); + } + return new Map(); + }, + }, + }, + }, + ).then((result) => { patchSettled = true; return result; }); - await Promise.resolve(); + await patchPreflight; expect(patchSettled).toBe(false); releaseRuntimeCleanup(); diff --git a/src/infra/state-migrations.media-persistence.historical-schema.test-support.ts b/src/infra/state-migrations.media-persistence.historical-schema.test-support.ts index c95faf78842d..5db6ad7cfb08 100644 --- a/src/infra/state-migrations.media-persistence.historical-schema.test-support.ts +++ b/src/infra/state-migrations.media-persistence.historical-schema.test-support.ts @@ -59,6 +59,11 @@ export function historicalV15AgentSchemaSql(): string { "CREATE TABLE IF NOT EXISTS memory_index_chunk_recall_metadata (", "CREATE TABLE IF NOT EXISTS memory_embedding_cache (", ); + sql = removeSchemaRange( + sql, + "-- Canonical cold-tier owner for reclaimed transcript generations.", + "CREATE TABLE IF NOT EXISTS transcript_rewrite_watermarks (", + ); return removeSchemaRange( sql, "CREATE TABLE IF NOT EXISTS standing_intents (", diff --git a/src/state/openclaw-agent-db-schema-helpers.ts b/src/state/openclaw-agent-db-schema-helpers.ts index c7aa382de0f0..dbd60736060a 100644 --- a/src/state/openclaw-agent-db-schema-helpers.ts +++ b/src/state/openclaw-agent-db-schema-helpers.ts @@ -31,6 +31,7 @@ import { AGENT_V14_CORE_SCHEMA_SQL, AGENT_V14_SESSION_SHARING_SCHEMA_SQL, } from "./openclaw-agent-session-sharing-schema.js"; +import { SESSION_TRANSCRIPT_ARCHIVES_TABLE } from "./openclaw-agent-session-transcript-archive-schema.js"; import { STANDING_INTENTS_FTS_SHADOW_TABLES, STANDING_INTENTS_FTS_TABLE, @@ -49,6 +50,7 @@ const AGENT_SCHEMA_COMPATIBILITY = { MEMORY_INDEX_CHUNK_PROVENANCE_TABLE, MEMORY_INDEX_CHUNK_RECALL_METADATA_TABLE, CONTEXT_ENGINE_TURN_OUTBOX_TABLE, + SESSION_TRANSCRIPT_ARCHIVES_TABLE, STANDING_INTENTS_TABLE, STANDING_INTENTS_FTS_TABLE, ...STANDING_INTENTS_FTS_SHADOW_TABLES, diff --git a/src/state/openclaw-agent-db.generated.d.ts b/src/state/openclaw-agent-db.generated.d.ts index 67aecf2b402e..c973189061f2 100644 --- a/src/state/openclaw-agent-db.generated.d.ts +++ b/src/state/openclaw-agent-db.generated.d.ts @@ -273,6 +273,22 @@ export interface SessionTranscriptActiveEvents { session_id: string; } +export interface SessionTranscriptArchives { + archive_blob: Uint8Array; + archive_name: string; + archive_sha256: string; + created_at: number; + encoding: string; + generation: string; + last_publish_attempt_at: number | null; + last_publish_error: string | null; + publish_attempts: Generated; + published_at: number | null; + reason: string; + session_id: string; + session_key: string; +} + export interface SessionTranscriptFts { message_id: string | null; role: string | null; @@ -450,6 +466,7 @@ export interface DB { session_nodes: SessionNodes; session_suggestions: SessionSuggestions; session_transcript_active_events: SessionTranscriptActiveEvents; + session_transcript_archives: SessionTranscriptArchives; session_transcript_fts: SessionTranscriptFts; session_transcript_fts_config: SessionTranscriptFtsConfig; session_transcript_fts_content: SessionTranscriptFtsContent; diff --git a/src/state/openclaw-agent-schema.sql b/src/state/openclaw-agent-schema.sql index 98314c52d632..767671751e11 100644 --- a/src/state/openclaw-agent-schema.sql +++ b/src/state/openclaw-agent-schema.sql @@ -324,6 +324,33 @@ CREATE TABLE IF NOT EXISTS transcript_events ( FOREIGN KEY (session_id) REFERENCES "session_windows"(session_id) ON DELETE CASCADE ) STRICT; +-- Canonical cold-tier owner for reclaimed transcript generations. The derived +-- .deleted/.reset file may be recreated from this row after a crash. +CREATE TABLE IF NOT EXISTS session_transcript_archives ( + session_id TEXT NOT NULL, + generation TEXT NOT NULL, + session_key TEXT NOT NULL, + reason TEXT NOT NULL CHECK (reason IN ('deleted', 'reset')), + encoding TEXT NOT NULL CHECK (encoding IN ('identity', 'zstd')), + archive_blob BLOB NOT NULL, + archive_sha256 TEXT NOT NULL CHECK (length(archive_sha256) = 64), + archive_name TEXT NOT NULL UNIQUE, + created_at INTEGER NOT NULL, + published_at INTEGER, + publish_attempts INTEGER NOT NULL DEFAULT 0 CHECK (publish_attempts >= 0), + last_publish_attempt_at INTEGER, + last_publish_error TEXT, + PRIMARY KEY (session_id, generation), + CHECK (archive_name NOT LIKE '%/%' AND archive_name NOT LIKE '%\%') +) STRICT; + +CREATE INDEX IF NOT EXISTS idx_agent_session_transcript_archives_pending + ON session_transcript_archives(created_at, session_id, generation) + WHERE published_at IS NULL; + +CREATE INDEX IF NOT EXISTS idx_agent_session_transcript_archives_retention + ON session_transcript_archives(created_at, session_id, generation); + CREATE TABLE IF NOT EXISTS transcript_rewrite_watermarks ( session_id TEXT NOT NULL PRIMARY KEY, generation TEXT NOT NULL, diff --git a/src/state/openclaw-agent-session-transcript-archive-schema.test.ts b/src/state/openclaw-agent-session-transcript-archive-schema.test.ts new file mode 100644 index 000000000000..b6aad7b019b2 --- /dev/null +++ b/src/state/openclaw-agent-session-transcript-archive-schema.test.ts @@ -0,0 +1,133 @@ +import { DatabaseSync } from "node:sqlite"; +import { afterEach, describe, expect, it } from "vitest"; +import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js"; +import { assertSqliteSchemaContains } from "../infra/sqlite-schema-contract.js"; +import { + closeOpenClawAgentDatabasesForTest, + openOpenClawAgentDatabase, +} from "./openclaw-agent-db.js"; +import { OPENCLAW_AGENT_SCHEMA_SQL } from "./openclaw-agent-schema.js"; +import { + ensureSessionTranscriptArchiveSchema, + SESSION_TRANSCRIPT_ARCHIVES_TABLE, +} from "./openclaw-agent-session-transcript-archive-schema.js"; + +const tempDirs = useAutoCleanupTempDirTracker(afterEach); + +afterEach(() => { + closeOpenClawAgentDatabasesForTest(); +}); + +function schemaWithoutTranscriptArchives(): string { + const start = OPENCLAW_AGENT_SCHEMA_SQL.indexOf( + `CREATE TABLE IF NOT EXISTS ${SESSION_TRANSCRIPT_ARCHIVES_TABLE} (`, + ); + const end = OPENCLAW_AGENT_SCHEMA_SQL.indexOf( + "CREATE TABLE IF NOT EXISTS transcript_rewrite_watermarks (", + start, + ); + expect(start).toBeGreaterThanOrEqual(0); + expect(end).toBeGreaterThan(start); + return `${OPENCLAW_AGENT_SCHEMA_SQL.slice(0, start)}${OPENCLAW_AGENT_SCHEMA_SQL.slice(end)}`; +} + +describe("session transcript archive schema", () => { + it("keeps a current database table-free until first archive use without changing its version", () => { + const stateDir = tempDirs.make("openclaw-session-archive-schema-"); + const options = { agentId: "main", env: { OPENCLAW_STATE_DIR: stateDir } }; + const initial = openOpenClawAgentDatabase(options); + const databasePath = initial.path; + closeOpenClawAgentDatabasesForTest(); + + const shipped = new DatabaseSync(databasePath); + shipped.exec(` + DROP INDEX idx_agent_session_transcript_archives_pending; + DROP INDEX idx_agent_session_transcript_archives_retention; + DROP TABLE ${SESSION_TRANSCRIPT_ARCHIVES_TABLE}; + `); + const versionBefore = shipped.prepare("PRAGMA user_version").get(); + const metadataBefore = shipped + .prepare("SELECT schema_version, updated_at FROM schema_meta WHERE meta_key = 'primary'") + .get(); + shipped.close(); + + const reopened = openOpenClawAgentDatabase(options); + expect( + reopened.db + .prepare("SELECT 1 FROM sqlite_schema WHERE type = 'table' AND name = ?") + .get(SESSION_TRANSCRIPT_ARCHIVES_TABLE), + ).toBeUndefined(); + + ensureSessionTranscriptArchiveSchema(reopened.db); + + expect( + reopened.db + .prepare("SELECT 1 FROM sqlite_schema WHERE type = 'table' AND name = ?") + .get(SESSION_TRANSCRIPT_ARCHIVES_TABLE), + ).toEqual({ 1: 1 }); + expect(reopened.db.prepare("PRAGMA user_version").get()).toEqual(versionBefore); + expect( + reopened.db + .prepare("SELECT schema_version, updated_at FROM schema_meta WHERE meta_key = 'primary'") + .get(), + ).toEqual(metadataBefore); + }); + + it("keeps a populated additive archive table usable by the previous schema contract", () => { + const database = new DatabaseSync(":memory:"); + try { + database.exec(OPENCLAW_AGENT_SCHEMA_SQL); + database + .prepare( + `INSERT INTO ${SESSION_TRANSCRIPT_ARCHIVES_TABLE} ( + session_id, generation, session_key, reason, encoding, archive_blob, archive_sha256, + archive_name, created_at + ) VALUES (?, ?, ?, 'deleted', 'identity', ?, ?, ?, ?)`, + ) + .run( + "session-1", + "generation-1", + "agent:main:session-1", + Buffer.from("{}\n"), + "ca3d163bab055381827226140568f3bef7eaac187cebd76878e0b63e9e442356", + "session-1.jsonl.deleted.2026-08-14T00-00-00.000Z", + Date.now(), + ); + + expect(() => + assertSqliteSchemaContains( + database, + "previous agent schema", + schemaWithoutTranscriptArchives(), + ), + ).not.toThrow(); + expect( + database.prepare("SELECT archive_name FROM session_transcript_archives").get(), + ).toEqual({ archive_name: "session-1.jsonl.deleted.2026-08-14T00-00-00.000Z" }); + } finally { + database.close(); + } + }); + + it("rejects a drifted archive table instead of treating it as an optional absence", () => { + const stateDir = tempDirs.make("openclaw-session-archive-drift-"); + const options = { agentId: "main", env: { OPENCLAW_STATE_DIR: stateDir } }; + const initial = openOpenClawAgentDatabase(options); + const databasePath = initial.path; + closeOpenClawAgentDatabasesForTest(); + + const drifted = new DatabaseSync(databasePath); + drifted.exec(` + DROP INDEX idx_agent_session_transcript_archives_pending; + DROP INDEX idx_agent_session_transcript_archives_retention; + DROP TABLE ${SESSION_TRANSCRIPT_ARCHIVES_TABLE}; + CREATE TABLE ${SESSION_TRANSCRIPT_ARCHIVES_TABLE} ( + session_id TEXT NOT NULL PRIMARY KEY, + archive_blob BLOB NOT NULL + ) STRICT; + `); + drifted.close(); + + expect(() => openOpenClawAgentDatabase(options)).toThrow(/session_transcript_archives|schema/u); + }); +}); diff --git a/src/state/openclaw-agent-session-transcript-archive-schema.ts b/src/state/openclaw-agent-session-transcript-archive-schema.ts new file mode 100644 index 000000000000..bab49609e158 --- /dev/null +++ b/src/state/openclaw-agent-session-transcript-archive-schema.ts @@ -0,0 +1,34 @@ +import type { DatabaseSync } from "node:sqlite"; +import { runSqliteImmediateTransactionSync } from "../infra/sqlite-transaction.js"; +import { OPENCLAW_AGENT_SCHEMA_SQL } from "./openclaw-agent-schema.js"; + +export const SESSION_TRANSCRIPT_ARCHIVES_TABLE = "session_transcript_archives"; + +const ARCHIVE_SCHEMA_START = `CREATE TABLE IF NOT EXISTS ${SESSION_TRANSCRIPT_ARCHIVES_TABLE} (`; +const ARCHIVE_SCHEMA_END = "CREATE TABLE IF NOT EXISTS transcript_rewrite_watermarks ("; +const ENSURED_DATABASES = new WeakSet(); + +function sessionTranscriptArchiveSchemaSql(): string { + const start = OPENCLAW_AGENT_SCHEMA_SQL.indexOf(ARCHIVE_SCHEMA_START); + const end = OPENCLAW_AGENT_SCHEMA_SQL.indexOf(ARCHIVE_SCHEMA_END, start); + if (start === -1 || end === -1) { + throw new Error("OpenClaw session transcript archive schema markers are missing."); + } + return OPENCLAW_AGENT_SCHEMA_SQL.slice(start, end); +} + +/** Lazily installs the additive canonical archive owner on first archive use. */ +export function ensureSessionTranscriptArchiveSchema(db: DatabaseSync): void { + if (ENSURED_DATABASES.has(db)) { + return; + } + const ensure = () => { + db.exec(sessionTranscriptArchiveSchemaSql()); // sqlite-allow-raw -- Canonical additive DDL only. + }; + if (db.isTransaction) { + ensure(); + return; + } + runSqliteImmediateTransactionSync(db, ensure); + ENSURED_DATABASES.add(db); +}