perf: Doctor rescans large session stores per repair group (#117068)

* perf(doctor): batch canonical session repairs

* perf(doctor): share delivery identity inventory

src/commands/doctor-session-canonical-keys.test.ts

@@ -79,6 +79,7 @@ describe("doctor canonical session-key repair", () => {
 
       expect(await repairCanonicalSessionKeys({ apply: true, cfg, env })).toMatchObject({
         foundGroups: 2,
+        repairBatches: 1,
         removedRows: 1,
         repairedGroups: 2,
       });
@@ -117,6 +118,48 @@ describe("doctor canonical session-key repair", () => {
     });
   });
 
+  it("bounds same-database repair batches while collapsing whole-store projections", async () => {
+    await withStateDirEnv("openclaw-doctor-canonical-batches-", async ({ stateDir }) => {
+      const env = { ...process.env, OPENCLAW_STATE_DIR: stateDir };
+      const storeTemplate = path.join(stateDir, "agents", "{agentId}", "sessions.json");
+      const storePath = resolveStorePath(storeTemplate, { agentId: "main", env });
+      const cfg = {
+        agents: { list: [{ id: "main", default: true }] },
+        session: { store: storeTemplate },
+      } as OpenClawConfig;
+      for (let index = 0; index < 65; index += 1) {
+        const target = `!BatchRoom${index}:example.org`;
+        const canonicalKey = `agent:main:matrix:channel:${target}`;
+        insertLegacySession({
+          agentId: "main",
+          entry: {
+            chatType: "channel",
+            delivery: normalizeSessionDeliveryState({
+              context: { channel: "matrix", to: target },
+            }),
+            sessionId: `batch-session-${index}`,
+            updatedAt: index,
+          },
+          env,
+          sessionKey: canonicalKey.toLowerCase(),
+          storePath,
+        });
+      }
+
+      expect(await repairCanonicalSessionKeys({ apply: true, cfg, env })).toMatchObject({
+        foundGroups: 65,
+        repairBatches: 2,
+        removedRows: 65,
+        repairedGroups: 65,
+      });
+      expect(await repairCanonicalSessionKeys({ apply: true, cfg, env })).toMatchObject({
+        foundGroups: 0,
+        repairBatches: 0,
+        repairedGroups: 0,
+      });
+    });
+  });
+
   it("is a no-op for fresh stores and remains idempotent after repair", async () => {
     await withStateDirEnv("openclaw-doctor-canonical-fresh-", async ({ stateDir }) => {
       const env = { ...process.env, OPENCLAW_STATE_DIR: stateDir };

src/commands/doctor-session-canonical-keys.ts

@@ -8,6 +8,7 @@ import {
   listSessionGenerationIdsForCanonicalRepair,
   loadTranscriptEvents,
   rehomeSessionDeliveryReferencesForCanonicalRepair,
+  rehomeSessionDeliveryReferencesForCanonicalRepairBatch,
 } from "../config/sessions/session-accessor.js";
 import type { SessionEntryLifecycleRemoval } from "../config/sessions/session-accessor.lifecycle-types.js";
 import { writeSqliteTranscriptArchive } from "../config/sessions/session-accessor.sqlite-archive.js";
@@ -28,7 +29,10 @@ import {
   resolveStoredSessionKeyForAgentStore,
 } from "../gateway/session-store-key.js";
 import { normalizeAgentId, parseAgentSessionKey } from "../routing/session-key.js";
-import { openOpenClawAgentDatabase } from "../state/openclaw-agent-db.js";
+import {
+  openOpenClawAgentDatabase,
+  type OpenClawAgentDatabase,
+} from "../state/openclaw-agent-db.js";
 import { resolveTargetSqlitePath } from "./doctor-session-sqlite-readers.js";
 
 type CanonicalSessionCandidate = {
@@ -67,11 +71,19 @@ function createCanonicalRepairRemoval(
 export type CanonicalSessionKeyRepairReport = {
   archivedTranscriptDirectories: string[];
   foundGroups: number;
+  repairBatches: number;
   removedRows: number;
   repairedGroups: number;
   scannedStores: number;
 };
 
+type CanonicalSessionRepairGroup = {
+  candidates: CanonicalSessionCandidate[];
+  removedRows: number;
+};
+
+const CANONICAL_SESSION_REPAIR_BATCH_GROUP_LIMIT = 64;
+
 type CanonicalSessionStore = {
   agentId: string;
   sqlitePath: string;
@@ -305,7 +317,7 @@ function selectCanonicalSessionCandidate(
 function groupRepairCandidates(
   candidates: readonly CanonicalSessionCandidate[],
   params: { cfg: OpenClawConfig; env: NodeJS.ProcessEnv },
-) {
+): CanonicalSessionRepairGroup[] {
   const byCanonicalKey = new Map<string, CanonicalSessionCandidate[]>();
   for (const candidate of candidates) {
     const sentinelOwner =
@@ -349,6 +361,139 @@ function groupRepairCandidates(
   });
 }
 
+type SingleDatabaseCanonicalRepairGroup = {
+  candidates: readonly CanonicalSessionCandidate[];
+  selected: NonNullable<ReturnType<typeof selectCanonicalSessionCandidate>>;
+};
+
+function resolveSingleDatabaseCanonicalRepairGroup(
+  candidates: readonly CanonicalSessionCandidate[],
+  params: { cfg: OpenClawConfig; env: NodeJS.ProcessEnv },
+): SingleDatabaseCanonicalRepairGroup | undefined {
+  const selected = selectCanonicalSessionCandidate(candidates, params);
+  if (
+    !selected ||
+    selected.winner.sqlitePath !== selected.destination.sqlitePath ||
+    candidates.some((candidate) => candidate.sqlitePath !== selected.destination.sqlitePath)
+  ) {
+    return undefined;
+  }
+  return { candidates, selected };
+}
+
+function createCanonicalDestinationRemovals(
+  candidates: readonly CanonicalSessionCandidate[],
+  selected: NonNullable<ReturnType<typeof selectCanonicalSessionCandidate>>,
+): SessionEntryLifecycleRemoval[] {
+  const relatedSessionIds = new Set(
+    [selected.entry.sessionId, selected.entry.previousSessionId].filter(
+      (value): value is string => typeof value === "string" && value.length > 0,
+    ),
+  );
+  return candidates
+    .filter(
+      (candidate) =>
+        candidate.sessionKey !== selected.winner.canonicalKey ||
+        candidate.rawEntryJson !== undefined,
+    )
+    .map((candidate) =>
+      createCanonicalRepairRemoval(candidate, {
+        archiveRemovedTranscript: !relatedSessionIds.has(candidate.entry.sessionId),
+        deleteOwnedWindows: false,
+      }),
+    );
+}
+
+function listCanonicalDestinationAliasKeys(
+  destinationStore: readonly CanonicalSessionCandidate[],
+  winner: CanonicalSessionCandidate,
+): string[] {
+  return destinationStore
+    .map((candidate) => candidate.sessionKey)
+    .filter((sessionKey) => sessionKey !== winner.canonicalKey);
+}
+
+function applyCanonicalDestinationArtifacts(params: {
+  copyWinnerAlias: boolean;
+  database: OpenClawAgentDatabase;
+  destinationStore: readonly CanonicalSessionCandidate[];
+  rehomeDeliveries: boolean;
+  winner: CanonicalSessionCandidate;
+}): void {
+  const destinationAliasKeys = listCanonicalDestinationAliasKeys(
+    params.destinationStore,
+    params.winner,
+  );
+  if (destinationAliasKeys.length > 0) {
+    if (params.rehomeDeliveries) {
+      rehomeSessionDeliveryReferencesForCanonicalRepair(
+        params.database,
+        params.winner.canonicalKey,
+        destinationAliasKeys,
+      );
+    }
+    copySessionNodeArtifactsForRepair(
+      params.database,
+      params.database,
+      destinationAliasKeys,
+      params.winner.canonicalKey,
+      { includeMembers: false },
+    );
+  }
+  if (!params.copyWinnerAlias || params.winner.sessionKey === params.winner.canonicalKey) {
+    return;
+  }
+  deleteSessionMembersForRepair(params.database, params.winner.canonicalKey);
+  copySessionNodeArtifactsForRepair(
+    params.database,
+    params.database,
+    [params.winner.sessionKey],
+    params.winner.canonicalKey,
+  );
+}
+
+async function repairCanonicalSessionGroupsInSingleDatabase(
+  groups: readonly SingleDatabaseCanonicalRepairGroup[],
+): Promise<string[]> {
+  const first = groups[0];
+  if (!first) {
+    return [];
+  }
+  const destination = first.selected.destination;
+  const result = await applySessionEntryLifecycleMutation({
+    agentId: destination.agentId,
+    allowCanonicalRepair: true,
+    afterUpsertsInTransaction: (database) => {
+      rehomeSessionDeliveryReferencesForCanonicalRepairBatch(
+        database,
+        groups.map((group) => ({
+          canonicalKey: group.selected.winner.canonicalKey,
+          previousKeys: listCanonicalDestinationAliasKeys(group.candidates, group.selected.winner),
+        })),
+      );
+      for (const group of groups) {
+        applyCanonicalDestinationArtifacts({
+          copyWinnerAlias: true,
+          database,
+          destinationStore: group.candidates,
+          rehomeDeliveries: false,
+          winner: group.selected.winner,
+        });
+      }
+    },
+    removals: groups.flatMap((group) =>
+      createCanonicalDestinationRemovals(group.candidates, group.selected),
+    ),
+    skipMaintenance: true,
+    storePath: destination.storePath,
+    upserts: groups.map((group) => ({
+      entry: group.selected.entry,
+      sessionKey: group.selected.winner.canonicalKey,
+    })),
+  });
+  return result.archivedTranscriptDirectories;
+}
+
 async function repairCanonicalSessionGroup(
   candidates: readonly CanonicalSessionCandidate[],
   params: { cfg: OpenClawConfig; env: NodeJS.ProcessEnv },
@@ -422,11 +567,6 @@ async function repairCanonicalSessionGroup(
       }
     }
   }
-  const relatedSessionIds = new Set(
-    [selected.entry.sessionId, selected.entry.previousSessionId].filter(
-      (value): value is string => typeof value === "string" && value.length > 0,
-    ),
-  );
   setCanonicalSqliteSessionMainKey(
     openOpenClawAgentDatabase({ agentId: destination.agentId, path: destination.sqlitePath }),
     params.cfg.session?.mainKey,
@@ -435,35 +575,13 @@ async function repairCanonicalSessionGroup(
     agentId: destination.agentId,
     allowCanonicalRepair: true,
     afterUpsertsInTransaction: (destinationDatabase) => {
-      const destinationAliasKeys = destinationStore
-        .map((candidate) => candidate.sessionKey)
-        .filter((sessionKey) => sessionKey !== winner.canonicalKey);
-      if (destinationAliasKeys.length > 0) {
-        rehomeSessionDeliveryReferencesForCanonicalRepair(
-          destinationDatabase,
-          winner.canonicalKey,
-          destinationAliasKeys,
-        );
-        copySessionNodeArtifactsForRepair(
-          destinationDatabase,
-          destinationDatabase,
-          destinationAliasKeys,
-          winner.canonicalKey,
-          { includeMembers: false },
-        );
-      }
-      if (
-        winner.sqlitePath === destination.sqlitePath &&
-        winner.sessionKey !== winner.canonicalKey
-      ) {
-        deleteSessionMembersForRepair(destinationDatabase, winner.canonicalKey);
-        copySessionNodeArtifactsForRepair(
-          destinationDatabase,
-          destinationDatabase,
-          [winner.sessionKey],
-          winner.canonicalKey,
-        );
-      }
+      applyCanonicalDestinationArtifacts({
+        copyWinnerAlias: winner.sqlitePath === destination.sqlitePath,
+        database: destinationDatabase,
+        destinationStore,
+        rehomeDeliveries: true,
+        winner,
+      });
       if (winner.sqlitePath !== destination.sqlitePath) {
         copySessionOwnedStateForCanonicalRepair({
           canonicalKey: winner.canonicalKey,
@@ -476,17 +594,7 @@ async function repairCanonicalSessionGroup(
         });
       }
     },
-    removals: destinationStore
-      .filter(
-        (candidate) =>
-          candidate.sessionKey !== winner.canonicalKey || candidate.rawEntryJson !== undefined,
-      )
-      .map((candidate) =>
-        createCanonicalRepairRemoval(candidate, {
-          archiveRemovedTranscript: !relatedSessionIds.has(candidate.entry.sessionId),
-          deleteOwnedWindows: false,
-        }),
-      ),
+    removals: createCanonicalDestinationRemovals(destinationStore, selected),
     skipMaintenance: true,
     storePath: destination.storePath,
     upserts: [{ entry: selected.entry, sessionKey: winner.canonicalKey }],
@@ -538,6 +646,7 @@ export async function repairCanonicalSessionKeys(params: {
     env,
   });
   const archivedTranscriptDirectories = new Set<string>();
+  let repairBatches = 0;
   let repairedGroups = 0;
   if (params.apply) {
     for (const store of stores) {
@@ -550,19 +659,65 @@ export async function repairCanonicalSessionKeys(params: {
   const candidates = collectCanonicalSessionCandidates({ cfg: params.cfg, env }, stores);
   const repairGroups = groupRepairCandidates(candidates, { cfg: params.cfg, env });
   if (params.apply) {
-    for (const group of repairGroups) {
-      for (const directory of await repairCanonicalSessionGroup(group.candidates, {
+    let index = 0;
+    while (index < repairGroups.length) {
+      const group = repairGroups[index];
+      if (!group) {
+        break;
+      }
+      const singleDatabaseGroup = resolveSingleDatabaseCanonicalRepairGroup(group.candidates, {
         cfg: params.cfg,
         env,
-      })) {
+      });
+      if (!singleDatabaseGroup) {
+        for (const directory of await repairCanonicalSessionGroup(group.candidates, {
+          cfg: params.cfg,
+          env,
+        })) {
+          archivedTranscriptDirectories.add(directory);
+        }
+        index += 1;
+        repairBatches += 1;
+        repairedGroups += 1;
+        continue;
+      }
+      const batch = [singleDatabaseGroup];
+      index += 1;
+      // Keep commits bounded and preserve the original order around cross-store moves, while
+      // collapsing the repeated whole-store projections for the common same-database path.
+      while (
+        index < repairGroups.length &&
+        batch.length < CANONICAL_SESSION_REPAIR_BATCH_GROUP_LIMIT
+      ) {
+        const nextGroup = repairGroups[index];
+        if (!nextGroup) {
+          break;
+        }
+        const nextSingleDatabaseGroup = resolveSingleDatabaseCanonicalRepairGroup(
+          nextGroup.candidates,
+          { cfg: params.cfg, env },
+        );
+        if (
+          !nextSingleDatabaseGroup ||
+          nextSingleDatabaseGroup.selected.destination.sqlitePath !==
+            singleDatabaseGroup.selected.destination.sqlitePath
+        ) {
+          break;
+        }
+        batch.push(nextSingleDatabaseGroup);
+        index += 1;
+      }
+      for (const directory of await repairCanonicalSessionGroupsInSingleDatabase(batch)) {
         archivedTranscriptDirectories.add(directory);
       }
-      repairedGroups += 1;
+      repairBatches += 1;
+      repairedGroups += batch.length;
     }
   }
   return {
     archivedTranscriptDirectories: [...archivedTranscriptDirectories].toSorted(),
     foundGroups: repairGroups.length,
+    repairBatches,
     removedRows: repairGroups.reduce((total, group) => total + group.removedRows, 0),
     repairedGroups,
     scannedStores: stores.length,

src/commands/doctor-session-transcripts.test.ts

@@ -116,6 +116,7 @@ describe("doctor session transcript repair", () => {
     repairCanonicalSessionKeys.mockReset().mockResolvedValue({
       archivedTranscriptDirectories: [],
       foundGroups: 0,
+      repairBatches: 0,
       removedRows: 0,
       repairedGroups: 0,
       scannedStores: 0,

src/commands/doctor-session-transcripts.ts

@@ -526,6 +526,7 @@ async function noteSessionSqliteMigrationHealth(params: {
   let canonicalKeyReport: CanonicalSessionKeyRepairReport = {
     archivedTranscriptDirectories: [],
     foundGroups: 0,
+    repairBatches: 0,
     removedRows: 0,
     repairedGroups: 0,
     scannedStores: 0,
@@ -585,7 +586,7 @@ async function noteSessionSqliteMigrationHealth(params: {
   if (canonicalKeyReport.foundGroups > 0) {
     note(
       params.shouldRepair
-        ? `- Canonicalized ${canonicalKeyReport.repairedGroups} session-key group(s), removed ${canonicalKeyReport.removedRows} duplicate or alias row(s), and preserved cross-store history in ${canonicalKeyReport.archivedTranscriptDirectories.length} archive director${canonicalKeyReport.archivedTranscriptDirectories.length === 1 ? "y" : "ies"}.`
+        ? `- Canonicalized ${canonicalKeyReport.repairedGroups} session-key group(s) in ${canonicalKeyReport.repairBatches} transaction batch(es), removed ${canonicalKeyReport.removedRows} duplicate or alias row(s), and preserved cross-store history in ${canonicalKeyReport.archivedTranscriptDirectories.length} archive director${canonicalKeyReport.archivedTranscriptDirectories.length === 1 ? "y" : "ies"}.`
         : `- Found ${canonicalKeyReport.foundGroups} non-canonical or duplicate session-key group(s). Run "openclaw doctor --fix" to preserve their history and canonicalize the rows.`,
       "Session SQLite",
     );

src/config/sessions/session-accessor.entry.ts

@@ -16,6 +16,7 @@ import {
   listSqliteSessionEntries,
   listSqliteSessionEntriesForCanonicalRepair as listSessionEntriesForCanonicalRepair,
   rehomeSqliteSessionDeliveryReferencesForCanonicalRepair as rehomeSessionDeliveryReferencesForCanonicalRepair,
+  rehomeSqliteSessionDeliveryReferencesForCanonicalRepairBatch as rehomeSessionDeliveryReferencesForCanonicalRepairBatch,
   listSqliteSessionEntriesReadOnly as listSessionEntriesReadOnly,
   listSqliteSessionEntryKeysReadOnly as listSessionEntryKeysReadOnly,
   loadExactSqliteSessionEntry as loadExactSessionEntry,
@@ -63,6 +64,7 @@ export {
   listSessionEntriesReadOnly,
   listSessionEntriesForCanonicalRepair,
   rehomeSessionDeliveryReferencesForCanonicalRepair,
+  rehomeSessionDeliveryReferencesForCanonicalRepairBatch,
   listSessionEntryKeysReadOnly,
   loadExactSessionEntry,
   loadExactSessionEntryReadOnly,

src/config/sessions/session-accessor.sqlite-canonical-repair.ts

@@ -140,31 +140,64 @@ export function rehomeSqliteSessionDeliveryReferencesForCanonicalRepair(
   canonicalKey: string,
   previousKeys: readonly string[],
 ): void {
-  const ownedKeys = new Set([canonicalKey, ...previousKeys]);
-  const db = getSessionKysely(database.db);
-  const competingIdentities = new Set(
-    executeSqliteQuerySync(
-      database.db,
-      db.selectFrom("session_nodes").select("session_key"),
-    ).rows.flatMap((row) =>
-      ownedKeys.has(row.session_key) ? [] : [normalizeStoreSessionKey(row.session_key.trim())],
-    ),
-  );
-  const aliases = resolveSqliteCanonicalRepairLookupKeys(canonicalKey, previousKeys).filter(
-    (key) =>
-      key !== canonicalKey &&
-      (ownedKeys.has(key) || !competingIdentities.has(normalizeStoreSessionKey(key.trim()))),
-  );
-  if (aliases.length === 0) {
+  rehomeSqliteSessionDeliveryReferencesForCanonicalRepairBatch(database, [
+    { canonicalKey, previousKeys },
+  ]);
+}
+
+/** Doctor-only batched delivery rewrite with one session identity inventory per database. */
+export function rehomeSqliteSessionDeliveryReferencesForCanonicalRepairBatch(
+  database: OpenClawAgentDatabase,
+  repairs: readonly { canonicalKey: string; previousKeys: readonly string[] }[],
+): void {
+  if (repairs.length === 0) {
     return;
   }
-  executeSqliteQuerySync(
+  const db = getSessionKysely(database.db);
+  const storedSessionKeys = executeSqliteQuerySync(
     database.db,
-    db
-      .updateTable("conversation_deliveries")
-      .set({ source_session_key: canonicalKey })
-      .where("source_session_key", "in", aliases),
-  );
+    db.selectFrom("session_nodes").select("session_key"),
+  ).rows.map((row) => row.session_key);
+  const storedSessionKeySet = new Set(storedSessionKeys);
+  const identityCounts = new Map<string, number>();
+  for (const sessionKey of storedSessionKeys) {
+    const identity = normalizeStoreSessionKey(sessionKey.trim());
+    identityCounts.set(identity, (identityCounts.get(identity) ?? 0) + 1);
+  }
+  for (const repair of repairs) {
+    const ownedKeys = new Set([repair.canonicalKey, ...repair.previousKeys]);
+    const ownedIdentityCounts = new Map<string, number>();
+    for (const sessionKey of ownedKeys) {
+      if (!storedSessionKeySet.has(sessionKey)) {
+        continue;
+      }
+      const identity = normalizeStoreSessionKey(sessionKey.trim());
+      ownedIdentityCounts.set(identity, (ownedIdentityCounts.get(identity) ?? 0) + 1);
+    }
+    const aliases = resolveSqliteCanonicalRepairLookupKeys(
+      repair.canonicalKey,
+      repair.previousKeys,
+    ).filter((key) => {
+      if (key === repair.canonicalKey) {
+        return false;
+      }
+      if (ownedKeys.has(key)) {
+        return true;
+      }
+      const identity = normalizeStoreSessionKey(key.trim());
+      return (identityCounts.get(identity) ?? 0) <= (ownedIdentityCounts.get(identity) ?? 0);
+    });
+    if (aliases.length === 0) {
+      continue;
+    }
+    executeSqliteQuerySync(
+      database.db,
+      db
+        .updateTable("conversation_deliveries")
+        .set({ source_session_key: repair.canonicalKey })
+        .where("source_session_key", "in", aliases),
+    );
+  }
 }
 
 type CanonicalRepairRow = Selectable<OpenClawAgentKyselyDatabase["session_nodes"]> & {

src/config/sessions/session-accessor.sqlite.ts

@@ -26,6 +26,7 @@ export {
   listSqliteSessionEntriesForCanonicalRepair,
   listSqliteSessionGenerationIdsForCanonicalRepair,
   rehomeSqliteSessionDeliveryReferencesForCanonicalRepair,
+  rehomeSqliteSessionDeliveryReferencesForCanonicalRepairBatch,
 } from "./session-accessor.sqlite-canonical-repair.js";
 export {
   cleanupSqliteSessionLifecycleArtifacts,

src/config/sessions/session-accessor.ts

@@ -127,6 +127,7 @@ export {
   listSessionEntriesReadOnly,
   listSessionEntriesForCanonicalRepair,
   rehomeSessionDeliveryReferencesForCanonicalRepair,
+  rehomeSessionDeliveryReferencesForCanonicalRepairBatch,
   listSessionEntryKeysReadOnly,
   loadExactSessionEntry,
   loadExactSessionEntryReadOnly,