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