diff --git a/src/agents/sessions/session-manager-codec.ts b/src/agents/sessions/session-manager-codec.ts index 3b316edd0697..156c3b388082 100644 --- a/src/agents/sessions/session-manager-codec.ts +++ b/src/agents/sessions/session-manager-codec.ts @@ -24,46 +24,43 @@ export function isSessionContextMetadataEntry(entry: SessionEntry): boolean { ); } -function migrateV1ToV2( - entries: FileEntry[], - entriesByOriginalIndex?: readonly (FileEntry | undefined)[], -): void { - const ids = new Set(); - let previousId: string | null = null; +export type SessionFileEntryMigrationState = { + createEntryId: (originalIndex: number) => string; + previousId: string | null; + resolveOriginalEntryId?: (originalIndex: number) => string | undefined; + sourceVersion: number; +}; - for (const entry of entries) { +export function migrateSessionFileEntryToCurrentVersion( + entry: FileEntry, + originalIndex: number, + state: SessionFileEntryMigrationState, +): void { + if (state.sourceVersion < 2) { if (entry.type === "session") { entry.version = 2; - continue; - } + } else { + entry.id = state.createEntryId(originalIndex); + entry.parentId = state.previousId; + state.previousId = entry.id; - entry.id = generateSessionEntryId(ids); - ids.add(entry.id); - entry.parentId = previousId; - previousId = entry.id; - - if (entry.type === "compaction") { - const compaction = entry as CompactionEntry & { firstKeptEntryIndex?: number }; - if (typeof compaction.firstKeptEntryIndex === "number") { - const targetEntry = entriesByOriginalIndex - ? entriesByOriginalIndex[compaction.firstKeptEntryIndex] - : entries[compaction.firstKeptEntryIndex]; - if (targetEntry && targetEntry.type !== "session") { - compaction.firstKeptEntryId = targetEntry.id; + if (entry.type === "compaction") { + const compaction = entry as CompactionEntry & { firstKeptEntryIndex?: number }; + if (typeof compaction.firstKeptEntryIndex === "number") { + const firstKeptEntryId = state.resolveOriginalEntryId?.(compaction.firstKeptEntryIndex); + if (firstKeptEntryId) { + compaction.firstKeptEntryId = firstKeptEntryId; + } + delete compaction.firstKeptEntryIndex; } - delete compaction.firstKeptEntryIndex; } } } -} -function migrateV2ToV3(entries: FileEntry[]): void { - for (const entry of entries) { + if (state.sourceVersion < 3) { if (entry.type === "session") { entry.version = 3; - continue; - } - if (entry.type === "message" && entry.message) { + } else if (entry.type === "message" && entry.message) { const message = entry.message as { role: string; customType?: string }; if (message.role === "hookMessage") { message.role = "custom"; @@ -82,11 +79,24 @@ export function migrateToCurrentVersion( if (version >= CURRENT_SESSION_VERSION) { return false; } - if (version < 2) { - migrateV1ToV2(entries, entriesByOriginalIndex); - } - if (version < 3) { - migrateV2ToV3(entries); + const ids = new Set(); + const state: SessionFileEntryMigrationState = { + createEntryId: () => { + const id = generateSessionEntryId(ids); + ids.add(id); + return id; + }, + previousId: null, + resolveOriginalEntryId: (originalIndex) => { + const targetEntry = entriesByOriginalIndex + ? entriesByOriginalIndex[originalIndex] + : entries[originalIndex]; + return targetEntry && targetEntry.type !== "session" ? targetEntry.id : undefined; + }, + sourceVersion: version, + }; + for (const [index, entry] of entries.entries()) { + migrateSessionFileEntryToCurrentVersion(entry, index, state); } return true; } diff --git a/src/commands/doctor-session-sqlite-readers.ts b/src/commands/doctor-session-sqlite-readers.ts index 228086680587..ed181c46025f 100644 --- a/src/commands/doctor-session-sqlite-readers.ts +++ b/src/commands/doctor-session-sqlite-readers.ts @@ -1,9 +1,16 @@ /** Read-only diagnostic readers used by the session SQLite doctor mode. */ +import { createHash } from "node:crypto"; import fs from "node:fs"; import type { DatabaseSync } from "node:sqlite"; import { TextDecoder } from "node:util"; import { isRecord } from "@openclaw/normalization-core/record-coerce"; -import { normalizeLoadedFileEntry, type FileEntry } from "../agents/sessions/session-manager.js"; +import { + migrateSessionFileEntryToCurrentVersion, + normalizeLoadedFileEntry, + partitionSessionFileEntries, + type SessionFileEntryMigrationState, +} from "../agents/sessions/session-manager-codec.js"; +import type { FileEntry } from "../agents/sessions/session-manager-types.js"; import type { TranscriptEvent } from "../config/sessions/session-accessor.js"; import type { SqliteTranscriptStorageRow } from "../config/sessions/session-accessor.sqlite-read.js"; import { resolveSqliteTargetFromSessionStorePath } from "../config/sessions/session-sqlite-target.js"; @@ -49,6 +56,7 @@ type TranscriptEventCountResult = | { status: "malformed"; message: string }; const JSONL_READ_CHUNK_BYTES = 64 * 1024; +const MAX_LEGACY_COMPACTION_TARGETS = 100_000; export function countTranscriptEventsForPath( transcriptPath: string | undefined, @@ -75,36 +83,208 @@ export function countTranscriptEventsForPath( export function createTranscriptEventReader( transcriptPath: string, + sessionId: string, ): (append: (event: TranscriptEvent) => void) => void { return (append) => { - for (const line of iterateJsonlLinesSync(transcriptPath)) { - const parsed = parseJsonlLine(line); - if (parsed) { - // Import is the migration boundary: repair legacy JSONL message shapes - // here because the SQLite runtime read path assumes canonical rows. - append(normalizeLoadedFileEntry(parsed as FileEntry) as TranscriptEvent); - } + for (const event of readTranscriptEventsForImport(transcriptPath, sessionId, false)) { + append(event as TranscriptEvent); } }; } export function createTranscriptEventPrefixReader( transcriptPath: string, + sessionId: string, ): (append: (event: TranscriptEvent) => void) => void { return (append) => { - try { - for (const line of iterateJsonlLinesSync(transcriptPath)) { - const parsed = parseJsonlLine(line); - if (parsed) { - append(normalizeLoadedFileEntry(parsed as FileEntry) as TranscriptEvent); - } - } - } catch { - // The caller records the malformed transcript issue; keep the readable prefix. + for (const event of readTranscriptEventsForImport(transcriptPath, sessionId, true)) { + append(event as TranscriptEvent); } }; } +function readTranscriptEventsForImport( + transcriptPath: string, + sessionId: string, + allowMalformedPrefix: boolean, +): Iterable { + // Production import owns the process-wide Gateway/SQLite-maintenance lock + // through commit and archive. Fingerprints catch non-cooperating external edits. + const sourceFingerprint = readTranscriptFileFingerprint(transcriptPath); + const plan = planTranscriptImport(transcriptPath, allowMalformedPrefix); + assertTranscriptFileUnchanged(transcriptPath, sourceFingerprint); + const classificationHeader = { + id: sessionId, + type: "session", + version: plan.sourceVersion, + } as unknown as FileEntry; + // V1 compactions refer to original row indexes. Stable index-derived IDs let + // the second pass resolve those links without retaining the transcript. + const idPrefix = createHash("sha256") + .update(transcriptPath) + .update("\0") + .update(sessionId) + .digest("hex") + .slice(0, 16); + + return { + *[Symbol.iterator]() { + assertTranscriptFileUnchanged(transcriptPath, sourceFingerprint); + const migratedTargetIds = new Map(); + const migrationState: SessionFileEntryMigrationState = { + createEntryId: (originalIndex) => `${idPrefix}-${originalIndex.toString(36)}`, + previousId: null, + resolveOriginalEntryId: (originalIndex) => migratedTargetIds.get(originalIndex), + sourceVersion: plan.sourceVersion, + }; + for (const { event: loadedEvent, originalIndex } of iterateTranscriptEvents( + transcriptPath, + allowMalformedPrefix, + )) { + let event = loadedEvent; + let recognizedEvent: FileEntry | undefined; + if (originalIndex === plan.headerIndex) { + const legacyHeader = event as unknown as Record; + const canonicalHeader: Record = { + ...legacyHeader, + id: sessionId, + type: "session", + }; + delete canonicalHeader.sessionId; + event = canonicalHeader as unknown as FileEntry; + recognizedEvent = event; + } else { + // Reuse the runtime partition contract one row at a time. The + // synthetic header carries the source version without being emitted. + recognizedEvent = partitionSessionFileEntries([classificationHeader, event]) + .fileEntriesByOriginalIndex[1]; + } + + if (recognizedEvent) { + migrateSessionFileEntryToCurrentVersion(recognizedEvent, originalIndex, migrationState); + if ( + recognizedEvent.type !== "session" && + plan.compactionTargetIndexes.has(originalIndex) + ) { + migratedTargetIds.set(originalIndex, recognizedEvent.id); + } + event = recognizedEvent; + } + yield event; + } + assertTranscriptFileUnchanged(transcriptPath, sourceFingerprint); + }, + }; +} + +type TranscriptImportPlan = { + compactionTargetIndexes: Set; + headerIndex: number; + sourceVersion: number; +}; + +class TranscriptImportLimitError extends Error {} + +type TranscriptFileFingerprint = { + ctimeNs: bigint; + dev: bigint; + ino: bigint; + mtimeNs: bigint; + size: bigint; +}; + +function readTranscriptFileFingerprint(transcriptPath: string): TranscriptFileFingerprint { + const stat = fs.statSync(transcriptPath, { bigint: true }); + return { + ctimeNs: stat.ctimeNs, + dev: stat.dev, + ino: stat.ino, + mtimeNs: stat.mtimeNs, + size: stat.size, + }; +} + +function assertTranscriptFileUnchanged( + transcriptPath: string, + expected: TranscriptFileFingerprint, +): void { + const current = readTranscriptFileFingerprint(transcriptPath); + if ( + current.ctimeNs !== expected.ctimeNs || + current.dev !== expected.dev || + current.ino !== expected.ino || + current.mtimeNs !== expected.mtimeNs || + current.size !== expected.size + ) { + throw new Error( + "Legacy transcript changed during import; stop active session writers and rerun `openclaw doctor --fix`.", + ); + } +} + +function planTranscriptImport( + transcriptPath: string, + allowMalformedPrefix: boolean, +): TranscriptImportPlan { + const plan: TranscriptImportPlan = { + compactionTargetIndexes: new Set(), + headerIndex: -1, + sourceVersion: 1, + }; + for (const { event, originalIndex } of iterateTranscriptEvents( + transcriptPath, + allowMalformedPrefix, + )) { + if (plan.headerIndex < 0 && isRecord(event) && event.type === "session") { + plan.headerIndex = originalIndex; + plan.sourceVersion = typeof event.version === "number" ? event.version : 1; + } + if ( + isRecord(event) && + event.type === "compaction" && + Number.isInteger(event.firstKeptEntryIndex) && + Number(event.firstKeptEntryIndex) >= 0 + ) { + const targetIndex = Number(event.firstKeptEntryIndex); + if ( + !plan.compactionTargetIndexes.has(targetIndex) && + plan.compactionTargetIndexes.size >= MAX_LEGACY_COMPACTION_TARGETS + ) { + throw new TranscriptImportLimitError( + `Transcript has more than ${MAX_LEGACY_COMPACTION_TARGETS} legacy compaction targets`, + ); + } + plan.compactionTargetIndexes.add(targetIndex); + } + } + return plan; +} + +function* iterateTranscriptEvents( + transcriptPath: string, + allowMalformedPrefix: boolean, +): Generator<{ event: FileEntry; originalIndex: number }> { + let originalIndex = 0; + try { + for (const line of iterateJsonlLinesSync(transcriptPath)) { + const parsed = parseJsonlLine(line); + if (!parsed) { + continue; + } + yield { + event: normalizeLoadedFileEntry(parsed as FileEntry), + originalIndex, + }; + originalIndex += 1; + } + } catch (error) { + if (!allowMalformedPrefix || error instanceof TranscriptImportLimitError) { + throw error; + } + // The caller records the malformed transcript issue; keep the readable prefix. + } +} + export function readSqliteEntryCount(target: SessionStoreTarget): number { const result = readOnlySqliteSessionEntries(target); return result.ok ? result.summaries.length : 0; diff --git a/src/commands/doctor-session-sqlite.test.ts b/src/commands/doctor-session-sqlite.test.ts index 103bad2d7bd6..8bf3b1d7c02e 100644 --- a/src/commands/doctor-session-sqlite.test.ts +++ b/src/commands/doctor-session-sqlite.test.ts @@ -6,6 +6,7 @@ import type { DatabaseSync } from "node:sqlite"; import { expectDefined } from "@openclaw/normalization-core"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { useAutoCleanupTempDirTracker } from "../../test/helpers/temp-dir.js"; +import { CURRENT_SESSION_VERSION, SessionManager } from "../agents/sessions/session-manager.js"; import { loadExactSqliteSessionEntry, loadSqliteTranscriptEventsSync, @@ -34,6 +35,7 @@ import { type ActiveSessionSqliteMigrationRun, } from "./doctor-session-sqlite-migration-run.js"; import { + createTranscriptEventReader, readOnlySqliteSessionEntries, resolveTargetSqlitePath, } from "./doctor-session-sqlite-readers.js"; @@ -415,7 +417,7 @@ describe("runDoctorSessionSqlite", () => { } }); - it("repairs legacy message and route shapes at the import boundary", async () => { + it("repairs legacy transcript and route shapes at the import boundary", async () => { const store = createLegacyStore({ entryOverrides: { route: "stale-custom-slot", @@ -423,7 +425,9 @@ describe("runDoctorSessionSqlite", () => { }, transcriptLines: [ '{"type":"session","sessionId":"session-1"}', + '{"type":"plugin_state","id":"opaque-1","payload":{"keep":"exact"}}', '{"type":"message","id":"m1","parentId":null,"message":{"role":"assistant","content":"legacy string"}}', + '{"type":"compaction","summary":"legacy summary","firstKeptEntryIndex":2,"tokensBefore":42}', ], }); @@ -448,9 +452,44 @@ describe("runDoctorSessionSqlite", () => { storePath: store.storePath, }); const message = events.find((event) => (event as { type?: string }).type === "message") as { + id?: string; message?: { content?: unknown }; }; + const compaction = events.find( + (event) => (event as { type?: string }).type === "compaction", + ) as { firstKeptEntryId?: string; parentId?: string }; + expect(events[0]).toMatchObject({ + id: "session-1", + type: "session", + version: CURRENT_SESSION_VERSION, + }); + expect(events[0]).not.toHaveProperty("sessionId"); + expect(events[1]).toEqual({ + id: "opaque-1", + payload: { keep: "exact" }, + type: "plugin_state", + }); expect(message?.message?.content).toEqual([{ type: "text", text: "legacy string" }]); + expect(compaction).toMatchObject({ + firstKeptEntryId: message.id, + parentId: message.id, + }); + const manager = SessionManager.open( + { + agentId: "main", + sessionId: "session-1", + sessionKey: "agent:main:main", + storePath: store.storePath, + }, + store.tempDir, + ); + expect( + manager.appendMessage({ + content: "post-import message", + role: "user", + timestamp: Date.now(), + }), + ).toEqual(expect.any(String)); closeOpenClawAgentDatabasesForTest(); const sqlite = nodeSqlite.requireNodeSqlite(); const migrated = new sqlite.DatabaseSync( @@ -473,6 +512,40 @@ describe("runDoctorSessionSqlite", () => { } }); + it("aborts import when the legacy transcript changes between passes", () => { + const store = createLegacyStore(); + const realStatSync = fs.statSync.bind(fs); + let fingerprintReads = 0; + const statSpy = vi.spyOn(fs, "statSync").mockImplementation(((candidate, options) => { + const stat = realStatSync(candidate, options as never); + if ( + path.resolve(String(candidate)) === path.resolve(store.transcriptPath) && + (options as { bigint?: boolean } | undefined)?.bigint === true + ) { + fingerprintReads += 1; + if (fingerprintReads === 2) { + fs.appendFileSync(store.transcriptPath, '{"type":"custom","customType":"late"}\n'); + } + } + return stat; + }) as typeof fs.statSync); + + try { + const events: unknown[] = []; + expect(() => + createTranscriptEventReader( + store.transcriptPath, + "session-1", + )((event) => { + events.push(event); + }), + ).toThrow(/stop active session writers and rerun `openclaw doctor --fix`/); + expect(events).toEqual([]); + } finally { + statSpy.mockRestore(); + } + }); + it("preserves the legacy transcript mtime as the SQLite mutation watermark", async () => { const store = createLegacyStore(); const transcriptMtimeMs = 1_700_000_000_000; @@ -2024,7 +2097,12 @@ describe("runDoctorSessionSqlite", () => { ); it("imports aliases that share one legacy transcript before archiving it", async () => { - const store = createLegacyStore(); + const store = createLegacyStore({ + transcriptLines: [ + '{"type":"session","sessionId":"session-1"}', + '{"type":"message","message":{"role":"user","content":"shared legacy message"}}', + ], + }); const legacyStore = JSON.parse(fs.readFileSync(store.storePath, "utf-8")) as Record< string, unknown diff --git a/src/commands/doctor-session-sqlite.ts b/src/commands/doctor-session-sqlite.ts index c2e32638cda0..88d4f3a15bab 100644 --- a/src/commands/doctor-session-sqlite.ts +++ b/src/commands/doctor-session-sqlite.ts @@ -90,7 +90,10 @@ type LegacySessionRecord = { transcriptPath?: string; }; -/** Runs the targeted doctor SQLite session migration/inspection submode. */ +/** + * Runs the targeted doctor SQLite session migration/inspection submode. + * Destructive production callers hold the Gateway/SQLite-maintenance state lock for the full call. + */ export async function runDoctorSessionSqlite( options: DoctorSessionSqliteOptions, ): Promise { @@ -326,13 +329,14 @@ async function inspectOrMigrateTarget(params: { appendSqliteDbStats(params.target, report); return report; } + const importedTranscriptSources = new Set(); for (const record of records) { if (params.mode === "dry-run") { countLegacyTranscript(record, report); continue; } if (params.mode === "import") { - await importLegacySessionRecord(params.target, record, report); + await importLegacySessionRecord(params.target, record, report, importedTranscriptSources); continue; } validateLegacySessionRecord(params.target, record, report); @@ -576,9 +580,15 @@ async function importLegacySessionRecord( target: SessionStoreTarget, record: LegacySessionRecord, report: DoctorSessionSqliteTargetReport, + importedTranscriptSources: Set, ): Promise { const result = countTranscriptEvents(record); const transcriptMtimeMs = readLegacyTranscriptMtimeMs(record); + const transcriptSourceKey = record.transcriptPath + ? `${record.entry.sessionId}\0${record.transcriptPath}` + : undefined; + const shouldImportTranscript = + transcriptSourceKey !== undefined && !importedTranscriptSources.has(transcriptSourceKey); if (result.status === "missing") { if (markAlreadyMigratedTranscript(target, record, report)) { return; @@ -603,11 +613,19 @@ async function importLegacySessionRecord( entry: record.entry, sessionKey: record.sessionKey, storePath: target.storePath, - ...(record.transcriptPath - ? { readTranscriptEvents: createTranscriptEventPrefixReader(record.transcriptPath) } + ...(record.transcriptPath && shouldImportTranscript + ? { + readTranscriptEvents: createTranscriptEventPrefixReader( + record.transcriptPath, + record.entry.sessionId, + ), + } : {}), ...(transcriptMtimeMs !== undefined ? { transcriptMtimeMs } : {}), }); + if (transcriptSourceKey) { + importedTranscriptSources.add(transcriptSourceKey); + } report.importedEntries += 1; report.importedTranscriptEvents += imported.transcriptEvents; report.issues.push({ @@ -622,11 +640,19 @@ async function importLegacySessionRecord( entry: record.entry, sessionKey: record.sessionKey, storePath: target.storePath, - ...(record.transcriptPath && result.status === "ok" - ? { readTranscriptEvents: createTranscriptEventReader(record.transcriptPath) } + ...(record.transcriptPath && result.status === "ok" && shouldImportTranscript + ? { + readTranscriptEvents: createTranscriptEventReader( + record.transcriptPath, + record.entry.sessionId, + ), + } : {}), ...(transcriptMtimeMs !== undefined ? { transcriptMtimeMs } : {}), }); + if (transcriptSourceKey) { + importedTranscriptSources.add(transcriptSourceKey); + } report.importedEntries += 1; report.importedTranscriptEvents += imported.transcriptEvents; }