From 9bfe292c8686c9f55282fb5773b8729f96b68051 Mon Sep 17 00:00:00 2001 From: Heming Zeng Date: Wed, 5 Aug 2026 05:40:59 +0800 Subject: [PATCH] fix(sessions): keep gateway responsive while large transcript archives are built (#112424) * fix(sessions): keep gateway responsive during transcript archiving * test(sessions): strengthen archive worker coverage * fix(sessions): bound archive worker concurrency * test(sessions): normalize rollback paths on Windows * fix(sessions): load source archive worker on node 22 * fix(sessions): align archive fencing with session windows * test(sessions): adapt cleanup race to archive worker * fix(sessions): harden archive worker lifecycle * style(sessions): format archive lifecycle tests * fix(sessions): release writer lane during archive work --------- Co-authored-by: Josh Lehman --- config/knip.config.ts | 2 + package.json | 2 +- scripts/ci-changed-scope.mjs | 4 +- scripts/release-check.ts | 1 + .../incognito-session-transcript.test.ts | 93 +- .../session-accessor.sqlite-archive.ts | 197 ++++- ...ion-accessor.sqlite-archive.worker.test.ts | 811 ++++++++++++++++++ .../session-accessor.sqlite-archive.worker.ts | 171 ++++ ...ssion-accessor.sqlite-cleanup-race.test.ts | 103 ++- ...session-accessor.sqlite-delete-snapshot.ts | 71 ++ .../sessions/session-accessor.sqlite-entry.ts | 26 +- ...session-accessor.sqlite-lifecycle-state.ts | 51 +- .../session-accessor.sqlite-lifecycle.ts | 74 +- .../session-accessor.sqlite-maintenance.ts | 59 +- .../session-accessor.sqlite-parent-session.ts | 4 +- .../session-accessor.sqlite-projection.ts | 18 +- .../sessions/session-history-eviction.ts | 2 +- src/infra/tsdown-config.test.ts | 1 + src/scripts/ci-changed-scope.windows.test.ts | 6 +- src/state/openclaw-agent-db.ts | 5 + .../sqlite-sessions-transcripts-flip-proof.ts | 10 + test/package-scripts.test.ts | 7 + test/release-check.test.ts | 2 + ...anscripts-flip-proof.built-cli.e2e.test.ts | 267 ++++++ ...essions-transcripts-flip-proof.e2e.test.ts | 4 +- tsdown.config.ts | 2 + 26 files changed, 1844 insertions(+), 149 deletions(-) create mode 100644 src/config/sessions/session-accessor.sqlite-archive.worker.test.ts create mode 100644 src/config/sessions/session-accessor.sqlite-archive.worker.ts create mode 100644 src/config/sessions/session-accessor.sqlite-delete-snapshot.ts diff --git a/config/knip.config.ts b/config/knip.config.ts index 7462b9400545..031c1ffd241a 100644 --- a/config/knip.config.ts +++ b/config/knip.config.ts @@ -122,6 +122,8 @@ const rootEntries = [ "scripts/e2e/*.{js,mjs,ts}!", "scripts/e2e/lib/**/{assertions,probe,mock-server}.{js,mjs,ts}!", "src/audit/audit-event-writer.worker.ts!", + // Loaded by URL from the SQLite lifecycle archive owner. + "src/config/sessions/session-accessor.sqlite-archive.worker.ts!", "src/state/openclaw-database-verify.worker.ts!", "src/agents/model-provider-auth.worker.ts!", // Loaded by URL from setup-inference-detection.ts; no static import edge exists. diff --git a/package.json b/package.json index a2b8404ea0eb..3602df095983 100644 --- a/package.json +++ b/package.json @@ -1898,7 +1898,7 @@ "test:unit:fast:audit": "node scripts/test-unit-fast-audit.mjs", "test:voicecall:closedloop": "node scripts/test-voicecall-closedloop.mjs", "test:watch": "node scripts/test-projects.mjs --watch", - "test:windows:ci": "node scripts/test-projects.mjs src/shared/runtime-import.test.ts src/infra/sqlite-snapshot.test.ts src/infra/ssh-client.windows.test.ts src/infra/update-managed-service-handoff.test.ts src/infra/exec-allowlist-pattern.test.ts src/infra/fs-safe-remove.test.ts src/snapshot/local-repository.windows.test.ts src/state/openclaw-database-paths.windows.test.ts src/commands/backup-verify.test.ts src/infra/state-migrations.legacy-session-store.test.ts src/test-utils/openclaw-test-state.test.ts src/agents/sessions/windows-git-bash-path.test.ts src/agents/bash-tools.exec.script-preflight.test.ts src/process/exec.windows.test.ts src/process/exec.windows.integration.test.ts src/process/windows-command.test.ts src/infra/windows-install-roots.test.ts src/node-host/invoke-system-run-allowlist.test.ts src/daemon/schtasks.startup-fallback.test.ts extensions/lobster/src/lobster-runner.test.ts extensions/mxc/test/mxc-backend.test.ts extensions/mxc/test/sandbox-policy-loader.test.ts test/e2e/qa-lab/runtime/package-openclaw-for-docker.e2e.test.ts test/scripts/direct-run-entrypoints.test.ts test/scripts/format-generated-module.test.ts test/scripts/npm-runner.test.ts test/scripts/openclaw-cross-os-installer.windows.test.ts test/scripts/openclaw-cross-os-release-workflow.test.ts test/scripts/pnpm-runner.test.ts test/scripts/run-with-env.test.ts test/scripts/ts-topology.test.ts test/scripts/ui.test.ts test/scripts/vitest-process-group.test.ts", + "test:windows:ci": "node scripts/test-projects.mjs src/shared/runtime-import.test.ts src/config/sessions/session-accessor.sqlite-archive.worker.test.ts src/infra/sqlite-snapshot.test.ts src/infra/ssh-client.windows.test.ts src/infra/update-managed-service-handoff.test.ts src/infra/exec-allowlist-pattern.test.ts src/infra/fs-safe-remove.test.ts src/snapshot/local-repository.windows.test.ts src/state/openclaw-database-paths.windows.test.ts src/commands/backup-verify.test.ts src/infra/state-migrations.legacy-session-store.test.ts src/test-utils/openclaw-test-state.test.ts src/agents/sessions/windows-git-bash-path.test.ts src/agents/bash-tools.exec.script-preflight.test.ts src/process/exec.windows.test.ts src/process/exec.windows.integration.test.ts src/process/windows-command.test.ts src/infra/windows-install-roots.test.ts src/node-host/invoke-system-run-allowlist.test.ts src/daemon/schtasks.startup-fallback.test.ts extensions/lobster/src/lobster-runner.test.ts extensions/mxc/test/mxc-backend.test.ts extensions/mxc/test/sandbox-policy-loader.test.ts test/e2e/qa-lab/runtime/package-openclaw-for-docker.e2e.test.ts test/scripts/direct-run-entrypoints.test.ts test/scripts/format-generated-module.test.ts test/scripts/npm-runner.test.ts test/scripts/openclaw-cross-os-installer.windows.test.ts test/scripts/openclaw-cross-os-release-workflow.test.ts test/scripts/pnpm-runner.test.ts test/scripts/run-with-env.test.ts test/scripts/ts-topology.test.ts test/scripts/ui.test.ts test/scripts/vitest-process-group.test.ts", "test:windows:schtasks:integration": "node scripts/run-with-env.mjs CI_WINDOWS_SCHTASKS_INTEGRATION=1 OPENCLAW_E2E_VERBOSE=1 OPENCLAW_VITEST_MAX_WORKERS=1 -- node scripts/run-vitest.mjs src/daemon/schtasks.integration.e2e.test.ts", "tool-display:check": "node --import tsx scripts/tool-display.ts --check", "tool-display:write": "node --import tsx scripts/tool-display.ts --write", diff --git a/scripts/ci-changed-scope.mjs b/scripts/ci-changed-scope.mjs index f0eb9207fd02..6e2ac53a2b38 100644 --- a/scripts/ci-changed-scope.mjs +++ b/scripts/ci-changed-scope.mjs @@ -50,9 +50,9 @@ const NODE_SCOPE_RE = /^(src\/|test\/|extensions\/|packages\/|scripts\/|ui\/|\.github\/|openclaw\.mjs$|package\.json$|pnpm-lock\.yaml$|pnpm-workspace\.yaml$|tsconfig.*\.json$|vitest.*\.ts$|tsdown\.config\.ts$|\.oxlintrc\.json$|\.oxfmtrc\.jsonc$)/; const WINDOWS_SQLITE_SCOPE_RE = /^src\/(?:state\/|.*sqlite.*\.ts$)/; const WINDOWS_SCOPE_RE = - /^(extensions\/mxc\/|src\/agents\/(?:bash-tools\.exec-script-(?:preflight|target)|bash-tools\.exec\.script-preflight\.test)\.ts$|src\/config\/sessions\/(?:session-accessor\.sqlite-archive|store\.session-lifecycle-mutation\.test)\.ts$|src\/process\/|src\/infra\/(?:(?:exec-allowlist-pattern|fs-safe-remove)(?:\.test)?|ssh-client(?:\.windows\.test)?|update-managed-service-handoff(?:\.test)?|windows-install-roots)\.ts$|src\/shared\/(?:import-specifier|runtime-import)(?:\.test)?\.ts$|src\/test-utils\/openclaw-test-state(?:\.test)?\.ts$|scripts\/(?:android-(?:app-i18n|pin-version)\.ts|ci-run-timings\.mjs|e2e\/lib\/package-compat\.mjs|generate-bundled-channel-config-metadata\.ts|install\.ps1|openclaw-cross-os-release-checks\.ts|plan-release-workflow-matrix\.mjs|run-additional-boundary-checks\.mjs|verify-docker-attestations\.mjs|github\/run-openclaw-cross-os-release-checks\.sh|(?:npm-runner|pnpm-runner|ui|vitest-process-group)\.(?:mjs|js)|lib\/(?:direct-run\.mjs|format-generated-module\.mjs|cross-os-release-checks\/[^/]+\.ts))$|test\/scripts\/(?:direct-run-entrypoints|format-generated-module|install-ps1|npm-runner|openclaw-cross-os-release-workflow|pnpm-runner|ui|vitest-process-group)\.test\.ts$|package\.json$|pnpm-lock\.yaml$|pnpm-workspace\.yaml$|\.github\/workflows\/(?:ci|openclaw-cross-os-release-checks-reusable)\.yml$|\.github\/actions\/setup-node-env\/action\.yml$|\.github\/actions\/setup-pnpm-store-cache\/action\.yml$)/; + /^(extensions\/mxc\/|src\/agents\/(?:bash-tools\.exec-script-(?:preflight|target)|bash-tools\.exec\.script-preflight\.test)\.ts$|src\/config\/sessions\/(?:session-accessor\.sqlite-archive(?:\.worker(?:\.test)?)?|store\.session-lifecycle-mutation\.test)\.ts$|src\/process\/|src\/infra\/(?:(?:exec-allowlist-pattern|fs-safe-remove)(?:\.test)?|ssh-client(?:\.windows\.test)?|update-managed-service-handoff(?:\.test)?|windows-install-roots)\.ts$|src\/shared\/(?:import-specifier|runtime-import)(?:\.test)?\.ts$|src\/test-utils\/openclaw-test-state(?:\.test)?\.ts$|scripts\/(?:android-(?:app-i18n|pin-version)\.ts|ci-run-timings\.mjs|e2e\/lib\/package-compat\.mjs|generate-bundled-channel-config-metadata\.ts|install\.ps1|openclaw-cross-os-release-checks\.ts|plan-release-workflow-matrix\.mjs|run-additional-boundary-checks\.mjs|verify-docker-attestations\.mjs|github\/run-openclaw-cross-os-release-checks\.sh|(?:npm-runner|pnpm-runner|ui|vitest-process-group)\.(?:mjs|js)|lib\/(?:direct-run\.mjs|format-generated-module\.mjs|cross-os-release-checks\/[^/]+\.ts))$|test\/scripts\/(?:direct-run-entrypoints|format-generated-module|install-ps1|npm-runner|openclaw-cross-os-release-workflow|pnpm-runner|ui|vitest-process-group)\.test\.ts$|package\.json$|pnpm-lock\.yaml$|pnpm-workspace\.yaml$|\.github\/workflows\/(?:ci|openclaw-cross-os-release-checks-reusable)\.yml$|\.github\/actions\/setup-node-env\/action\.yml$|\.github\/actions\/setup-pnpm-store-cache\/action\.yml$)/; const WINDOWS_TEST_SCOPE_RE = - /^(extensions\/mxc\/test\/(?:mxc-backend|sandbox-policy-loader)\.test\.ts$|src\/agents\/bash-tools\.exec\.script-preflight\.test\.ts$|src\/config\/sessions\/store\.session-lifecycle-mutation\.test\.ts$|src\/process\/(?:exec\.windows|windows-command)\.test\.ts$|src\/infra\/(?:exec-allowlist-pattern|fs-safe-remove|ssh-client\.windows|update-managed-service-handoff|windows-install-roots)\.test\.ts$|src\/shared\/runtime-import\.test\.ts$|src\/state\/openclaw-database-paths\.windows\.test\.ts$|src\/test-utils\/openclaw-test-state\.test\.ts$|test\/scripts\/(?:direct-run-entrypoints|format-generated-module|npm-runner|openclaw-cross-os-release-workflow|pnpm-runner|ui|vitest-process-group)\.test\.ts$)/; + /^(extensions\/mxc\/test\/(?:mxc-backend|sandbox-policy-loader)\.test\.ts$|src\/agents\/bash-tools\.exec\.script-preflight\.test\.ts$|src\/config\/sessions\/(?:session-accessor\.sqlite-archive\.worker|store\.session-lifecycle-mutation)\.test\.ts$|src\/process\/(?:exec\.windows|windows-command)\.test\.ts$|src\/infra\/(?:exec-allowlist-pattern|fs-safe-remove|ssh-client\.windows|update-managed-service-handoff|windows-install-roots)\.test\.ts$|src\/shared\/runtime-import\.test\.ts$|src\/state\/openclaw-database-paths\.windows\.test\.ts$|src\/test-utils\/openclaw-test-state\.test\.ts$|test\/scripts\/(?:direct-run-entrypoints|format-generated-module|npm-runner|openclaw-cross-os-release-workflow|pnpm-runner|ui|vitest-process-group)\.test\.ts$)/; const WINDOWS_DAEMON_SCOPE_RE = /^src\/daemon\/(?:schtasks(?:[-.][^/]+)?|runtime-hints\.windows-paths(?:\.test)?|test-helpers\/schtasks-(?:base-mocks|fixtures))\.ts$/; const CONTROL_UI_I18N_SCOPE_RE = diff --git a/scripts/release-check.ts b/scripts/release-check.ts index b04a7c9235ca..b97ae7f4ea65 100755 --- a/scripts/release-check.ts +++ b/scripts/release-check.ts @@ -114,6 +114,7 @@ const requiredPathGroups = [ "dist/agents/compaction-planning.worker.js", "dist/agents/model-provider-auth.worker.js", "dist/audit/audit-event-writer.worker.js", + "dist/config/sessions/session-accessor.sqlite-archive.worker.js", "dist/config/sessions/session-transcript-reconcile.worker.js", "dist/state/openclaw-database-verify.worker.js", "dist/system-agent/setup-inference-detection.worker.js", diff --git a/src/config/sessions/incognito-session-transcript.test.ts b/src/config/sessions/incognito-session-transcript.test.ts index e40b77130812..a34993080698 100644 --- a/src/config/sessions/incognito-session-transcript.test.ts +++ b/src/config/sessions/incognito-session-transcript.test.ts @@ -3,9 +3,19 @@ import os from "node:os"; import path from "node:path"; import { afterEach, describe, expect, it } from "vitest"; import { SessionManager } from "../../agents/sessions/session-manager.js"; -import { closeOpenClawAgentDatabasesForTest } from "../../state/openclaw-agent-db.js"; +import { + closeOpenClawAgentDatabasesForTest, + resolveIncognitoOpenClawAgentSqlitePath, +} from "../../state/openclaw-agent-db.js"; import { resolveStorePath } from "./paths.js"; -import { createSessionEntryWithTranscript, loadSessionEntry } from "./session-accessor.js"; +import { + createSessionEntryWithTranscript, + listSessionEntries, + loadSessionEntry, + loadTranscriptEvents, + patchSessionEntry, +} from "./session-accessor.js"; +import { replaceSqliteTranscriptEvents } from "./session-accessor.sqlite.js"; const sessionKey = "agent:main:dashboard:incognito-round-trip"; @@ -81,4 +91,83 @@ describe("incognito transcript access", () => { fs.rmSync(cwd, { force: true, recursive: true }); } }); + + it("prunes incognito transcripts in process without publishing a disk archive", async () => { + const stateDir = fs.realpathSync( + fs.mkdtempSync(path.join(fs.realpathSync(os.tmpdir()), "incognito-maintenance-")), + ); + const env = { ...process.env, OPENCLAW_STATE_DIR: stateDir }; + const storePath = resolveIncognitoOpenClawAgentSqlitePath({ agentId: "main", env }); + const archiveDirectory = path.join(path.dirname(path.dirname(storePath)), "sessions"); + const staleScope = { + agentId: "main", + env, + sessionKey: "agent:main:dashboard:incognito-stale", + storePath, + }; + const activeScope = { + agentId: "main", + env, + sessionKey: "agent:main:dashboard:incognito-active", + storePath, + }; + const now = Date.now(); + + try { + await patchSessionEntry( + staleScope, + () => ({ sessionId: "incognito-stale-session", updatedAt: now }), + { + fallbackEntry: { sessionId: "incognito-stale-session", updatedAt: now }, + replaceEntry: true, + skipMaintenance: true, + }, + ); + await replaceSqliteTranscriptEvents({ ...staleScope, sessionId: "incognito-stale-session" }, [ + { + id: "incognito-stale-event", + timestamp: new Date(now).toISOString(), + type: "metadata", + }, + ]); + await patchSessionEntry( + activeScope, + () => ({ sessionId: "incognito-active-session", updatedAt: now + 1 }), + { + fallbackEntry: { sessionId: "incognito-active-session", updatedAt: now + 1 }, + replaceEntry: true, + skipMaintenance: true, + }, + ); + + await patchSessionEntry(activeScope, () => ({ model: "gpt-test" }), { + maintenanceConfig: { + highWaterBytes: null, + maxDiskBytes: null, + maxEntries: 1, + mode: "enforce", + modelRunPruneAfterMs: 24 * 60 * 60 * 1000, + pruneAfterMs: 365 * 24 * 60 * 60 * 1000, + resetArchiveRetentionMs: null, + }, + }); + + expect( + listSessionEntries({ agentId: "main", env, storePath }).map( + (summary) => summary.sessionKey, + ), + ).toEqual([activeScope.sessionKey]); + await expect( + loadTranscriptEvents({ + ...staleScope, + sessionId: "incognito-stale-session", + }), + ).resolves.toEqual([]); + expect(fs.existsSync(storePath)).toBe(false); + expect(fs.existsSync(archiveDirectory)).toBe(false); + } finally { + closeOpenClawAgentDatabasesForTest(); + fs.rmSync(stateDir, { force: true, recursive: true }); + } + }); }); diff --git a/src/config/sessions/session-accessor.sqlite-archive.ts b/src/config/sessions/session-accessor.sqlite-archive.ts index 047058cc0683..4079dd7b1ddb 100644 --- a/src/config/sessions/session-accessor.sqlite-archive.ts +++ b/src/config/sessions/session-accessor.sqlite-archive.ts @@ -1,7 +1,10 @@ import { randomUUID } from "node:crypto"; import fs from "node:fs"; import path from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { Worker } from "node:worker_threads"; import { syncDirectoryBestEffortSync } from "../../infra/directory-durability.js"; +import { KeyedAsyncQueue } from "../../plugin-sdk/keyed-async-queue.js"; import { encodeSessionArchiveContent, readSessionArchiveContentSync, @@ -10,19 +13,58 @@ import { import { formatSessionArchiveTimestamp, type SessionArchiveReason } from "./artifacts.js"; import type { SessionLifecycleArchivedTranscript } from "./session-accessor.sqlite-contract.js"; +export type SqliteSessionStateDeleteSnapshot = { + acpParentStreamEventCount: number; + generation: string | null; + lastSeq: number | null; + sessionUpdatedAt: number | null; + trajectoryLastSeq: number | null; + transcriptUpdatedAt: number | null; +}; + export type SqliteSessionStateDeletePlan = { + agentId: string; archiveDirectory: string; archiveTranscript: boolean; - content: string; - hadTranscriptState: boolean; + databasePath: string; reason: "deleted" | "reset"; sessionId: string; + snapshot: SqliteSessionStateDeleteSnapshot; }; export type MaterializedSqliteSessionStateDeletePlan = SqliteSessionStateDeletePlan & { archivedTranscript: SessionLifecycleArchivedTranscript | null; }; +export type SqliteTranscriptArchiveWorkerPlan = Pick< + SqliteSessionStateDeletePlan, + "agentId" | "archiveDirectory" | "databasePath" | "reason" | "sessionId" | "snapshot" +>; + +export type SqliteTranscriptArchiveWorkerResult = { + archivedPath: string | null; + sessionId: string; +}; + +export type SqliteTranscriptArchiveWorkerMessage = { + type: "done"; + results: SqliteTranscriptArchiveWorkerResult[]; +}; + +export function sqliteSessionStateDeleteSnapshotsEqual( + left: SqliteSessionStateDeleteSnapshot, + right: SqliteSessionStateDeleteSnapshot, +): 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 + ); +} + function resolveSqliteTranscriptArchivePath(params: { archiveDirectory: string; reason: SessionArchiveReason; @@ -54,7 +96,7 @@ function findMatchingSqliteTranscriptArchive(params: { } const prefix = `${params.sessionId}.jsonl.${params.reason}.`; for (const entry of entries) { - if (!entry.startsWith(prefix)) { + if (!entry.startsWith(prefix) || entry.endsWith(".tmp")) { continue; } const archivePath = path.join(params.archiveDirectory, entry); @@ -107,8 +149,8 @@ export function writeSqliteTranscriptArchive(params: { writeDurableFileExclusive(tempPath, encoded.bytes); fs.renameSync(tempPath, archivePath); syncDirectoryBestEffortSync(params.archiveDirectory); - // Full readback is bounded by the same single-generation content the - // delete plan already buffers (Node string limits cap both); a partial + // Full readback is bounded by the same single-generation content held by + // this Worker (Node string limits cap both); a partial // or corrupt archive must fail here, before any rows are reclaimed. if (readSessionArchiveContentSync(archivePath) !== params.content) { fs.rmSync(archivePath, { force: true }); @@ -138,28 +180,123 @@ function writeDurableFileExclusive(filePath: string, content: Buffer): void { } } -// Runs duplicate probing, archive write, rename, and fsync outside SQLite -// write transactions; deletion later consumes this durable proof. -export function materializeSqliteSessionStateDeletePlans( +function resolveSqliteTranscriptArchiveWorkerUrl(currentModuleUrl = import.meta.url): URL { + const currentPath = fileURLToPath(currentModuleUrl); + const normalized = currentPath.replaceAll(path.sep, "/"); + const distMarker = "/dist/"; + const distIndex = normalized.lastIndexOf(distMarker); + if (distIndex >= 0) { + const distRoot = currentPath.slice(0, distIndex + distMarker.length); + return pathToFileURL( + path.join(distRoot, "config", "sessions", "session-accessor.sqlite-archive.worker.js"), + ); + } + const extension = path.extname(currentPath) || ".js"; + return new URL(`./session-accessor.sqlite-archive.worker${extension}`, currentModuleUrl); +} + +function resolveSourceWorkerExecArgv(): string[] { + // Node 22 can strip the .ts entrypoint itself, but `--import tsx` does not + // register tsx's ESM resolver inside a Worker. Explicitly register the + // supported programmatic API so source-tree .js specifiers map back to .ts. + // Built .js workers do not use this development/test-only preload. + const tsxApiUrl = import.meta.resolve("tsx/esm/api"); + const registerTsx = `import { register } from ${JSON.stringify(tsxApiUrl)}; register();`; + return ["--import", `data:text/javascript,${encodeURIComponent(registerTsx)}`]; +} + +function normalizeArchiveWorkerError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)); +} + +function spawnSqliteTranscriptArchiveWorker( + plans: readonly SqliteTranscriptArchiveWorkerPlan[], +): Promise { + const workerUrl = resolveSqliteTranscriptArchiveWorkerUrl(); + let worker: Worker; + try { + const sourceWorkerExecArgv = workerUrl.pathname.endsWith(".ts") + ? resolveSourceWorkerExecArgv() + : undefined; + worker = new Worker(workerUrl, { + workerData: { type: "sqlite-transcript-archive-v1", plans }, + execArgv: sourceWorkerExecArgv, + }); + } catch (error) { + return Promise.reject(normalizeArchiveWorkerError(error)); + } + + return new Promise((resolve, reject) => { + let results: SqliteTranscriptArchiveWorkerResult[] | undefined; + let workerError: Error | undefined; + worker.once("message", (message: SqliteTranscriptArchiveWorkerMessage) => { + results = message.results; + }); + 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. + workerError = normalizeArchiveWorkerError(error); + }); + worker.once("exit", (code) => { + worker.removeAllListeners(); + if (workerError) { + reject(workerError); + return; + } + if (code !== 0) { + reject(new Error(`SQLite transcript archive worker exited with code ${code}`)); + return; + } + if (!results) { + reject(new Error("SQLite transcript archive worker exited without results")); + return; + } + resolve(results); + }); + }); +} + +// Serialize lifecycle archive Workers so this path cannot multiply +// whole-buffer usage across several Worker heaps at once. +const sqliteTranscriptArchiveWorkerQueue = new KeyedAsyncQueue(); +const SQLITE_TRANSCRIPT_ARCHIVE_WORKER_QUEUE_KEY = "lifecycle-archive"; + +function runSqliteTranscriptArchiveWorker( + plans: readonly SqliteTranscriptArchiveWorkerPlan[], +): Promise { + return sqliteTranscriptArchiveWorkerQueue.enqueue( + SQLITE_TRANSCRIPT_ARCHIVE_WORKER_QUEUE_KEY, + () => spawnSqliteTranscriptArchiveWorker(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 async function materializeSqliteSessionStateDeletePlans( plans: readonly SqliteSessionStateDeletePlan[], -): MaterializedSqliteSessionStateDeletePlan[] { - return dedupeSqliteSessionStateDeletePlans(plans).map((plan) => { - // Empty content means no transcript to preserve (e.g. trajectory-only - // sessions). Writing a verified-but-empty archive would fake extraction; - // trajectory runtime events are diagnostic telemetry and are reclaimed - // without an archive artifact. - const archivedTranscript = - plan.archiveTranscript && plan.content.length > 0 - ? { - archivedPath: writeSqliteTranscriptArchive({ - archiveDirectory: plan.archiveDirectory, - content: plan.content, - reason: plan.reason, - sessionId: plan.sessionId, - }), - sourcePath: path.join(plan.archiveDirectory, `${plan.sessionId}.jsonl`), - } - : null; +): Promise { + const deduped = dedupeSqliteSessionStateDeletePlans(plans); + const archivePlans = deduped.filter((plan) => plan.archiveTranscript); + const workerResults = + archivePlans.length > 0 ? await runSqliteTranscriptArchiveWorker(archivePlans) : []; + const resultBySessionId = new Map(workerResults.map((result) => [result.sessionId, result])); + + return deduped.map((plan) => { + if (!plan.archiveTranscript) { + return Object.assign({}, plan, { 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 }); }); } @@ -176,7 +313,13 @@ function dedupeSqliteSessionStateDeletePlans( deduped.set(plan.sessionId, plan); continue; } - if (existing.content !== plan.content || existing.reason !== plan.reason) { + if ( + existing.agentId !== plan.agentId || + existing.archiveDirectory !== plan.archiveDirectory || + existing.databasePath !== plan.databasePath || + existing.reason !== plan.reason || + !sqliteSessionStateDeleteSnapshotsEqual(existing.snapshot, plan.snapshot) + ) { throw new Error(`Conflicting SQLite transcript archive plans for ${plan.sessionId}`); } if (!existing.archiveTranscript && plan.archiveTranscript) { diff --git a/src/config/sessions/session-accessor.sqlite-archive.worker.test.ts b/src/config/sessions/session-accessor.sqlite-archive.worker.test.ts new file mode 100644 index 000000000000..25ba39dd740c --- /dev/null +++ b/src/config/sessions/session-accessor.sqlite-archive.worker.test.ts @@ -0,0 +1,811 @@ +// SQLite transcript archive worker tests cover off-main execution and snapshot fencing. +import { createHash, randomBytes } from "node:crypto"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; +import { performance } from "node:perf_hooks"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { recordAcpParentStreamEvents } from "../../agents/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"; +import { + closeOpenClawAgentDatabasesForTest, + openOpenClawAgentDatabase, + runOpenClawAgentWriteTransaction, +} 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 { + deleteSessionEntryLifecycle, + loadSessionEntry, + loadTranscriptEvents, + replaceSessionEntry, +} from "./session-accessor.js"; +import { materializeSqliteSessionStateDeletePlans } from "./session-accessor.sqlite-archive.js"; +import { materializeSqliteTranscriptArchiveInWorker } from "./session-accessor.sqlite-archive.worker.js"; +import { + deleteMaterializedSqliteSessionStatePlans, + planSqliteSessionStateDeleteIfUnreferenced, +} from "./session-accessor.sqlite-lifecycle-state.js"; +import { touchTranscriptMutationInTransaction } from "./session-accessor.sqlite-transcript-state.js"; +import { replaceSqliteTranscriptEvents } from "./session-accessor.sqlite.js"; +import { resolveSqliteTargetFromSessionStorePath } from "./session-sqlite-target.js"; + +type TestTranscriptEvent = { + id: string; + [key: string]: unknown; +}; + +describe("SQLite transcript archive worker", () => { + let tempDir: string; + let storePath: string; + + beforeEach(() => { + tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "openclaw-sqlite-archive-worker-")); + storePath = path.join(tempDir, "agents", "main", "sessions", "sessions.json"); + }); + + afterEach(() => { + closeOpenClawAgentDatabasesForTest(); + fs.rmSync(tempDir, { recursive: true, force: true }); + }); + + it("keeps the event loop responsive while a transcript archive is built", async () => { + const sessionId = "off-main-archive-session"; + const sessionKey = "agent:main:off-main-archive"; + await replaceSessionEntry({ sessionKey, storePath }, { sessionId, updatedAt: Date.now() }); + const events = Array.from({ length: 64 }, (_, index) => + createTranscriptEvent( + `${sessionId}-${index}`, + index === 0 + ? `first: 你好\n${randomBytes(576 * 1024).toString("base64")}` + : index === 63 + ? `last: 🦞\n${randomBytes(576 * 1024).toString("base64")}` + : `${index}:${randomBytes(576 * 1024).toString("base64")}`, + ), + ); + await replaceSqliteTranscriptEvents({ sessionKey, sessionId, storePath }, events); + + const heartbeatTimes = [performance.now()]; + const heartbeat = setInterval(() => { + heartbeatTimes.push(performance.now()); + }, 5); + let materialized: Awaited>; + try { + const database = openLifecycleTestDatabase(storePath); + const plan = planArchiveWorker(database, path.dirname(storePath), sessionId); + materialized = await materializeSqliteSessionStateDeletePlans([plan]); + } finally { + heartbeatTimes.push(performance.now()); + clearInterval(heartbeat); + } + + const heartbeatGaps: number[] = []; + for (let index = 1; index < heartbeatTimes.length; index += 1) { + const current = heartbeatTimes[index]; + const previous = heartbeatTimes[index - 1]; + if (current !== undefined && previous !== undefined) { + heartbeatGaps.push(current - previous); + } + } + expect(heartbeatTimes.length - 2).toBeGreaterThan(5); + expect(Math.max(...heartbeatGaps)).toBeLessThan(150); + expect(materialized).toHaveLength(1); + const archivedPath = materialized[0]?.archivedTranscript?.archivedPath; + expect(archivedPath).toBeTruthy(); + const expectedContent = `${events.map((event) => JSON.stringify(event)).join("\n")}\n`; + const archivedContent = readSessionArchiveContentSync(archivedPath ?? ""); + expect(Buffer.byteLength(archivedContent)).toBe(Buffer.byteLength(expectedContent)); + expect(sha256(archivedContent)).toBe(sha256(expectedContent)); + const archiveLines = readArchiveLines(archivedPath); + 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 () => { + const sessionId = "durable-delete-session"; + const sessionKey = "agent:main:durable-delete"; + await replaceSessionEntry( + { sessionKey, storePath }, + { + sessionId, + updatedAt: Date.now(), + }, + ); + await replaceSqliteTranscriptEvents({ sessionKey, sessionId, storePath }, [ + 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 = materializeSqliteTranscriptArchiveInWorker( + 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, + target: { + canonicalKey: sessionKey, + storeKeys: [sessionKey], + }, + }); + expect(result.deleted).toBe(true); + expect(result.archivedTranscripts.map((archive) => archive.archivedPath)).toEqual([ + archivedPath, + ]); + }); + + it("archives a logical agent transcript through the exact database's physical owner", async () => { + const sharedDatabasePath = path.join(tempDir, "shared.sqlite"); + const mainSessionId = "shared-physical-owner-main-session"; + const mainSessionKey = "agent:main:shared-physical-owner-main"; + const opsSessionId = "shared-physical-owner-ops-session"; + const opsSessionKey = "agent:ops:shared-physical-owner-ops"; + const mainScope = { + agentId: "main", + defaultAgentId: "main", + sessionId: mainSessionId, + sessionKey: mainSessionKey, + storePath: sharedDatabasePath, + }; + const opsScope = { + agentId: "ops", + defaultAgentId: "main", + sessionId: opsSessionId, + sessionKey: opsSessionKey, + storePath: sharedDatabasePath, + }; + const mainEvent = createTranscriptEvent(mainSessionId, "keep physical-owner transcript"); + const opsEvent = createTranscriptEvent(opsSessionId, "archive logical-owner transcript"); + + await replaceSessionEntry(mainScope, { sessionId: mainSessionId, updatedAt: Date.now() }); + await replaceSqliteTranscriptEvents(mainScope, [mainEvent]); + await replaceSessionEntry(opsScope, { sessionId: opsSessionId, updatedAt: Date.now() }); + await replaceSqliteTranscriptEvents(opsScope, [opsEvent]); + + const opsTarget = resolveSqliteTargetFromSessionStorePath(sharedDatabasePath, { + agentId: opsScope.agentId, + defaultAgentId: opsScope.defaultAgentId, + }); + const database = openLifecycleTestDatabase(sharedDatabasePath); + expect(opsTarget).toMatchObject({ + agentId: "main", + path: sharedDatabasePath, + shared: true, + }); + 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 materialized = await materializeSqliteSessionStateDeletePlans([plan]); + const archivedPath = materialized[0]?.archivedTranscript?.archivedPath; + expect(readArchiveLines(archivedPath ?? undefined)).toEqual([JSON.stringify(opsEvent)]); + + deleteMaterializedPlans(database, materialized, opsSessionKey); + + await expect(loadTranscriptEvents(opsScope)).resolves.toEqual([]); + await expect(loadTranscriptEvents(mainScope)).resolves.toEqual([mainEvent]); + expect(loadSessionEntry(mainScope)).toMatchObject({ sessionId: mainSessionId }); + }); + + it("rejects transcript changes between deletion planning and the worker snapshot", async () => { + const sessionId = "changed-before-worker-snapshot"; + const scope = { + sessionKey: "agent:main:changed-before-worker-snapshot", + sessionId, + storePath, + }; + const original = createTranscriptEvent(sessionId, "original transcript"); + await replaceSqliteTranscriptEvents(scope, [original]); + const database = openLifecycleTestDatabase(storePath); + const plan = planArchiveWorker(database, path.dirname(storePath), sessionId); + + await replaceSqliteTranscriptEvents(scope, [ + original, + createTranscriptEvent("concurrent-event", "concurrent append"), + ]); + + await expect(materializeSqliteSessionStateDeletePlans([plan])).rejects.toThrow( + `SQLite session state changed before archive materialization for ${sessionId}`, + ); + await expect(loadTranscriptEvents(scope)).resolves.toHaveLength(2); + const archiveDirectory = path.dirname(storePath); + const archiveNames = fs.existsSync(archiveDirectory) ? fs.readdirSync(archiveDirectory) : []; + expect(archiveNames.filter((entry) => entry.startsWith(`${sessionId}.jsonl.deleted.`))).toEqual( + [], + ); + }); + + it("rejects deduped plans with different transcript snapshots", async () => { + const sessionId = "conflicting-plan-snapshots"; + await replaceSqliteTranscriptEvents( + { sessionKey: "agent:main:conflicting-plan-snapshots", sessionId, storePath }, + [createTranscriptEvent(sessionId, "original transcript")], + ); + const database = openLifecycleTestDatabase(storePath); + const plan = planArchiveWorker(database, path.dirname(storePath), sessionId); + const conflictingPlan = { + ...plan, + snapshot: { + ...plan.snapshot, + transcriptUpdatedAt: (plan.snapshot.transcriptUpdatedAt ?? 0) + 1, + }, + }; + + await expect(materializeSqliteSessionStateDeletePlans([plan, conflictingPlan])).rejects.toThrow( + `Conflicting SQLite transcript archive plans for ${sessionId}`, + ); + }); + + it("rejects the first append after planning an empty transcript", async () => { + const sessionId = "empty-then-appended-transcript"; + const scope = { + sessionKey: "agent:main:empty-then-appended-transcript", + sessionId, + storePath, + }; + await replaceSessionEntry(scope, { sessionId, updatedAt: Date.now() }); + const database = openLifecycleTestDatabase(storePath); + const plan = planArchiveWorker(database, path.dirname(storePath), sessionId); + expect(plan.snapshot.lastSeq).toBeNull(); + + await replaceSqliteTranscriptEvents(scope, [ + createTranscriptEvent(sessionId, "first concurrent append"), + ]); + + await expect(materializeSqliteSessionStateDeletePlans([plan])).rejects.toThrow( + `SQLite session state changed before archive materialization for ${sessionId}`, + ); + 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 replaceSqliteTranscriptEvents(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 replaceSqliteTranscriptEvents(recoveryScope, [ + createTranscriptEvent(recoverySessionId, "archive after queued failure"), + ]); + const recoveryPlan = planArchiveWorker(database, path.dirname(storePath), recoverySessionId); + + const failedArchive = materializeSqliteSessionStateDeletePlans([plan]); + const recoveredArchive = materializeSqliteSessionStateDeletePlans([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"; + const scope = { sessionKey, sessionId, storePath }; + await replaceSessionEntry(scope, { sessionId, updatedAt: Date.now() }); + await replaceSqliteTranscriptEvents(scope, [ + { + type: "message", + id: "archive-worker-lifecycle-failure-message", + parentId: null, + message: { + role: "user", + content: [{ type: "text", text: "preserve every lifecycle row" }], + }, + timestamp: Date.now(), + } as unknown as TestTranscriptEvent, + ]); + appendSqliteTrajectoryRuntimeEvents({ sessionId, storePath }, [ + createTestTrajectoryEvent(sessionId), + ]); + const database = openLifecycleTestDatabase(storePath); + recordAcpParentStreamEvents({ + agentId: database.agentId, + path: database.path, + sessionId, + runId: "archive-worker-lifecycle-failure-run", + events: [{ event: { type: "output", text: "preserve ACP state" }, createdAt: Date.now() }], + }); + const db = getNodeSqliteKysely(database.db); + const readLifecycleCounts = () => ({ + acp: executeSqliteQuerySync( + database.db, + db.selectFrom("acp_parent_stream_events").select("seq").where("session_id", "=", sessionId), + ).rows.length, + fts: executeSqliteQuerySync( + database.db, + db + .selectFrom("session_transcript_fts") + .select("session_id") + .where("session_id", "=", sessionId), + ).rows.length, + indexState: executeSqliteQuerySync( + database.db, + db + .selectFrom("session_transcript_index_state") + .select("session_id") + .where("session_id", "=", sessionId), + ).rows.length, + nodes: executeSqliteQuerySync( + database.db, + db + .selectFrom("session_nodes") + .select("current_session_id") + .where("current_session_id", "=", sessionId), + ).rows.length, + rewriteWatermarks: executeSqliteQuerySync( + database.db, + db + .selectFrom("transcript_rewrite_watermarks") + .select("session_id") + .where("session_id", "=", sessionId), + ).rows.length, + trajectory: executeSqliteQuerySync( + database.db, + db + .selectFrom("trajectory_runtime_events") + .select("seq") + .where("session_id", "=", sessionId), + ).rows.length, + transcript: executeSqliteQuerySync( + database.db, + db.selectFrom("transcript_events").select("seq").where("session_id", "=", sessionId), + ).rows.length, + windows: executeSqliteQuerySync( + database.db, + db.selectFrom("session_windows").select("session_id").where("session_id", "=", sessionId), + ).rows.length, + }); + const before = readLifecycleCounts(); + + await expect( + deleteSessionEntryLifecycle({ + archiveTranscript: true, + storePath, + target: { canonicalKey: sessionKey, storeKeys: [sessionKey] }, + }), + ).rejects.toThrow("Cannot archive SQLite transcript outside"); + + expect(loadSessionEntry({ sessionKey, storePath })?.sessionId).toBe(sessionId); + await expect(loadTranscriptEvents(scope)).resolves.toHaveLength(1); + expect(readLifecycleCounts()).toEqual(before); + expect(before).toEqual({ + acp: 1, + fts: 1, + indexState: 1, + nodes: 1, + rewriteWatermarks: 1, + trajectory: 1, + transcript: 1, + windows: 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"; + await replaceSqliteTranscriptEvents({ sessionKey, sessionId, storePath }, [ + createTranscriptEvent(sessionId, "archived snapshot"), + ]); + const database = openLifecycleTestDatabase(storePath); + const db = getNodeSqliteKysely(database.db); + const plan = planSqliteSessionStateDeleteIfUnreferenced({ + archiveDirectory: path.dirname(storePath), + database, + referencedSessionIds: new Set(), + sessionId, + }); + if (!plan) { + throw new Error("expected an unreferenced SQLite transcript delete plan"); + } + const materialized = await materializeSqliteSessionStateDeletePlans([plan]); + + appendTranscriptEvent(database, sessionId); + + expect(() => deleteMaterializedPlans(database, materialized, sessionKey)).toThrow( + `SQLite session state changed before deletion for ${sessionId}`, + ); + expect( + executeSqliteQuerySync( + database.db, + db.selectFrom("transcript_events").select("seq").where("session_id", "=", sessionId), + ).rows, + ).toHaveLength(2); + }); + + it.each(["rewrite generation", "transcript mutation watermark", "window metadata"] as const)( + "keeps rows when the %s changes after archive materialization", + async (kind) => { + const sessionId = `stale-${ + kind === "rewrite generation" + ? "generation" + : kind === "transcript mutation watermark" + ? "watermark" + : "window" + }-snapshot`; + const sessionKey = `agent:main:${sessionId}`; + await replaceSqliteTranscriptEvents({ sessionKey, sessionId, storePath }, [ + createTranscriptEvent(sessionId, "archived transcript"), + ]); + const database = openLifecycleTestDatabase(storePath); + const db = getNodeSqliteKysely(database.db); + const plan = planArchiveWorker(database, path.dirname(storePath), sessionId); + expect(plan.snapshot.generation).not.toBeNull(); + expect(plan.snapshot.sessionUpdatedAt).not.toBeNull(); + expect(plan.snapshot.transcriptUpdatedAt).not.toBeNull(); + const materialized = await materializeSqliteSessionStateDeletePlans([plan]); + + if (kind === "rewrite generation") { + executeSqliteQuerySync( + database.db, + db + .updateTable("transcript_rewrite_watermarks") + .set({ + generation: `${plan.snapshot.generation ?? "missing"}-changed`, + updated_at: Date.now(), + }) + .where("session_id", "=", sessionId), + ); + } else if (kind === "transcript mutation watermark") { + executeSqliteQuerySync( + database.db, + db + .updateTable("session_windows") + .set({ + transcript_updated_at: (plan.snapshot.transcriptUpdatedAt ?? 0) + 1, + }) + .where("session_id", "=", sessionId), + ); + } else { + executeSqliteQuerySync( + database.db, + db + .updateTable("session_windows") + .set({ + updated_at: (plan.snapshot.sessionUpdatedAt ?? 0) + 1, + }) + .where("session_id", "=", sessionId), + ); + } + + expect(() => deleteMaterializedPlans(database, materialized, sessionKey)).toThrow( + `SQLite session state changed before deletion for ${sessionId}`, + ); + expect( + executeSqliteQuerySync( + database.db, + db.selectFrom("transcript_events").select("seq").where("session_id", "=", sessionId), + ).rows, + ).toHaveLength(1); + }, + ); + + it("keeps rows when a non-archive delete plan becomes stale", async () => { + const sessionId = "stale-non-archive-snapshot-session"; + const sessionKey = "agent:main:stale-non-archive-snapshot"; + await replaceSqliteTranscriptEvents({ sessionKey, sessionId, storePath }, [ + createTranscriptEvent(sessionId, "planned transcript"), + ]); + const database = openLifecycleTestDatabase(storePath); + const db = getNodeSqliteKysely(database.db); + const plan = planSqliteSessionStateDeleteIfUnreferenced({ + archiveDirectory: path.dirname(storePath), + archiveTranscript: false, + database, + referencedSessionIds: new Set(), + sessionId, + }); + if (!plan) { + throw new Error("expected an unreferenced SQLite transcript delete plan"); + } + const materialized = await materializeSqliteSessionStateDeletePlans([plan]); + + appendTranscriptEvent(database, sessionId); + + expect(() => deleteMaterializedPlans(database, materialized, sessionKey)).toThrow( + `SQLite session state changed before deletion for ${sessionId}`, + ); + expect( + executeSqliteQuerySync( + database.db, + db.selectFrom("transcript_events").select("seq").where("session_id", "=", sessionId), + ).rows, + ).toHaveLength(2); + }); + + it.each(["trajectory", "ACP parent-stream"] as const)( + "keeps rows when %s state changes after archive materialization", + async (kind) => { + const sessionId = `stale-${kind === "trajectory" ? "trajectory" : "acp"}-snapshot-session`; + const sessionKey = `agent:main:${sessionId}`; + await replaceSqliteTranscriptEvents({ sessionKey, sessionId, storePath }, [ + createTranscriptEvent(sessionId, "archived transcript"), + ]); + const database = openLifecycleTestDatabase(storePath); + const db = getNodeSqliteKysely(database.db); + const plan = planArchiveWorker(database, path.dirname(storePath), sessionId); + const materialized = await materializeSqliteSessionStateDeletePlans([plan]); + + if (kind === "trajectory") { + appendSqliteTrajectoryRuntimeEvents({ sessionId, storePath }, [ + createTestTrajectoryEvent(sessionId), + ]); + } else { + recordAcpParentStreamEvents({ + agentId: database.agentId, + path: database.path, + sessionId, + runId: "run-1", + events: [{ event: { type: "output", text: "concurrent" }, createdAt: Date.now() }], + }); + } + + expect(() => deleteMaterializedPlans(database, materialized, sessionKey)).toThrow( + `SQLite session state changed before deletion for ${sessionId}`, + ); + const rows = + kind === "trajectory" + ? executeSqliteQuerySync( + database.db, + db + .selectFrom("trajectory_runtime_events") + .select("seq") + .where("session_id", "=", sessionId), + ).rows + : executeSqliteQuerySync( + database.db, + db + .selectFrom("acp_parent_stream_events") + .select("seq") + .where("session_id", "=", sessionId), + ).rows; + 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 replaceSqliteTranscriptEvents( + { 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 = materializeSqliteTranscriptArchiveInWorker( + 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 replaceSqliteTranscriptEvents({ 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 = materializeSqliteTranscriptArchiveInWorker( + 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 { + return JSON.parse(createTranscriptEventLine(sessionId, content)) as TestTranscriptEvent; +} + +function createTranscriptEventLine(sessionId: string, content: string): string { + return JSON.stringify({ type: "session", id: sessionId, content }); +} + +function createTestTrajectoryEvent(sessionId: string): TrajectoryEvent { + return { + traceSchema: "openclaw-trajectory", + schemaVersion: 1, + traceId: sessionId, + source: "runtime", + type: "test.concurrent-delete", + ts: "2026-07-22T00:00:00.000Z", + seq: 1, + sessionId, + }; +} + +function readArchiveLines(archivePath: string | undefined): string[] { + expect(archivePath).toBeTruthy(); + return readSessionArchiveContentSync(archivePath ?? "") + .trim() + .split("\n"); +} + +function sha256(content: string): string { + return createHash("sha256").update(content).digest("hex"); +} + +function openLifecycleTestDatabase(storePath: string) { + const target = resolveSqliteTargetFromSessionStorePath(storePath); + if (!target.path) { + throw new Error(`Could not resolve SQLite database path for ${storePath}`); + } + return openOpenClawAgentDatabase({ + agentId: target.agentId ?? "main", + path: target.path, + }); +} + +function planArchiveWorker( + database: ReturnType, + archiveDirectory: string, + sessionId: string, +) { + const plan = planSqliteSessionStateDeleteIfUnreferenced({ + archiveDirectory, + database, + referencedSessionIds: new Set(), + sessionId, + }); + if (!plan) { + throw new Error(`expected an archive plan for ${sessionId}`); + } + return plan; +} + +function appendTranscriptEvent( + database: ReturnType, + sessionId: string, +): void { + runOpenClawAgentWriteTransaction( + (transactionDb) => { + const db = getNodeSqliteKysely(transactionDb.db); + executeSqliteQuerySync( + transactionDb.db, + db.insertInto("transcript_events").values({ + session_id: sessionId, + seq: 1, + event_json: createTranscriptEventLine("concurrent-event", "concurrent append"), + created_at: Date.now(), + }), + ); + touchTranscriptMutationInTransaction(transactionDb, sessionId); + }, + { agentId: database.agentId, path: database.path }, + ); +} + +function deleteMaterializedPlans( + database: ReturnType, + plans: Parameters[1], + excludedSessionKey: string, +): void { + runOpenClawAgentWriteTransaction( + (transactionDb) => + deleteMaterializedSqliteSessionStatePlans( + transactionDb, + plans, + undefined, + new Set([excludedSessionKey]), + ), + { agentId: database.agentId, path: database.path }, + ); +} diff --git a/src/config/sessions/session-accessor.sqlite-archive.worker.ts b/src/config/sessions/session-accessor.sqlite-archive.worker.ts new file mode 100644 index 000000000000..cf61990fe6e1 --- /dev/null +++ b/src/config/sessions/session-accessor.sqlite-archive.worker.ts @@ -0,0 +1,171 @@ +/** Worker entrypoint for SQLite transcript archive materialization off the gateway event loop. */ +import { parentPort, workerData } from "node:worker_threads"; +import { executeSqliteQuerySync, getNodeSqliteKysely } from "../../infra/kysely-sync.js"; +import { withOpenClawAgentDatabaseReadOnly } from "../../state/openclaw-agent-db-readonly.js"; +import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js"; +import { + sqliteSessionStateDeleteSnapshotsEqual, + type SqliteSessionStateDeleteSnapshot, + type SqliteTranscriptArchiveWorkerMessage, + type SqliteTranscriptArchiveWorkerPlan, + type SqliteTranscriptArchiveWorkerResult, + writeSqliteTranscriptArchive, +} from "./session-accessor.sqlite-archive.js"; +import { readSqliteSessionStateDeleteSnapshot } from "./session-accessor.sqlite-delete-snapshot.js"; +import { serializeJsonlLines } from "./transcript-jsonl.js"; + +type TranscriptArchiveDatabase = Pick; + +function isSqliteTranscriptArchiveWorkerData(value: unknown): boolean { + return ( + Boolean(value) && + typeof value === "object" && + !Array.isArray(value) && + (value as { type?: unknown }).type === "sqlite-transcript-archive-v1" + ); +} + +function parseSessionStateDeleteSnapshot(value: unknown): SqliteSessionStateDeleteSnapshot | null { + if (!value || typeof value !== "object" || Array.isArray(value)) { + return null; + } + const snapshot = value as Record; + if ( + typeof snapshot.acpParentStreamEventCount !== "number" || + (snapshot.generation !== null && typeof snapshot.generation !== "string") || + (snapshot.lastSeq !== null && typeof snapshot.lastSeq !== "number") || + (snapshot.sessionUpdatedAt !== null && typeof snapshot.sessionUpdatedAt !== "number") || + (snapshot.trajectoryLastSeq !== null && typeof snapshot.trajectoryLastSeq !== "number") || + (snapshot.transcriptUpdatedAt !== null && typeof snapshot.transcriptUpdatedAt !== "number") + ) { + return null; + } + return { + acpParentStreamEventCount: snapshot.acpParentStreamEventCount, + generation: snapshot.generation, + lastSeq: snapshot.lastSeq, + sessionUpdatedAt: snapshot.sessionUpdatedAt, + trajectoryLastSeq: snapshot.trajectoryLastSeq, + transcriptUpdatedAt: snapshot.transcriptUpdatedAt, + }; +} + +function parseWorkerPlans(value: unknown): SqliteTranscriptArchiveWorkerPlan[] | 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: SqliteTranscriptArchiveWorkerPlan[] = []; + for (const planValue of plans) { + if (!planValue || typeof planValue !== "object" || Array.isArray(planValue)) { + return undefined; + } + const plan = planValue as Record; + const snapshot = parseSessionStateDeleteSnapshot(plan.snapshot); + if ( + typeof plan.agentId !== "string" || + typeof plan.archiveDirectory !== "string" || + typeof plan.databasePath !== "string" || + (plan.reason !== "deleted" && plan.reason !== "reset") || + typeof plan.sessionId !== "string" || + !snapshot + ) { + return undefined; + } + parsed.push({ + agentId: plan.agentId, + archiveDirectory: plan.archiveDirectory, + databasePath: plan.databasePath, + reason: plan.reason, + sessionId: plan.sessionId, + snapshot, + }); + } + return parsed; +} + +function readTranscriptArchiveContent( + database: import("node:sqlite").DatabaseSync, + sessionId: string, +): string { + const db = getNodeSqliteKysely(database); + const lines = executeSqliteQuerySync( + database, + db + .selectFrom("transcript_events") + .select("event_json") + .where("session_id", "=", sessionId) + .orderBy("seq", "asc"), + ).rows.map((row) => row.event_json); + return serializeJsonlLines(lines); +} + +export function materializeSqliteTranscriptArchiveInWorker( + plan: SqliteTranscriptArchiveWorkerPlan, +): SqliteTranscriptArchiveWorkerResult { + const opened = withOpenClawAgentDatabaseReadOnly( + (database) => { + let transactionOpen = false; + try { + // sqlite-allow-raw: metadata and transcript rows must come from one read snapshot. + database.db.exec("BEGIN"); + transactionOpen = true; + const snapshot = readSqliteSessionStateDeleteSnapshot(database.db, plan.sessionId); + if (!sqliteSessionStateDeleteSnapshotsEqual(snapshot, plan.snapshot)) { + throw new Error( + `SQLite session state changed before archive materialization for ${plan.sessionId}`, + ); + } + const content = readTranscriptArchiveContent(database.db, plan.sessionId); + database.db.exec("COMMIT"); // sqlite-allow-raw: closes the consistent read snapshot. + transactionOpen = false; + return { content, snapshot }; + } catch (error) { + if (transactionOpen) { + database.db.exec("ROLLBACK"); // sqlite-allow-raw: releases a failed read snapshot. + } + throw error; + } + }, + { agentId: plan.agentId, path: plan.databasePath }, + ); + if (!opened.found) { + throw new Error( + `Cannot archive SQLite transcript ${plan.sessionId}: ${opened.reason.replaceAll("-", " ")}`, + ); + } + const { content } = opened.value; + const archivedPath = + content.length > 0 + ? writeSqliteTranscriptArchive({ + archiveDirectory: plan.archiveDirectory, + content, + reason: plan.reason, + sessionId: plan.sessionId, + }) + : null; + return { archivedPath, sessionId: plan.sessionId }; +} + +function runWorkerPort( + port: NonNullable, + plans: readonly SqliteTranscriptArchiveWorkerPlan[], +): void { + const results = plans.map((plan) => materializeSqliteTranscriptArchiveInWorker(plan)); + port.postMessage({ type: "done", results } satisfies SqliteTranscriptArchiveWorkerMessage); + 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"); + } + runWorkerPort(parentPort, plans); +} diff --git a/src/config/sessions/session-accessor.sqlite-cleanup-race.test.ts b/src/config/sessions/session-accessor.sqlite-cleanup-race.test.ts index d1fff1431de9..52b0c28a1b0c 100644 --- a/src/config/sessions/session-accessor.sqlite-cleanup-race.test.ts +++ b/src/config/sessions/session-accessor.sqlite-cleanup-race.test.ts @@ -1,4 +1,3 @@ -import fs from "node:fs"; import path from "node:path"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { useAutoCleanupTempDirTracker } from "../../../test/helpers/temp-dir.js"; @@ -19,6 +18,28 @@ import { replaceSqliteTranscriptEvents } from "./session-accessor.sqlite.js"; import { resolveSqliteTargetFromSessionStorePath } from "./session-sqlite-target.js"; import type { SessionEntry } from "./types.js"; +const archiveMaterializationHook = vi.hoisted(() => ({ + beforeMaterialize: undefined as (() => Promise) | undefined, + afterMaterialize: undefined as (() => void) | undefined, +})); + +// Place test mutations after the real Worker finishes but before cleanup opens +// its final transaction, without relying on cross-isolate filesystem timing. +vi.mock("./session-accessor.sqlite-archive.js", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + materializeSqliteSessionStateDeletePlans: async ( + ...args: Parameters + ) => { + await archiveMaterializationHook.beforeMaterialize?.(); + const result = await actual.materializeSqliteSessionStateDeletePlans(...args); + archiveMaterializationHook.afterMaterialize?.(); + return result; + }, + }; +}); + const tempDirs = useAutoCleanupTempDirTracker(afterEach); describe("SQLite lifecycle cleanup races", () => { @@ -31,6 +52,8 @@ describe("SQLite lifecycle cleanup races", () => { }); afterEach(() => { + archiveMaterializationHook.beforeMaterialize = undefined; + archiveMaterializationHook.afterMaterialize = undefined; closeOpenClawAgentDatabasesForTest(); }); @@ -357,18 +380,13 @@ describe("SQLite lifecycle cleanup races", () => { expect(planned.deletePlans).toHaveLength(1); const refreshedEntry = { label: "refreshed", sessionId, updatedAt: now + 1 }; - const originalRenameSync = fs.renameSync; let refreshed = false; - const renameSpy = vi.spyOn(fs, "renameSync").mockImplementation((...args) => { - const result = originalRenameSync(...args); - if (!refreshed && String(args[1]).includes(`${sessionId}.jsonl.deleted.`)) { - refreshed = true; - database.db - .prepare("UPDATE session_nodes SET entry_json = ?, updated_at = ? WHERE session_key = ?") - .run(JSON.stringify(refreshedEntry), refreshedEntry.updatedAt, sessionKey); - } - return result; - }); + archiveMaterializationHook.afterMaterialize = () => { + refreshed = true; + database.db + .prepare("UPDATE session_nodes SET entry_json = ?, updated_at = ? WHERE session_key = ?") + .run(JSON.stringify(refreshedEntry), refreshedEntry.updatedAt, sessionKey); + }; try { await expect( @@ -381,7 +399,7 @@ describe("SQLite lifecycle cleanup races", () => { }), ).rejects.toThrow("SQLite lifecycle cleanup entry changed"); } finally { - renameSpy.mockRestore(); + archiveMaterializationHook.afterMaterialize = undefined; } expect(refreshed).toBe(true); @@ -391,6 +409,65 @@ describe("SQLite lifecycle cleanup races", () => { ]); }); + it("releases the store writer while a transcript archive is materialized", async () => { + const deletedKey = "agent:main:cleanup-race-deleted"; + const deletedSessionId = "cleanup-race-deleted-session"; + const writerKey = "agent:main:cleanup-race-writer"; + await replaceSessionEntry( + { sessionKey: deletedKey, storePath }, + { sessionId: deletedSessionId, updatedAt: Date.now() }, + ); + await replaceSqliteTranscriptEvents( + { sessionKey: deletedKey, sessionId: deletedSessionId, storePath }, + [ + { + type: "session", + id: deletedSessionId, + content: "archive while another writer progresses", + }, + ], + ); + await replaceSessionEntry( + { sessionKey: writerKey, storePath }, + { sessionId: "cleanup-race-writer-session", updatedAt: Date.now() }, + ); + + let markMaterializationStarted: () => void = () => undefined; + const materializationStarted = new Promise((resolve) => { + markMaterializationStarted = resolve; + }); + let releaseMaterialization: () => void = () => undefined; + const materializationGate = new Promise((resolve) => { + releaseMaterialization = resolve; + }); + archiveMaterializationHook.beforeMaterialize = async () => { + markMaterializationStarted(); + await materializationGate; + }; + + const deletion = deleteSessionEntryLifecycle({ + archiveTranscript: true, + storePath, + target: { canonicalKey: deletedKey, storeKeys: [deletedKey] }, + }); + await materializationStarted; + const writer = replaceSessionEntry( + { sessionKey: writerKey, storePath }, + { sessionId: "cleanup-race-writer-session", label: "progressed", updatedAt: Date.now() }, + ); + const progressedDuringMaterialization = await Promise.race([ + writer.then(() => true), + new Promise((resolve) => { + setTimeout(() => resolve(false), 500); + }), + ]); + releaseMaterialization(); + + await expect(deletion).resolves.toMatchObject({ deleted: true }); + await expect(writer).resolves.toMatchObject({ label: "progressed" }); + expect(progressedDuringMaterialization).toBe(true); + }); + it("retains unplanned historical windows behind a placeholder node", async () => { const sessionKey = "agent:main:unplanned-history"; const currentEntry: SessionEntry = { diff --git a/src/config/sessions/session-accessor.sqlite-delete-snapshot.ts b/src/config/sessions/session-accessor.sqlite-delete-snapshot.ts new file mode 100644 index 000000000000..692c6110d14e --- /dev/null +++ b/src/config/sessions/session-accessor.sqlite-delete-snapshot.ts @@ -0,0 +1,71 @@ +import { executeSqliteQueryTakeFirstSync, getNodeSqliteKysely } from "../../infra/kysely-sync.js"; +import type { DB as OpenClawAgentKyselyDatabase } from "../../state/openclaw-agent-db.generated.js"; +import type { SqliteSessionStateDeleteSnapshot } from "./session-accessor.sqlite-archive.js"; + +type SessionStateDeleteSnapshotDatabase = Pick< + OpenClawAgentKyselyDatabase, + | "acp_parent_stream_events" + | "session_windows" + | "trajectory_runtime_events" + | "transcript_events" + | "transcript_rewrite_watermarks" +>; + +function normalizeOptionalSqliteNumber(value: number | bigint | null | undefined): number | null { + return value === null || value === undefined ? null : Number(value); +} + +/** Captures the owner window and canonical child state writable outside the lifecycle queue. */ +export function readSqliteSessionStateDeleteSnapshot( + database: import("node:sqlite").DatabaseSync, + sessionId: string, +): SqliteSessionStateDeleteSnapshot { + const db = getNodeSqliteKysely(database); + const window = executeSqliteQueryTakeFirstSync( + database, + db + .selectFrom("session_windows") + .select(["transcript_updated_at", "updated_at"]) + .where("session_id", "=", sessionId), + ); + const rewriteWatermark = executeSqliteQueryTakeFirstSync( + database, + db + .selectFrom("transcript_rewrite_watermarks") + .select("generation") + .where("session_id", "=", sessionId), + ); + const lastEvent = executeSqliteQueryTakeFirstSync( + database, + db + .selectFrom("transcript_events") + .select("seq") + .where("session_id", "=", sessionId) + .orderBy("seq", "desc") + .limit(1), + ); + const lastTrajectory = executeSqliteQueryTakeFirstSync( + database, + db + .selectFrom("trajectory_runtime_events") + .select("seq") + .where("session_id", "=", sessionId) + .orderBy("seq", "desc") + .limit(1), + ); + const acpParentStream = executeSqliteQueryTakeFirstSync( + database, + db + .selectFrom("acp_parent_stream_events") + .select((eb) => eb.fn.countAll().as("event_count")) + .where("session_id", "=", sessionId), + ); + return { + acpParentStreamEventCount: normalizeOptionalSqliteNumber(acpParentStream?.event_count) ?? 0, + generation: rewriteWatermark?.generation ?? null, + lastSeq: lastEvent?.seq ?? 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-entry.ts b/src/config/sessions/session-accessor.sqlite-entry.ts index ab8104b84624..df43dcc826e2 100644 --- a/src/config/sessions/session-accessor.sqlite-entry.ts +++ b/src/config/sessions/session-accessor.sqlite-entry.ts @@ -46,7 +46,7 @@ import { emitCommittedSessionIdentityDiff } from "./session-accessor.sqlite-iden import type { SqliteSessionEntryMaintenancePlan } from "./session-accessor.sqlite-lifecycle-types.js"; import { applySqliteSessionEntryMaintenance, - finalizeSqliteSessionEntryMaintenancePlansBestEffort, + finalizeSqliteSessionEntryMaintenancePlansAfterWriterReleaseBestEffort, } from "./session-accessor.sqlite-maintenance.js"; import { createFallbackSessionEntry, @@ -535,13 +535,13 @@ async function patchSqliteSessionEntrySnapshot( params: SqliteSessionEntrySnapshotPatchParams, ): Promise { const { options, resolved, sessionKey } = params; - return await runExclusiveSqliteSessionWrite(resolved, async () => { + const committed = await runExclusiveSqliteSessionWrite(resolved, async () => { const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); const prepared = params.readSnapshot(database); const existing = params.existingEntry(prepared); const writeBase = existing ?? options.fallbackEntry; if (!writeBase) { - return null; + return { maintenancePlans: [], result: null }; } const patch = await params.update(cloneSessionEntry(writeBase), { existingEntry: existing ? cloneSessionEntry(existing) : undefined, @@ -601,14 +601,20 @@ async function patchSqliteSessionEntrySnapshot( result = cloneSessionEntry(next); }, toDatabaseOptions(resolved)); emitCommittedSessionIdentityDiff(previousIdentity, currentIdentity); - finalizeSqliteSessionEntryMaintenancePlansBestEffort(resolved, maintenancePlans); - kickSessionHistoryDiskBudgetMaintenance({ - ...(resolved.agentId ? { agentId: resolved.agentId } : {}), - storePath: params.storePath, - ...(options.maintenanceConfig ? { maintenanceConfig: options.maintenanceConfig } : {}), - }); - return result; + return { maintenancePlans, result }; }); + // Worker materialization runs after the initial write releases the lane; + // final deletion reacquires it and revalidates every planned row. + await finalizeSqliteSessionEntryMaintenancePlansAfterWriterReleaseBestEffort( + resolved, + committed.maintenancePlans, + ); + kickSessionHistoryDiskBudgetMaintenance({ + ...(resolved.agentId ? { agentId: resolved.agentId } : {}), + storePath: params.storePath, + ...(options.maintenanceConfig ? { maintenanceConfig: options.maintenanceConfig } : {}), + }); + return committed.result; } /** Forks one parent SQLite transcript into a new child transcript. */ diff --git a/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts b/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts index 8087ed41182b..26690e1004ee 100644 --- a/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts +++ b/src/config/sessions/session-accessor.sqlite-lifecycle-state.ts @@ -4,16 +4,21 @@ import { executeSqliteQueryTakeFirstSync, } from "../../infra/kysely-sync.js"; import { normalizeAgentId, parseAgentSessionKey } from "../../routing/session-key.js"; -import type { OpenClawAgentDatabase } from "../../state/openclaw-agent-db.js"; -import type { - MaterializedSqliteSessionStateDeletePlan, - SqliteSessionStateDeletePlan, +import { + isIncognitoOpenClawAgentDatabase, + type OpenClawAgentDatabase, +} from "../../state/openclaw-agent-db.js"; +import { + sqliteSessionStateDeleteSnapshotsEqual, + type MaterializedSqliteSessionStateDeletePlan, + type SqliteSessionStateDeletePlan, } from "./session-accessor.sqlite-archive.js"; import type { SessionEntryLifecycleRemoval, SessionEntryLifecycleUpsert, SessionLifecycleArchivedTranscript, } from "./session-accessor.sqlite-contract.js"; +import { readSqliteSessionStateDeleteSnapshot } from "./session-accessor.sqlite-delete-snapshot.js"; import { deleteSqliteSessionEntryRows, readExactSessionEntryJsonForCanonicalRepair, @@ -33,7 +38,6 @@ import { cloneSessionEntry, getSessionKysely } from "./session-accessor.sqlite-s import { parseSqliteSessionEntryJson as parseSessionEntryRow } from "./session-accessor.sqlite-status.js"; import { buildSessionResetBoundaryPlan } from "./session-reset-boundary-event.js"; import { deleteSessionTranscriptIndexInTransaction } from "./session-transcript-index.js"; -import { serializeJsonlLines } from "./transcript-jsonl.js"; import type { SessionEntry } from "./types.js"; // Transcript-state reclamation owner. Planning stays async-free; transactions revalidate before delete. @@ -198,21 +202,6 @@ export function readReferencedSqliteSessionIdsAfterTargetMutation( return sessionIds; } -function readSqliteTranscriptArchiveLines( - database: OpenClawAgentDatabase, - sessionId: string, -): string[] { - const db = getSessionKysely(database.db); - return executeSqliteQuerySync( - database.db, - db - .selectFrom("transcript_events") - .select("event_json") - .where("session_id", "=", sessionId) - .orderBy("seq", "asc"), - ).rows.map((row) => row.event_json); -} - export function planSqliteSessionStateDeleteIfUnreferenced(params: { archiveTranscript?: boolean; archiveDirectory: string; @@ -224,15 +213,15 @@ export function planSqliteSessionStateDeleteIfUnreferenced(params: { if (params.referencedSessionIds.has(params.sessionId)) { return null; } - const lines = readSqliteTranscriptArchiveLines(params.database, params.sessionId); return { + agentId: params.database.agentId, archiveDirectory: params.archiveDirectory, - archiveTranscript: params.archiveTranscript !== false, - content: serializeJsonlLines(lines), - hadTranscriptState: - readSessionTranscriptUpdatedAt(params.database, params.sessionId) !== undefined, + archiveTranscript: + params.archiveTranscript !== false && !isIncognitoOpenClawAgentDatabase(params.database), + databasePath: params.database.path, reason: params.reason ?? "deleted", sessionId: params.sessionId, + snapshot: readSqliteSessionStateDeleteSnapshot(params.database.db, params.sessionId), }; } @@ -251,16 +240,12 @@ export function deleteMaterializedSqliteSessionStatePlans( if (referencedSessionIds.has(plan.sessionId)) { continue; } - if (plan.archiveTranscript) { - const currentContent = serializeJsonlLines( - readSqliteTranscriptArchiveLines(database, plan.sessionId), - ); - if (currentContent !== plan.content) { - throw new Error(`SQLite transcript changed before archive deletion for ${plan.sessionId}`); - } + const currentSnapshot = readSqliteSessionStateDeleteSnapshot(database.db, plan.sessionId); + if (!sqliteSessionStateDeleteSnapshotsEqual(currentSnapshot, plan.snapshot)) { + throw new Error(`SQLite session state changed before deletion for ${plan.sessionId}`); } deleteSqliteSessionStateRows(database, plan.sessionId); - if (plan.hadTranscriptState && plan.archivedTranscript) { + if (plan.snapshot.lastSeq !== null && plan.archivedTranscript) { archivedTranscripts.push(plan.archivedTranscript); } } diff --git a/src/config/sessions/session-accessor.sqlite-lifecycle.ts b/src/config/sessions/session-accessor.sqlite-lifecycle.ts index 153543e3b46c..8ffb8b8ab965 100644 --- a/src/config/sessions/session-accessor.sqlite-lifecycle.ts +++ b/src/config/sessions/session-accessor.sqlite-lifecycle.ts @@ -136,7 +136,9 @@ export async function cleanupSqliteSessionLifecycleArtifacts( orphanTranscriptMinAgeMs: params.orphanTranscriptMinAgeMs, nowMs: params.nowMs ?? Date.now(), }); - const materializedPlans = materializeSqliteSessionStateDeletePlans(cleanupPlan.deletePlans); + const materializedPlans = await materializeSqliteSessionStateDeletePlans( + cleanupPlan.deletePlans, + ); let removedEntries = 0; let archivedTranscripts: SessionLifecycleArchivedTranscript[] = []; runOpenClawAgentWriteTransaction((transactionDb) => { @@ -297,16 +299,12 @@ async function deleteSqliteSessionEntryLifecycleLocked( allowLockedEntryRemoval: boolean, expectedPluginOwnerId?: string, ): Promise { - return await runExclusiveSqliteSessionWrite(resolved, async () => { - let result: DeleteSessionEntryLifecycleResult = { - archivedTranscripts: [], - deleted: false, - }; + const prepared = await runExclusiveSqliteSessionWrite(resolved, async () => { const database = openOpenClawAgentDatabase(toDatabaseOptions(resolved)); const targetSnapshot = readSqliteLifecycleTargetSnapshot(database, params.target); const current = targetSnapshot.primary; if (!current) { - return result; + return null; } if (current.entry.modelSelectionLocked === true && !allowLockedEntryRemoval) { throw new Error(MODEL_SELECTION_LOCK_REMOVAL_MESSAGE); @@ -345,8 +343,8 @@ async function deleteSqliteSessionEntryLifecycleLocked( ) : []; const entryPlanIds = new Set(entryPlans.map((plan) => plan.sessionId)); - // Ids only — planning (which loads full transcript content) happens - // lazily one generation at a time after the main transaction. + // Ids only — archive extraction happens lazily one generation at a time + // outside the SQLite write transaction. const historicalGenerationIds = deleteTranscriptState ? readSqliteSessionGenerationIdsForKeys(database, [ params.target.canonicalKey, @@ -403,7 +401,7 @@ async function deleteSqliteSessionEntryLifecycleLocked( if (!plan) { continue; } - const materializedGeneration = materializeSqliteSessionStateDeletePlans([plan]); + const materializedGeneration = await materializeSqliteSessionStateDeletePlans([plan]); const archivedGeneration: SessionLifecycleArchivedTranscript[] = []; runOpenClawAgentWriteTransaction((transactionDb) => { // Authoritative fence: admissions are process-local sync state and this @@ -430,11 +428,24 @@ async function deleteSqliteSessionEntryLifecycleLocked( emitArchivedSqliteTranscriptUpdates(archivedGeneration); historicalArchivedTranscripts.push(...archivedGeneration); } - const materializedPlans = materializeSqliteSessionStateDeletePlans(entryPlans); + return { current, entryPlans, historicalArchivedTranscripts, targetSnapshot }; + }); + if (!prepared) { + return { archivedTranscripts: [], deleted: false }; + } + + // Archive materialization is the expensive phase. It must run between short + // writer-lane sections so unrelated writes to this store can keep progressing. + const materializedPlans = await materializeSqliteSessionStateDeletePlans(prepared.entryPlans); + const result = await runExclusiveSqliteSessionWrite(resolved, async () => { + let committed: DeleteSessionEntryLifecycleResult = { + archivedTranscripts: [], + deleted: false, + }; runOpenClawAgentWriteTransaction((transactionDb) => { const transactionSnapshot = readSqliteLifecycleTargetSnapshot(transactionDb, params.target); assertSqliteLifecycleTargetSnapshotUnchanged( - targetSnapshot, + prepared.targetSnapshot, transactionSnapshot, "delete session entry", ); @@ -458,28 +469,33 @@ async function deleteSqliteSessionEntryLifecycleLocked( ...params.target.storeKeys, ...transactionSnapshot.rows.map((row) => row.sessionKey), ]); - result = { + committed = { archivedTranscripts, deleted: true, - deletedEntry: cloneSessionEntry(current.entry), - ...(current.entry.sessionId ? { deletedSessionId: current.entry.sessionId } : {}), + deletedEntry: cloneSessionEntry(prepared.current.entry), + ...(prepared.current.entry.sessionId + ? { deletedSessionId: prepared.current.entry.sessionId } + : {}), }; }, toDatabaseOptions(resolved)); - if (result.deleted) { - emitSessionIdentityMutation({ - kind: "delete", - previous: { - ...(current.entry.sessionId ? { sessionId: current.entry.sessionId } : {}), - sessionKeys: targetSnapshot.rows.map((row) => row.sessionKey), - }, - }); - } - emitArchivedSqliteTranscriptUpdates(result.archivedTranscripts); - // Historical generations were emitted per commit above; merge them into - // the result after the final emit so callers still see every archive. - result.archivedTranscripts.push(...historicalArchivedTranscripts); - return result; + return committed; }); + if (result.deleted) { + emitSessionIdentityMutation({ + kind: "delete", + previous: { + ...(prepared.current.entry.sessionId + ? { sessionId: prepared.current.entry.sessionId } + : {}), + sessionKeys: prepared.targetSnapshot.rows.map((row) => row.sessionKey), + }, + }); + } + emitArchivedSqliteTranscriptUpdates(result.archivedTranscripts); + // Historical generations were emitted per commit above; merge them into + // the result after the final emit so callers still see every archive. + result.archivedTranscripts.push(...prepared.historicalArchivedTranscripts); + return result; } /** Deletes one persisted session entry using SQLite session rows. */ diff --git a/src/config/sessions/session-accessor.sqlite-maintenance.ts b/src/config/sessions/session-accessor.sqlite-maintenance.ts index 08f03e6cafce..5d8ea6308fa2 100644 --- a/src/config/sessions/session-accessor.sqlite-maintenance.ts +++ b/src/config/sessions/session-accessor.sqlite-maintenance.ts @@ -26,6 +26,7 @@ import type { SqliteSessionEntryMaintenancePlan } from "./session-accessor.sqlit import { cloneSessionEntry, getSessionKysely, + runExclusiveSqliteSessionWrite, toDatabaseOptions, type ResolvedSqliteReadScope, } from "./session-accessor.sqlite-scope.js"; @@ -236,28 +237,55 @@ export function applySqliteSessionEntryMaintenance( }; } -export function finalizeSqliteSessionEntryMaintenancePlansBestEffort( +export async function finalizeSqliteSessionEntryMaintenancePlansBestEffort( scope: Pick, plans: readonly SqliteSessionEntryMaintenancePlan[], -): SessionLifecycleArchivedTranscript[] { +): Promise { + return await finalizeSqliteSessionEntryMaintenancePlansWithCommit(scope, plans, async (commit) => + commit(), + ); +} + +/** Finalizes maintenance after its caller releases the per-store writer lane. */ +export async function finalizeSqliteSessionEntryMaintenancePlansAfterWriterReleaseBestEffort( + scope: Pick, + plans: readonly SqliteSessionEntryMaintenancePlan[], +): Promise { + return await finalizeSqliteSessionEntryMaintenancePlansWithCommit( + scope, + plans, + async (commit) => await runExclusiveSqliteSessionWrite(scope, async () => commit()), + ); +} + +async function finalizeSqliteSessionEntryMaintenancePlansWithCommit( + scope: Pick, + plans: readonly SqliteSessionEntryMaintenancePlan[], + commit: ( + fn: () => SessionLifecycleArchivedTranscript[], + ) => Promise, +): Promise { const entryRemovals = plans.flatMap((plan) => plan.entryRemovals); const stateDeletePlans = plans.flatMap((plan) => plan.stateDeletePlans); if (entryRemovals.length === 0 && stateDeletePlans.length === 0) { return []; } try { - const materializedPlans = materializeSqliteSessionStateDeletePlans(stateDeletePlans); - let archivedTranscripts: SessionLifecycleArchivedTranscript[] = []; - runOpenClawAgentWriteTransaction((database) => { - assertPlannedSqliteLifecycleArtifactEntriesUnchanged(database, entryRemovals); - archivedTranscripts = deleteMaterializedSqliteSessionStatePlans( - database, - materializedPlans, - undefined, - new Set(entryRemovals.map((removal) => removal.sessionKey)), - ); - deletePlannedSqliteLifecycleArtifactEntries(database, entryRemovals); - }, toDatabaseOptions(scope)); + const materializedPlans = await materializeSqliteSessionStateDeletePlans(stateDeletePlans); + const archivedTranscripts = await commit(() => { + let committed: SessionLifecycleArchivedTranscript[] = []; + runOpenClawAgentWriteTransaction((database) => { + assertPlannedSqliteLifecycleArtifactEntriesUnchanged(database, entryRemovals); + committed = deleteMaterializedSqliteSessionStatePlans( + database, + materializedPlans, + undefined, + new Set(entryRemovals.map((removal) => removal.sessionKey)), + ); + deletePlannedSqliteLifecycleArtifactEntries(database, entryRemovals); + }, toDatabaseOptions(scope)); + return committed; + }); emitCommittedSessionEntryRemovals(entryRemovals); return archivedTranscripts; } catch (error) { @@ -273,6 +301,3 @@ export function finalizeSqliteSessionEntryMaintenancePlansBestEffort( return []; } } - -// Revalidates transcript bytes before row deletion so a concurrent append is -// not dropped by an archive prepared from older content. diff --git a/src/config/sessions/session-accessor.sqlite-parent-session.ts b/src/config/sessions/session-accessor.sqlite-parent-session.ts index a35cb7e5f98e..3d86aee772eb 100644 --- a/src/config/sessions/session-accessor.sqlite-parent-session.ts +++ b/src/config/sessions/session-accessor.sqlite-parent-session.ts @@ -260,7 +260,7 @@ export async function forkSqliteSessionEntryFromParentTarget( }; }, toDatabaseOptions(resolved)); emitCommittedSessionIdentityDiff(previousIdentity, currentIdentity); - finalizeSqliteSessionEntryMaintenancePlansBestEffort(resolved, maintenancePlans); + await finalizeSqliteSessionEntryMaintenancePlansBestEffort(resolved, maintenancePlans); return result; }); } @@ -311,7 +311,7 @@ async function persistSqliteParentForkSkipPatch(params: { currentIdentity = readSqliteSessionIdentitySnapshot(database, params.sessionTarget.storeKeys); }, toDatabaseOptions(params.resolved)); emitCommittedSessionIdentityDiff(previousIdentity, currentIdentity); - finalizeSqliteSessionEntryMaintenancePlansBestEffort(params.resolved, maintenancePlans); + await finalizeSqliteSessionEntryMaintenancePlansBestEffort(params.resolved, maintenancePlans); return cloneSessionEntry(next); } diff --git a/src/config/sessions/session-accessor.sqlite-projection.ts b/src/config/sessions/session-accessor.sqlite-projection.ts index 06bceac33509..12e47c9b0759 100644 --- a/src/config/sessions/session-accessor.sqlite-projection.ts +++ b/src/config/sessions/session-accessor.sqlite-projection.ts @@ -194,7 +194,7 @@ export async function applySqliteSessionEntryReplacements(params: { }); } } - finalizeSqliteSessionEntryMaintenancePlansBestEffort(resolved, maintenancePlans); + await finalizeSqliteSessionEntryMaintenancePlansBestEffort(resolved, maintenancePlans); return operation.result; }); } @@ -281,7 +281,7 @@ export async function applySqliteSessionStoreProjection(params: { toDatabaseOptions(resolved), { operationLabel: "session.store-projection" }, ); - finalizeSqliteSessionEntryMaintenancePlansBestEffort(resolved, maintenancePlans); + await finalizeSqliteSessionEntryMaintenancePlansBestEffort(resolved, maintenancePlans); return operation.result; }); } @@ -359,7 +359,9 @@ export async function applySqliteSessionEntryLifecycleMutation(params: { }); let materializedRemovalPlans: MaterializedSqliteSessionStateDeletePlan[] = []; try { - materializedRemovalPlans = materializeSqliteSessionStateDeletePlans(projected.deletePlans); + materializedRemovalPlans = await materializeSqliteSessionStateDeletePlans( + projected.deletePlans, + ); } catch (error) { captureArtifactCleanupError(error); } @@ -514,10 +516,8 @@ export async function applySqliteSessionEntryLifecycleMutation(params: { ); }, toDatabaseOptions(resolved)); emitCommittedLifecycleIdentityMutations({ projected, removedSessionKeys }); - const maintenanceArchivedTranscripts = finalizeSqliteSessionEntryMaintenancePlansBestEffort( - resolved, - maintenancePlans, - ); + const maintenanceArchivedTranscripts = + await finalizeSqliteSessionEntryMaintenancePlansBestEffort(resolved, maintenancePlans); archivedTranscripts = [...archivedTranscripts, ...maintenanceArchivedTranscripts]; const afterCount = readSqliteSessionEntryCount( openOpenClawAgentDatabase(toDatabaseOptions(resolved)), @@ -592,7 +592,7 @@ export async function purgeSqliteDeletedAgentSessionEntries( referencedSessionIds, }), ); - const materializedPlans = materializeSqliteSessionStateDeletePlans(deletePlans); + const materializedPlans = await materializeSqliteSessionStateDeletePlans(deletePlans); const removedSessionKeys = entryRemovals.map((removal) => removal.sessionKey); let archivedTranscripts: SessionLifecycleArchivedTranscript[] = []; const maintenancePlans: SqliteSessionEntryMaintenancePlan[] = []; @@ -616,7 +616,7 @@ export async function purgeSqliteDeletedAgentSessionEntries( emitCommittedSessionEntryRemovals(entryRemovals); archivedTranscripts = [ ...archivedTranscripts, - ...finalizeSqliteSessionEntryMaintenancePlansBestEffort(resolved, maintenancePlans), + ...(await finalizeSqliteSessionEntryMaintenancePlansBestEffort(resolved, maintenancePlans)), ]; const afterCount = readSqliteSessionEntryCount( openOpenClawAgentDatabase(toDatabaseOptions(resolved)), diff --git a/src/config/sessions/session-history-eviction.ts b/src/config/sessions/session-history-eviction.ts index a3d26c64e5a2..68a424509138 100644 --- a/src/config/sessions/session-history-eviction.ts +++ b/src/config/sessions/session-history-eviction.ts @@ -366,7 +366,7 @@ async function enforceSessionHistoryMaintenanceSerialized( // Extract-before-delete is the retention invariant. Admission is fenced across archive // creation, then rechecked inside the write transaction before any rows are reclaimed. - const materialized = materializeSqliteSessionStateDeletePlans([plan]); + const materialized = await materializeSqliteSessionStateDeletePlans([plan]); let deleted = false; let archivedTranscripts: ReturnType = []; diff --git a/src/infra/tsdown-config.test.ts b/src/infra/tsdown-config.test.ts index bca176d32b01..103959db3367 100644 --- a/src/infra/tsdown-config.test.ts +++ b/src/infra/tsdown-config.test.ts @@ -153,6 +153,7 @@ describe("tsdown config", () => { "cli/gateway-lifecycle.runtime", "agents/compaction-planning.worker", "agents/model-provider-auth.worker", + "config/sessions/session-accessor.sqlite-archive.worker", "state/openclaw-database-verify.worker", "system-agent/setup-inference-detection.worker", "plugins/memory-state", diff --git a/src/scripts/ci-changed-scope.windows.test.ts b/src/scripts/ci-changed-scope.windows.test.ts index 205a5195c870..88a31427248f 100644 --- a/src/scripts/ci-changed-scope.windows.test.ts +++ b/src/scripts/ci-changed-scope.windows.test.ts @@ -5,7 +5,11 @@ const { detectChangedScope } = await import("../../scripts/ci-changed-scope.mjs" describe("detectChangedScope Windows routing", () => { it("routes SQLite transcript archive changes to Windows", () => { - for (const archivePath of ["src/config/sessions/session-accessor.sqlite-archive.ts"]) { + for (const archivePath of [ + "src/config/sessions/session-accessor.sqlite-archive.ts", + "src/config/sessions/session-accessor.sqlite-archive.worker.test.ts", + "src/config/sessions/session-accessor.sqlite-archive.worker.ts", + ]) { expect(detectChangedScope([archivePath]), archivePath).toMatchObject({ runNode: true, runWindows: true, diff --git a/src/state/openclaw-agent-db.ts b/src/state/openclaw-agent-db.ts index 12ffb9e55237..33aa15a1b4a0 100644 --- a/src/state/openclaw-agent-db.ts +++ b/src/state/openclaw-agent-db.ts @@ -579,6 +579,11 @@ export function listOpenIncognitoAgentDatabases(): Array<{ agentId: string; stor ); } +/** Returns whether this exact process-held database is incognito/in-memory. */ +export function isIncognitoOpenClawAgentDatabase(database: OpenClawAgentDatabase): boolean { + return incognitoDatabases.has(database); +} + /** List process-held agent databases without opening or inspecting fixture state. */ export function listOpenClawAgentDatabasesForTest(): Array<{ agentId: string; path: string }> { return [...cachedDatabases.values()] diff --git a/test/helpers/sqlite-sessions-transcripts-flip-proof.ts b/test/helpers/sqlite-sessions-transcripts-flip-proof.ts index 59576d2f842d..4ae7bd7e7d41 100644 --- a/test/helpers/sqlite-sessions-transcripts-flip-proof.ts +++ b/test/helpers/sqlite-sessions-transcripts-flip-proof.ts @@ -36,6 +36,8 @@ import { readSessionTranscriptEvents, resolveSessionTranscriptIdentity, } from "../../src/plugin-sdk/session-transcript-runtime.js"; +import { closeOpenClawAgentDatabasesForTest } from "../../src/state/openclaw-agent-db.js"; +import { closeOpenClawStateDatabaseForTest } from "../../src/state/openclaw-state-db.js"; import { sleep } from "../../src/utils.js"; import { normalizeSessionDeliveryState } from "../../src/utils/delivery-context.shared.js"; import { createOpenClawTestInstance } from "./openclaw-test-instance.js"; @@ -114,9 +116,14 @@ export async function runSqliteSessionsTranscriptsFlipProof(options: RunOptions name: `sqlite-sessions-transcripts-flip-${randomUUID()}`, config: buildMockOpenAiConfig(mockOpenAiPort), env: { + ALL_PROXY: undefined, + HTTP_PROXY: undefined, + HTTPS_PROXY: undefined, + NO_PROXY: "127.0.0.1,localhost", OPENAI_API_KEY: "sk-openclaw-e2e-mock", OPENCLAW_TEST_MINIMAL_GATEWAY: undefined, OPENCLAW_SKIP_PROVIDERS: undefined, + no_proxy: "127.0.0.1,localhost", }, startTimeoutMs: 90_000, stopTimeoutMs: 3_000, @@ -340,6 +347,9 @@ export async function runSqliteSessionsTranscriptsFlipProof(options: RunOptions await record("failure"); } finally { await stopChildProcess(mockOpenAi); + await inst.stopGateway(); + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); await inst.cleanup(); } diff --git a/test/package-scripts.test.ts b/test/package-scripts.test.ts index f74fd47818d9..8635c7011dba 100644 --- a/test/package-scripts.test.ts +++ b/test/package-scripts.test.ts @@ -245,6 +245,13 @@ describe("package scripts", () => { ); }); + it("runs SQLite transcript archive worker coverage in Windows CI", () => { + const windowsCi = readPackageJson().scripts["test:windows:ci"]; + expect(windowsCi).toContain( + "src/config/sessions/session-accessor.sqlite-archive.worker.test.ts", + ); + }); + it("runs cross-OS installer behavior coverage in Windows CI", () => { expect(readPackageJson().scripts["test:windows:ci"]).toContain( "test/scripts/openclaw-cross-os-installer.windows.test.ts", diff --git a/test/release-check.test.ts b/test/release-check.test.ts index 91703ff5f7c1..ff278b6c87a4 100644 --- a/test/release-check.test.ts +++ b/test/release-check.test.ts @@ -727,6 +727,7 @@ describe("collectMissingPackPaths", () => { "dist/agents/compaction-planning.worker.js", "dist/agents/model-provider-auth.worker.js", "dist/audit/audit-event-writer.worker.js", + "dist/config/sessions/session-accessor.sqlite-archive.worker.js", "dist/config/sessions/session-transcript-reconcile.worker.js", "dist/state/openclaw-database-verify.worker.js", "dist/system-agent/setup-inference-detection.worker.js", @@ -767,6 +768,7 @@ describe("collectMissingPackPaths", () => { "dist/agents/compaction-planning.worker.js", "dist/agents/model-provider-auth.worker.js", "dist/audit/audit-event-writer.worker.js", + "dist/config/sessions/session-accessor.sqlite-archive.worker.js", "dist/config/sessions/session-transcript-reconcile.worker.js", "dist/state/openclaw-database-verify.worker.js", "dist/system-agent/setup-inference-detection.worker.js", diff --git a/test/scripts/sqlite-sessions-transcripts-flip-proof.built-cli.e2e.test.ts b/test/scripts/sqlite-sessions-transcripts-flip-proof.built-cli.e2e.test.ts index d27d415cbcde..6da4c0ca6673 100644 --- a/test/scripts/sqlite-sessions-transcripts-flip-proof.built-cli.e2e.test.ts +++ b/test/scripts/sqlite-sessions-transcripts-flip-proof.built-cli.e2e.test.ts @@ -1,5 +1,25 @@ // Built-CLI SQLite flip proof requires dist entrypoints before running the gateway lifecycle. +import { createHash, randomBytes, randomUUID } from "node:crypto"; +import fs from "node:fs"; +import path from "node:path"; +import { performance } from "node:perf_hooks"; +import { DatabaseSync } from "node:sqlite"; import { describe, expect, it } from "vitest"; +import { readSessionArchiveContentSync } from "../../src/config/sessions/archive-compression.js"; +import { + loadSessionEntry, + loadTranscriptEvents, + replaceSessionEntry, +} from "../../src/config/sessions/session-accessor.js"; +import { replaceSqliteTranscriptEvents } from "../../src/config/sessions/session-accessor.sqlite.js"; +import { resolveSqliteTargetFromSessionStorePath } from "../../src/config/sessions/session-sqlite-target.js"; +import { + connectGatewayClient, + disconnectGatewayClient, +} from "../../src/gateway/test-helpers.e2e.js"; +import { closeOpenClawAgentDatabasesForTest } from "../../src/state/openclaw-agent-db.js"; +import { closeOpenClawStateDatabaseForTest } from "../../src/state/openclaw-state-db.js"; +import { createOpenClawTestInstance } from "../helpers/openclaw-test-instance.js"; import { assertSqliteFlipProofCore } from "../helpers/sqlite-sessions-transcripts-flip-proof-assertions.ts"; import { runSqliteSessionsTranscriptsFlipProof } from "../helpers/sqlite-sessions-transcripts-flip-proof.ts"; @@ -12,4 +32,251 @@ describe("SQLite sessions/transcripts flip built CLI proof", () => { ); assertSqliteFlipProofCore(report); }, 420_000); + + it("keeps built gateway RPC responsive while deleting a large transcript", async () => { + const inst = await createOpenClawTestInstance({ + name: `sqlite-archive-responsive-${randomUUID()}`, + startTimeoutMs: 90_000, + stopTimeoutMs: 5_000, + }); + inst.state.applyEnv(); + const sessionId = "sqlite-large-archive-responsive"; + const sessionKey = "agent:main:dashboard:sqlite-large-archive-responsive"; + const writerSessionKey = "agent:main:dashboard:sqlite-large-archive-writer"; + const warmupSessionId = "sqlite-archive-worker-warmup"; + const warmupSessionKey = "agent:main:dashboard:sqlite-archive-worker-warmup"; + const storePath = path.join(inst.stateDir, "agents", "main", "sessions", "sessions.json"); + const archiveDirectory = path.dirname(storePath); + const events = createLargeTranscriptEvents(sessionId); + const expectedArchiveContent = `${events.map((event) => JSON.stringify(event)).join("\n")}\n`; + let deleteClient: Awaited> | undefined; + let probeClient: Awaited> | undefined; + + try { + await replaceSessionEntry({ sessionKey, storePath }, { sessionId, updatedAt: Date.now() }); + await replaceSessionEntry( + { sessionKey: writerSessionKey, storePath }, + { sessionId: "sqlite-large-archive-writer", updatedAt: Date.now() }, + ); + await replaceSqliteTranscriptEvents({ sessionKey, sessionId, storePath }, events); + await replaceSessionEntry( + { sessionKey: warmupSessionKey, storePath }, + { sessionId: warmupSessionId, updatedAt: Date.now() }, + ); + await replaceSqliteTranscriptEvents( + { sessionKey: warmupSessionKey, sessionId: warmupSessionId, storePath }, + [ + { + type: "session", + id: warmupSessionId, + content: "warm the built archive worker", + } as unknown as TestTranscriptEvent, + ], + ); + const databasePath = requireSqliteDatabasePath(storePath); + expect(readSessionRowCounts(databasePath, sessionId)).toEqual({ + fts: 1, + sessionWindows: 1, + transcriptEvents: events.length, + }); + closeOpenClawAgentDatabasesForTest(); + + await expect(inst.entrypoint()).resolves.toEqual( + expect.arrayContaining([expect.stringMatching(/^dist\/index\.(?:js|mjs)$/u)]), + ); + await inst.startGateway(); + [deleteClient, probeClient] = await Promise.all([ + connectGatewayClient({ + url: inst.url, + token: inst.gatewayToken, + clientDisplayName: "sqlite-large-archive-delete", + requestTimeoutMs: 120_000, + timeoutMs: 20_000, + }), + connectGatewayClient({ + url: inst.url, + token: inst.gatewayToken, + clientDisplayName: "sqlite-large-archive-presence", + requestTimeoutMs: 2_000, + timeoutMs: 20_000, + }), + ]); + // Cold-opening and indexing the pre-seeded 64 MiB database is outside + // the deletion latency measurement below and can exceed the normal RPC + // timeout on Windows CI hosts. Finish that one-time initialization first. + await deleteClient.request("sessions.list", {}, { timeoutMs: 120_000 }); + for (let attempt = 0; attempt < 3; attempt += 1) { + await probeClient.request("system-presence", {}, { timeoutMs: 2_000 }); + } + // Prime the built sidecar and OS file cache with a tiny transcript so the + // latency assertion below measures data-size-dependent archive work. + await deleteClient.request( + "sessions.delete", + { key: warmupSessionKey, deleteTranscript: true }, + { timeoutMs: 20_000 }, + ); + + let archivePublishedAt: number | undefined; + let deleteSettled = false; + const publicationPoll = setInterval(() => { + if (findPublishedArchive(archiveDirectory, sessionId)) { + archivePublishedAt ??= performance.now(); + } + }, 5); + const deletion = deleteClient + .request<{ archived?: string[]; deleted?: boolean; ok?: boolean }>( + "sessions.delete", + { key: sessionKey, deleteTranscript: true }, + { timeoutMs: 120_000 }, + ) + .finally(() => { + deleteSettled = true; + }); + void deletion.catch(() => undefined); + const writerStartedAt = performance.now(); + const writerResult = await probeClient.request<{ key?: string; ok?: boolean }>( + "sessions.patch", + { key: writerSessionKey, label: "writer-progressed-during-archive" }, + { timeoutMs: 2_000 }, + ); + const writerLatencyMs = performance.now() - writerStartedAt; + expect(writerResult).toMatchObject({ ok: true, key: writerSessionKey }); + expect(writerLatencyMs).toBeLessThan(500); + expect(deleteSettled).toBe(false); + expect(archivePublishedAt).toBeUndefined(); + const prePublicationProbeLatencies: number[] = []; + const shouldProbeBeforePublication = () => !deleteSettled && archivePublishedAt === undefined; + try { + while (shouldProbeBeforePublication()) { + const probeStartedAt = performance.now(); + await probeClient.request("system-presence", {}, { timeoutMs: 2_000 }); + const probeCompletedAt = performance.now(); + // Record every probe that started before publication was observed. + // A synchronous implementation can delay this response until after + // publication; dropping that crossing sample would hide the stall. + prePublicationProbeLatencies.push(probeCompletedAt - probeStartedAt); + await new Promise((resolve) => { + setTimeout(resolve, 5); + }); + } + } finally { + clearInterval(publicationPoll); + } + + const deleteResult = await deletion; + expect(deleteResult).toMatchObject({ ok: true, deleted: true }); + expect(prePublicationProbeLatencies.length).toBeGreaterThan(5); + // Keep enough headroom for Windows scheduling and a probe that crosses + // into the existing synchronous SQLite/FTS deletion tail. The former + // synchronous archive path instead exceeds the probe's 2s RPC timeout. + expect(Math.max(...prePublicationProbeLatencies)).toBeLessThan(500); + const archivedPath = deleteResult.archived?.[0]; + expect(archivedPath).toBeTruthy(); + + await Promise.all([ + disconnectGatewayClient(deleteClient), + disconnectGatewayClient(probeClient), + ]); + deleteClient = undefined; + probeClient = undefined; + await inst.stopGateway(); + + const archivedContent = readSessionArchiveContentSync(archivedPath ?? ""); + expect(Buffer.byteLength(archivedContent)).toBe(Buffer.byteLength(expectedArchiveContent)); + expect(sha256(archivedContent)).toBe(sha256(expectedArchiveContent)); + expect(loadSessionEntry({ sessionKey, storePath })).toBeUndefined(); + await expect(loadTranscriptEvents({ sessionKey, sessionId, storePath })).resolves.toEqual([]); + expect(readSessionRowCounts(databasePath, sessionId)).toEqual({ + fts: 0, + sessionWindows: 0, + transcriptEvents: 0, + }); + } finally { + await Promise.allSettled( + [deleteClient, probeClient] + .filter((client): client is NonNullable => client !== undefined) + .map((client) => disconnectGatewayClient(client)), + ); + await inst.stopGateway(); + closeOpenClawAgentDatabasesForTest(); + closeOpenClawStateDatabaseForTest(); + await inst.cleanup(); + } + }, 180_000); }); + +type TestTranscriptEvent = Parameters[1][number]; + +function createLargeTranscriptEvents(sessionId: string): TestTranscriptEvent[] { + const indexedMessage = { + type: "message", + id: "sqlite-large-archive-indexed-message", + parentId: null, + message: { + role: "user", + content: [{ type: "text", text: "large archive searchable marker" }], + }, + timestamp: Date.now(), + } as unknown as TestTranscriptEvent; + return [ + indexedMessage, + ...Array.from( + { length: 63 }, + (_, index) => + ({ + type: "session", + id: `${sessionId}-${index}`, + content: `${index}:${randomBytes(768 * 1024).toString("base64")}`, + }) as unknown as TestTranscriptEvent, + ), + ]; +} + +function findPublishedArchive(archiveDirectory: string, sessionId: string): string | undefined { + const prefix = `${sessionId}.jsonl.deleted.`; + try { + return fs + .readdirSync(archiveDirectory) + .find((entry) => entry.startsWith(prefix) && !entry.endsWith(".tmp")); + } catch { + return undefined; + } +} + +function requireSqliteDatabasePath(storePath: string): string { + const target = resolveSqliteTargetFromSessionStorePath(storePath); + if (!target.path) { + throw new Error(`could not resolve SQLite database path for ${storePath}`); + } + return target.path; +} + +function readSessionRowCounts( + databasePath: string, + sessionId: string, +): { + fts: number; + sessionWindows: number; + transcriptEvents: number; +} { + const database = new DatabaseSync(databasePath, { readOnly: true }); + try { + const count = (table: "session_transcript_fts" | "session_windows" | "transcript_events") => { + const row = database + .prepare(`SELECT COUNT(*) AS count FROM ${table} WHERE session_id = ?`) + .get(sessionId) as { count: number }; + return row.count; + }; + return { + fts: count("session_transcript_fts"), + sessionWindows: count("session_windows"), + transcriptEvents: count("transcript_events"), + }; + } finally { + database.close(); + } +} + +function sha256(content: string): string { + return createHash("sha256").update(content).digest("hex"); +} diff --git a/test/scripts/sqlite-sessions-transcripts-flip-proof.e2e.test.ts b/test/scripts/sqlite-sessions-transcripts-flip-proof.e2e.test.ts index 2597a282e2c2..a554befb6e61 100644 --- a/test/scripts/sqlite-sessions-transcripts-flip-proof.e2e.test.ts +++ b/test/scripts/sqlite-sessions-transcripts-flip-proof.e2e.test.ts @@ -51,12 +51,12 @@ describe("SQLite sessions/transcripts flip proof harness", () => { expect(report.rollbackRestore?.manifestPath).toContain("session-sqlite-migration-runs"); expect( report.rollbackRestore?.restoredFiles.some((filePath) => - filePath.endsWith("/sqlite-rollback-restore.jsonl"), + filePath.replaceAll("\\", "/").endsWith("/sqlite-rollback-restore.jsonl"), ), ).toBe(true); expect( report.rollbackRestore?.idempotentRestoreSkippedFiles.some((filePath) => - filePath.endsWith("/sqlite-rollback-restore.jsonl"), + filePath.replaceAll("\\", "/").endsWith("/sqlite-rollback-restore.jsonl"), ), ).toBe(true); expect(report.scaleMigration).toMatchObject({ diff --git a/tsdown.config.ts b/tsdown.config.ts index 64ba55c4034a..e6a5a1db3fef 100644 --- a/tsdown.config.ts +++ b/tsdown.config.ts @@ -290,6 +290,8 @@ function buildCoreDistEntries(): Record { "agents/compaction-planning.worker": "src/agents/compaction-planning.worker.ts", "agents/model-provider-auth.worker": "src/agents/model-provider-auth.worker.ts", "audit/audit-event-writer.worker": "src/audit/audit-event-writer.worker.ts", + "config/sessions/session-accessor.sqlite-archive.worker": + "src/config/sessions/session-accessor.sqlite-archive.worker.ts", "config/sessions/session-transcript-reconcile.worker": "src/config/sessions/session-transcript-reconcile.worker.ts", "state/openclaw-database-verify.worker": "src/state/openclaw-database-verify.worker.ts",