fix(sessions): migrate legacy transcripts during SQLite import (#116077)

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<string>();
-  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<string>();
+  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;
 }

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<FileEntry> {
+  // 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<number, string>();
+      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<string, unknown>;
+          const canonicalHeader: Record<string, unknown> = {
+            ...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<number>;
+  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;

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

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<DoctorSessionSqliteReport> {
@@ -326,13 +329,14 @@ async function inspectOrMigrateTarget(params: {
     appendSqliteDbStats(params.target, report);
     return report;
   }
+  const importedTranscriptSources = new Set<string>();
   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<string>,
 ): Promise<void> {
   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;
 }