mirror of
https://github.com/openclaw/openclaw.git
synced 2026-08-12 21:53:00 -06:00
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 <josh@martian.engineering>
This commit is contained in:
Heming Zeng
@@ -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.
|
||||
|
||||
+1
-1
@@ -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",
|
||||
|
||||
@@ -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 =
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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 });
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
@@ -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<SqliteTranscriptArchiveWorkerResult[]> {
|
||||
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<SqliteTranscriptArchiveWorkerResult[]> {
|
||||
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<MaterializedSqliteSessionStateDeletePlan[]> {
|
||||
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) {
|
||||
|
||||
@@ -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<ReturnType<typeof materializeSqliteSessionStateDeletePlans>>;
|
||||
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<OpenClawAgentKyselyDatabase>(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<OpenClawAgentKyselyDatabase>(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<OpenClawAgentKyselyDatabase>(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<OpenClawAgentKyselyDatabase>(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<OpenClawAgentKyselyDatabase>(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<typeof openLifecycleTestDatabase>,
|
||||
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<typeof openLifecycleTestDatabase>,
|
||||
sessionId: string,
|
||||
): void {
|
||||
runOpenClawAgentWriteTransaction(
|
||||
(transactionDb) => {
|
||||
const db = getNodeSqliteKysely<OpenClawAgentKyselyDatabase>(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<typeof openLifecycleTestDatabase>,
|
||||
plans: Parameters<typeof deleteMaterializedSqliteSessionStatePlans>[1],
|
||||
excludedSessionKey: string,
|
||||
): void {
|
||||
runOpenClawAgentWriteTransaction(
|
||||
(transactionDb) =>
|
||||
deleteMaterializedSqliteSessionStatePlans(
|
||||
transactionDb,
|
||||
plans,
|
||||
undefined,
|
||||
new Set([excludedSessionKey]),
|
||||
),
|
||||
{ agentId: database.agentId, path: database.path },
|
||||
);
|
||||
}
|
||||
@@ -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<OpenClawAgentKyselyDatabase, "transcript_events">;
|
||||
|
||||
function isSqliteTranscriptArchiveWorkerData(value: unknown): boolean {
|
||||
return (
|
||||
Boolean(value) &&
|
||||
typeof value === "object" &&
|
||||
!Array.isArray(value) &&
|
||||
(value as { type?: unknown }).type === "sqlite-transcript-archive-v1"
|
||||
);
|
||||
}
|
||||
|
||||
function parseSessionStateDeleteSnapshot(value: unknown): SqliteSessionStateDeleteSnapshot | null {
|
||||
if (!value || typeof value !== "object" || Array.isArray(value)) {
|
||||
return null;
|
||||
}
|
||||
const snapshot = value as Record<string, unknown>;
|
||||
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<string, unknown>;
|
||||
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<TranscriptArchiveDatabase>(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<typeof parentPort>,
|
||||
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);
|
||||
}
|
||||
@@ -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<void>) | 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<typeof import("./session-accessor.sqlite-archive.js")>();
|
||||
return {
|
||||
...actual,
|
||||
materializeSqliteSessionStateDeletePlans: async (
|
||||
...args: Parameters<typeof actual.materializeSqliteSessionStateDeletePlans>
|
||||
) => {
|
||||
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<void>((resolve) => {
|
||||
markMaterializationStarted = resolve;
|
||||
});
|
||||
let releaseMaterialization: () => void = () => undefined;
|
||||
const materializationGate = new Promise<void>((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<false>((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 = {
|
||||
|
||||
@@ -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<SessionStateDeleteSnapshotDatabase>(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<number | bigint>().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,
|
||||
};
|
||||
}
|
||||
@@ -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<TSnapshot>(
|
||||
params: SqliteSessionEntrySnapshotPatchParams<TSnapshot>,
|
||||
): Promise<SessionEntry | null> {
|
||||
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<TSnapshot>(
|
||||
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. */
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<DeleteSessionEntryLifecycleResult> {
|
||||
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. */
|
||||
|
||||
@@ -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<ResolvedSqliteReadScope, "agentId" | "env" | "path">,
|
||||
plans: readonly SqliteSessionEntryMaintenancePlan[],
|
||||
): SessionLifecycleArchivedTranscript[] {
|
||||
): Promise<SessionLifecycleArchivedTranscript[]> {
|
||||
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<ResolvedSqliteReadScope, "agentId" | "env" | "path">,
|
||||
plans: readonly SqliteSessionEntryMaintenancePlan[],
|
||||
): Promise<SessionLifecycleArchivedTranscript[]> {
|
||||
return await finalizeSqliteSessionEntryMaintenancePlansWithCommit(
|
||||
scope,
|
||||
plans,
|
||||
async (commit) => await runExclusiveSqliteSessionWrite(scope, async () => commit()),
|
||||
);
|
||||
}
|
||||
|
||||
async function finalizeSqliteSessionEntryMaintenancePlansWithCommit(
|
||||
scope: Pick<ResolvedSqliteReadScope, "agentId" | "env" | "path">,
|
||||
plans: readonly SqliteSessionEntryMaintenancePlan[],
|
||||
commit: (
|
||||
fn: () => SessionLifecycleArchivedTranscript[],
|
||||
) => Promise<SessionLifecycleArchivedTranscript[]>,
|
||||
): Promise<SessionLifecycleArchivedTranscript[]> {
|
||||
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.
|
||||
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
|
||||
@@ -194,7 +194,7 @@ export async function applySqliteSessionEntryReplacements<T>(params: {
|
||||
});
|
||||
}
|
||||
}
|
||||
finalizeSqliteSessionEntryMaintenancePlansBestEffort(resolved, maintenancePlans);
|
||||
await finalizeSqliteSessionEntryMaintenancePlansBestEffort(resolved, maintenancePlans);
|
||||
return operation.result;
|
||||
});
|
||||
}
|
||||
@@ -281,7 +281,7 @@ export async function applySqliteSessionStoreProjection<T>(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)),
|
||||
|
||||
@@ -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<typeof deleteMaterializedSqliteSessionStatePlans> =
|
||||
[];
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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()]
|
||||
|
||||
@@ -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();
|
||||
}
|
||||
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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<ReturnType<typeof connectGatewayClient>> | undefined;
|
||||
let probeClient: Awaited<ReturnType<typeof connectGatewayClient>> | 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<void>((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<typeof client> => client !== undefined)
|
||||
.map((client) => disconnectGatewayClient(client)),
|
||||
);
|
||||
await inst.stopGateway();
|
||||
closeOpenClawAgentDatabasesForTest();
|
||||
closeOpenClawStateDatabaseForTest();
|
||||
await inst.cleanup();
|
||||
}
|
||||
}, 180_000);
|
||||
});
|
||||
|
||||
type TestTranscriptEvent = Parameters<typeof replaceSqliteTranscriptEvents>[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");
|
||||
}
|
||||
|
||||
@@ -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({
|
||||
|
||||
@@ -290,6 +290,8 @@ function buildCoreDistEntries(): Record<string, string> {
|
||||
"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",
|
||||
|
||||
Reference in New Issue
Block a user