fix(memory-wiki): stop duplicate bridge imports under polling (#91828)

* fix(memory-wiki): coalesce concurrent source syncs

Co-authored-by: mushuiyu_xydt <yang.haoyu@xydigit.com>

* test(memory-wiki): track alias fixture cleanup

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

extensions/memory-wiki/src/mutation-coordinator.test.ts

@@ -1,5 +1,10 @@
+import fs from "node:fs/promises";
+import path from "node:path";
 import { describe, expect, it } from "vitest";
 import { withMemoryWikiVaultMutation } from "./mutation-coordinator.js";
+import { createMemoryWikiTestHarness } from "./test-helpers.js";
+
+const { createTempDir } = createMemoryWikiTestHarness();
 
 function deferred() {
   let resolve!: () => void;
@@ -57,6 +62,40 @@ describe("withMemoryWikiVaultMutation", () => {
     await Promise.all([first, second]);
   });
 
+  it("serializes physical aliases of one vault", async () => {
+    const root = await createTempDir("memory-wiki-vault-alias-");
+    const vaultPath = path.join(root, "vault");
+    const aliasPath = path.join(root, "vault-alias");
+    await fs.mkdir(vaultPath);
+    await fs.symlink(vaultPath, aliasPath, process.platform === "win32" ? "junction" : "dir");
+    const firstEntered = deferred();
+    const releaseFirst = deferred();
+    let aliasEntered = false;
+    let first: Promise<void> | undefined;
+    let alias: Promise<void> | undefined;
+
+    try {
+      first = withMemoryWikiVaultMutation(vaultPath, async () => {
+        firstEntered.resolve();
+        await releaseFirst.promise;
+      });
+      await firstEntered.promise;
+      alias = withMemoryWikiVaultMutation(aliasPath, async () => {
+        aliasEntered = true;
+      });
+      await Promise.resolve();
+      await Promise.resolve();
+      expect(aliasEntered).toBe(false);
+
+      releaseFirst.resolve();
+      await Promise.all([first, alias]);
+      expect(aliasEntered).toBe(true);
+    } finally {
+      releaseFirst.resolve();
+      await Promise.allSettled([first, alias].filter((item) => item !== undefined));
+    }
+  });
+
   it("queues detached work after its inherited transaction has ended", async () => {
     const releaseDetached = deferred();
     const holderEntered = deferred();

extensions/memory-wiki/src/mutation-coordinator.ts

@@ -1,5 +1,6 @@
 // Memory Wiki plugin module serializes vault mutation transactions.
 import { AsyncLocalStorage } from "node:async_hooks";
+import fs from "node:fs/promises";
 import path from "node:path";
 import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
 
@@ -7,6 +8,51 @@ type ActiveVaultMutation = { active: boolean };
 
 const activeVaultMutations = new AsyncLocalStorage<ReadonlyMap<string, ActiveVaultMutation>>();
 const vaultMutationQueue = new KeyedAsyncQueue();
+const MAX_CACHED_VAULT_KEYS = 256;
+const canonicalVaultKeys = new Map<string, Promise<string>>();
+
+function normalizeCanonicalVaultKey(vaultPath: string): string {
+  return process.platform === "win32" ? vaultPath.toLowerCase() : vaultPath;
+}
+
+async function resolveCanonicalVaultKey(resolvedPath: string): Promise<string> {
+  const suffix: string[] = [];
+  let candidate = resolvedPath;
+  while (true) {
+    try {
+      const existingPath = await fs.realpath(candidate);
+      return normalizeCanonicalVaultKey(path.join(existingPath, ...suffix));
+    } catch (error) {
+      const code = (error as NodeJS.ErrnoException).code;
+      const parent = path.dirname(candidate);
+      if ((code !== "ENOENT" && code !== "ENOTDIR") || parent === candidate) {
+        return normalizeCanonicalVaultKey(resolvedPath);
+      }
+      suffix.unshift(path.basename(candidate));
+      candidate = parent;
+    }
+  }
+}
+
+export async function resolveMemoryWikiVaultMutationKey(vaultPath: string): Promise<string> {
+  const resolvedPath = path.resolve(vaultPath);
+  const cached = canonicalVaultKeys.get(resolvedPath);
+  if (cached) {
+    return await cached;
+  }
+
+  // Vault config is process-stable. Resolve physical aliases once, then keep
+  // the cache bounded so request-time mutations never freshness-poll paths.
+  const canonical = resolveCanonicalVaultKey(resolvedPath);
+  if (canonicalVaultKeys.size >= MAX_CACHED_VAULT_KEYS) {
+    const oldest = canonicalVaultKeys.keys().next().value;
+    if (oldest) {
+      canonicalVaultKeys.delete(oldest);
+    }
+  }
+  canonicalVaultKeys.set(resolvedPath, canonical);
+  return await canonical;
+}
 
 /**
  * Keep coordinated vault read-modify-write transactions isolated from concurrent work in this process.
@@ -16,7 +62,7 @@ export async function withMemoryWikiVaultMutation<T>(
   vaultPath: string,
   mutation: () => Promise<T>,
 ): Promise<T> {
-  const key = path.resolve(vaultPath);
+  const key = await resolveMemoryWikiVaultMutationKey(vaultPath);
   const active = activeVaultMutations.getStore();
   if (active?.get(key)?.active) {
     return await mutation();

extensions/memory-wiki/src/source-page-shared.test.ts

@@ -6,6 +6,19 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { renderMarkdownFence, renderWikiMarkdown } from "./markdown.js";
 import { writeImportedSourcePage } from "./source-page-shared.js";
 
+const { fsRootMock } = vi.hoisted(() => ({ fsRootMock: vi.fn() }));
+
+vi.mock("openclaw/plugin-sdk/security-runtime", async (importOriginal) => {
+  const actual = await importOriginal<typeof import("openclaw/plugin-sdk/security-runtime")>();
+  return {
+    ...actual,
+    root: (...args: Parameters<typeof actual.root>) => {
+      fsRootMock(args[0]);
+      return actual.root(...args);
+    },
+  };
+});
+
 function buildSourcePage(raw: string, updatedAt: string): string {
   return renderWikiMarkdown({
     frontmatter: {
@@ -39,6 +52,7 @@ describe("writeImportedSourcePage", () => {
 
   afterEach(async () => {
     vi.useRealTimers();
+    fsRootMock.mockClear();
     await fs.rm(suiteRoot, { recursive: true, force: true });
   });
 
@@ -72,6 +86,90 @@ describe("writeImportedSourcePage", () => {
     expect(state.entries["unsafe:source"]?.sourceUpdatedAtMs).toBe(8_700_000_000_000_000);
   });
 
+  it("skips 1,914 unchanged pages before opening the guarded vault", async () => {
+    const sourcePath = path.join(suiteRoot, "unchanged-source.md");
+    const pagePath = "sources/unchanged.md";
+    const pageAbsolutePath = path.join(suiteRoot, pagePath);
+    await fs.mkdir(path.dirname(pageAbsolutePath), { recursive: true });
+    await fs.writeFile(pageAbsolutePath, "already imported", "utf8");
+    const entry = {
+      group: "bridge" as const,
+      pagePath,
+      sourcePath,
+      sourceUpdatedAtMs: 123,
+      sourceSize: 456,
+      renderFingerprint: "unchanged",
+    };
+    const syncKeys = Array.from({ length: 1_914 }, (_, index) => `bridge:${index}`);
+    const state: Parameters<typeof writeImportedSourcePage>[0]["state"] = {
+      version: 1,
+      entries: Object.fromEntries(syncKeys.map((syncKey) => [syncKey, { ...entry }])),
+    };
+    const buildRendered = vi.fn();
+    fsRootMock.mockClear();
+
+    const results = await Promise.all(
+      syncKeys.map((syncKey) =>
+        writeImportedSourcePage({
+          vaultRoot: suiteRoot,
+          syncKey,
+          sourcePath,
+          sourceUpdatedAtMs: entry.sourceUpdatedAtMs,
+          sourceSize: entry.sourceSize,
+          renderFingerprint: entry.renderFingerprint,
+          pagePath,
+          group: "bridge",
+          state,
+          buildRendered,
+        }),
+      ),
+    );
+
+    expect(fsRootMock).not.toHaveBeenCalled();
+    expect(buildRendered).not.toHaveBeenCalled();
+    expect(results).toHaveLength(1_914);
+    expect(results.every((result) => !result.changed && !result.created)).toBe(true);
+  });
+
+  it("recreates an unchanged source entry when its page is missing", async () => {
+    const sourcePath = path.join(suiteRoot, "missing-page-source.md");
+    const pagePath = "sources/missing.md";
+    await fs.writeFile(sourcePath, "restored body", "utf8");
+    const state: Parameters<typeof writeImportedSourcePage>[0]["state"] = {
+      version: 1,
+      entries: {
+        missing: {
+          group: "bridge",
+          pagePath,
+          sourcePath,
+          sourceUpdatedAtMs: 123,
+          sourceSize: 13,
+          renderFingerprint: "missing",
+        },
+      },
+    };
+    fsRootMock.mockClear();
+
+    const result = await writeImportedSourcePage({
+      vaultRoot: suiteRoot,
+      syncKey: "missing",
+      sourcePath,
+      sourceUpdatedAtMs: 123,
+      sourceSize: 13,
+      renderFingerprint: "missing",
+      pagePath,
+      group: "bridge",
+      state,
+      buildRendered: (raw) => raw,
+    });
+
+    expect(result).toEqual({ pagePath, changed: true, created: true });
+    expect(fsRootMock).toHaveBeenCalledTimes(1);
+    await expect(fs.readFile(path.join(suiteRoot, pagePath), "utf8")).resolves.toBe(
+      "restored body",
+    );
+  });
+
   it("preserves the human Notes block when an imported source page is updated", async () => {
     const sourcePath = path.join(suiteRoot, "imported.txt");
     const pagePath = "sources/imported.md";

extensions/memory-wiki/src/source-page-shared.ts

@@ -44,6 +44,20 @@ export async function writeImportedSourcePage(params: {
   state: ImportedSourceState;
   buildRendered: (raw: string, updatedAt: string) => string;
 }): Promise<{ pagePath: string; changed: boolean; created: boolean }> {
+  const shouldSkip = await shouldSkipImportedSourceWrite({
+    vaultRoot: params.vaultRoot,
+    syncKey: params.syncKey,
+    expectedPagePath: params.pagePath,
+    expectedSourcePath: params.sourcePath,
+    sourceUpdatedAtMs: params.sourceUpdatedAtMs,
+    sourceSize: params.sourceSize,
+    renderFingerprint: params.renderFingerprint,
+    state: params.state,
+  });
+  if (shouldSkip) {
+    return { pagePath: params.pagePath, changed: false, created: false };
+  }
+
   const vault = await fsRoot(params.vaultRoot);
   const pageStat = await vault.stat(params.pagePath).catch((error: unknown) => {
     if (
@@ -56,20 +70,6 @@ export async function writeImportedSourcePage(params: {
   });
   const created = !pageStat;
   const updatedAt = timestampMsToIsoString(params.sourceUpdatedAtMs) ?? new Date().toISOString();
-  const shouldSkip = await shouldSkipImportedSourceWrite({
-    vaultRoot: params.vaultRoot,
-    syncKey: params.syncKey,
-    expectedPagePath: params.pagePath,
-    expectedSourcePath: params.sourcePath,
-    sourceUpdatedAtMs: params.sourceUpdatedAtMs,
-    sourceSize: params.sourceSize,
-    renderFingerprint: params.renderFingerprint,
-    state: params.state,
-  });
-  if (shouldSkip) {
-    return { pagePath: params.pagePath, changed: false, created };
-  }
-
   const raw = await fs.readFile(params.sourcePath, "utf8");
   const rendered = params.buildRendered(raw, updatedAt);
   const existing = pageStat ? await readExistingImportedSourcePage(vault, params.pagePath) : "";

extensions/memory-wiki/src/source-sync-state.test.ts

@@ -2,7 +2,10 @@
 import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
-import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime";
+import type {
+  OpenKeyedStoreOptions,
+  PluginStateKeyedStore,
+} from "openclaw/plugin-sdk/plugin-state-runtime";
 import {
   createPluginStateKeyedStoreForTests,
   resetPluginStateStoreForTests,
@@ -13,9 +16,11 @@ import {
   configureMemoryWikiSourceSyncStateStore,
   createMemoryWikiSourceSyncStateStore,
   MEMORY_WIKI_SOURCE_SYNC_STATE_MAX_ENTRIES,
+  pruneImportedSourceEntries,
   readLegacyMemoryWikiSourceSyncState,
   readMemoryWikiSourceSyncState,
   resolveMemoryWikiSourceSyncStatePath,
+  setImportedSourceEntry,
   writeMemoryWikiSourceSyncState,
 } from "./source-sync-state.js";
 
@@ -33,6 +38,69 @@ function openStore(env: NodeJS.ProcessEnv) {
   );
 }
 
+function createCountingStore(options?: { maxEntries?: number }) {
+  const values = new Map<string, unknown>();
+  const calls = { register: 0, delete: 0, entries: 0 };
+  let openOptions: OpenKeyedStoreOptions | undefined;
+  const openKeyedStore = <T>(storeOptions: OpenKeyedStoreOptions): PluginStateKeyedStore<T> => {
+    openOptions = storeOptions;
+    return {
+      async register(key, value) {
+        calls.register += 1;
+        if (!values.has(key) && options?.maxEntries && values.size >= options.maxEntries) {
+          const oldestKey = values.keys().next().value;
+          if (oldestKey) {
+            values.delete(oldestKey);
+          }
+        }
+        values.set(key, value);
+      },
+      async registerIfAbsent(key, value) {
+        if (values.has(key)) {
+          return false;
+        }
+        values.set(key, value);
+        return true;
+      },
+      async lookup(key) {
+        return values.get(key) as T | undefined;
+      },
+      async consume(key) {
+        const value = values.get(key) as T | undefined;
+        values.delete(key);
+        return value;
+      },
+      async delete(key) {
+        calls.delete += 1;
+        return values.delete(key);
+      },
+      async entries() {
+        calls.entries += 1;
+        return [...values.entries()].map(([key, value]) => ({
+          key,
+          value: value as T,
+          createdAt: 0,
+        }));
+      },
+      async clear() {
+        values.clear();
+      },
+    };
+  };
+  return {
+    store: createMemoryWikiSourceSyncStateStore(openKeyedStore),
+    calls,
+    get openOptions() {
+      return openOptions;
+    },
+    resetCalls() {
+      calls.register = 0;
+      calls.delete = 0;
+      calls.entries = 0;
+    },
+  };
+}
+
 describe("memory wiki source sync state", () => {
   beforeEach(() => {
     resetPluginStateStoreForTests();
@@ -88,6 +156,106 @@ describe("memory wiki source sync state", () => {
     });
   });
 
+  it("persists only changed rows in a 1,914-entry state snapshot", async () => {
+    const vaultRoot = path.join(await makeTempDir(), "vault");
+    const counting = createCountingStore();
+    const entries = Object.fromEntries(
+      Array.from({ length: 1_914 }, (_, index) => [
+        `source-${index}`,
+        {
+          group: "bridge" as const,
+          pagePath: `sources/source-${index}.md`,
+          sourcePath: `/tmp/source-${index}.md`,
+          sourceUpdatedAtMs: index,
+          sourceSize: index,
+          renderFingerprint: `fingerprint-${index}`,
+        },
+      ]),
+    );
+    await writeMemoryWikiSourceSyncState(vaultRoot, { version: 1, entries }, counting.store);
+    expect(counting.openOptions?.overflowPolicy).toBe("reject-new");
+
+    const state = await readMemoryWikiSourceSyncState(vaultRoot, counting.store);
+    counting.resetCalls();
+    await writeMemoryWikiSourceSyncState(vaultRoot, state, counting.store);
+    expect(counting.calls).toEqual({ register: 0, delete: 0, entries: 0 });
+
+    const changed = state.entries["source-0"];
+    expect(changed).toBeDefined();
+    setImportedSourceEntry({
+      state,
+      syncKey: "source-0",
+      entry: { ...changed!, sourceSize: changed!.sourceSize + 1 },
+    });
+    await writeMemoryWikiSourceSyncState(vaultRoot, state, counting.store);
+    expect(counting.calls).toEqual({ register: 1, delete: 0, entries: 0 });
+
+    counting.resetCalls();
+    await pruneImportedSourceEntries({
+      vaultRoot,
+      group: "bridge",
+      activeKeys: new Set(Object.keys(state.entries).filter((key) => key !== "source-1")),
+      state,
+    });
+    await writeMemoryWikiSourceSyncState(vaultRoot, state, counting.store);
+    expect(counting.calls).toEqual({ register: 0, delete: 1, entries: 0 });
+
+    const persisted = await readMemoryWikiSourceSyncState(vaultRoot, counting.store);
+    expect(persisted.entries["source-0"]?.sourceSize).toBe(1);
+    expect(persisted.entries["source-1"]).toBeUndefined();
+    expect(Object.keys(persisted.entries)).toHaveLength(1_913);
+  });
+
+  it("deletes replaced rows before upserts at the store capacity", async () => {
+    const vaultRoot = path.join(await makeTempDir(), "vault");
+    const counting = createCountingStore({ maxEntries: 2 });
+    const makeEntry = (index: number) => ({
+      group: "bridge" as const,
+      pagePath: `sources/source-${index}.md`,
+      sourcePath: `/tmp/source-${index}.md`,
+      sourceUpdatedAtMs: index,
+      sourceSize: index,
+      renderFingerprint: `fingerprint-${index}`,
+    });
+    await writeMemoryWikiSourceSyncState(
+      vaultRoot,
+      {
+        version: 1,
+        entries: { "source-0": makeEntry(0), "source-1": makeEntry(1) },
+      },
+      counting.store,
+    );
+
+    const tracked = await readMemoryWikiSourceSyncState(vaultRoot, counting.store);
+    await pruneImportedSourceEntries({
+      vaultRoot,
+      group: "bridge",
+      activeKeys: new Set(["source-0"]),
+      state: tracked,
+    });
+    setImportedSourceEntry({
+      state: tracked,
+      syncKey: "source-2",
+      entry: makeEntry(2),
+    });
+    await writeMemoryWikiSourceSyncState(vaultRoot, tracked, counting.store);
+    await expect(readMemoryWikiSourceSyncState(vaultRoot, counting.store)).resolves.toMatchObject({
+      entries: { "source-0": makeEntry(0), "source-2": makeEntry(2) },
+    });
+
+    await writeMemoryWikiSourceSyncState(
+      vaultRoot,
+      {
+        version: 1,
+        entries: { "source-0": makeEntry(0), "source-3": makeEntry(3) },
+      },
+      counting.store,
+    );
+    await expect(readMemoryWikiSourceSyncState(vaultRoot, counting.store)).resolves.toMatchObject({
+      entries: { "source-0": makeEntry(0), "source-3": makeEntry(3) },
+    });
+  });
+
   it("keeps legacy file reads separate for doctor migration", async () => {
     const vaultRoot = await makeTempDir();
     const legacyPath = resolveMemoryWikiSourceSyncStatePath(vaultRoot);

extensions/memory-wiki/src/source-sync-state.ts

@@ -24,9 +24,23 @@ type MemoryWikiImportedSourceState = {
   entries: Record<string, MemoryWikiImportedSourceStateEntry>;
 };
 
+type MemoryWikiSourceSyncStateChanges = {
+  upsertKeys: Set<string>;
+  deleteKeys: Set<string>;
+};
+
+type MemoryWikiSourceSyncStateWritePlan = {
+  upsertKeys: string[];
+  deleteKeys: string[];
+};
+
 type MemoryWikiSourceSyncStateStore = {
   read: (vaultRoot: string) => Promise<MemoryWikiImportedSourceState>;
-  write: (vaultRoot: string, state: MemoryWikiImportedSourceState) => Promise<void>;
+  write: (
+    vaultRoot: string,
+    state: MemoryWikiImportedSourceState,
+    plan?: MemoryWikiSourceSyncStateWritePlan,
+  ) => Promise<void>;
 };
 
 type MemoryWikiSourceSyncStateRecord = MemoryWikiImportedSourceStateEntry & {
@@ -44,6 +58,10 @@ const EMPTY_STATE: MemoryWikiImportedSourceState = {
 
 let configuredSourceSyncStore: MemoryWikiSourceSyncStateStore | undefined;
 const memorySourceSyncStateByVault = new Map<string, MemoryWikiImportedSourceState>();
+const sourceSyncStateChanges = new WeakMap<
+  MemoryWikiImportedSourceState,
+  MemoryWikiSourceSyncStateChanges
+>();
 
 export function resolveMemoryWikiSourceSyncStatePath(vaultRoot: string): string {
   return path.join(vaultRoot, ".openclaw-wiki", "source-sync.json");
@@ -147,6 +165,7 @@ export function createMemoryWikiSourceSyncStateStore(
     openKeyedStore<MemoryWikiSourceSyncStateRecord>({
       namespace: MEMORY_WIKI_SOURCE_SYNC_STATE_NAMESPACE,
       maxEntries: MEMORY_WIKI_SOURCE_SYNC_STATE_MAX_ENTRIES,
+      overflowPolicy: "reject-new",
     });
 
   return {
@@ -169,26 +188,45 @@ export function createMemoryWikiSourceSyncStateStore(
       }
       return { version: 1, entries };
     },
-    async write(vaultRoot, state) {
+    async write(vaultRoot, state, plan) {
       assertSourceSyncStateWithinLimit(state);
       const vaultRootKey = resolveVaultRootKey(vaultRoot);
       const store = openStore();
-      const normalized = normalizeSourceSyncState(state);
-      const nextKeys = new Set<string>();
-      for (const [syncKey, entry] of Object.entries(normalized.entries)) {
-        const key = resolveStateEntryKey(vaultRootKey, syncKey);
-        nextKeys.add(key);
-        await store.register(key, {
-          ...entry,
-          vaultRootKey,
-          syncKey,
-        });
+      if (plan) {
+        for (const syncKey of plan.deleteKeys) {
+          await store.delete(resolveStateEntryKey(vaultRootKey, syncKey));
+        }
+        for (const syncKey of plan.upsertKeys) {
+          const entry = state.entries[syncKey];
+          if (!entry) {
+            throw new Error(`Missing tracked Memory Wiki source sync entry: ${syncKey}`);
+          }
+          await store.register(resolveStateEntryKey(vaultRootKey, syncKey), {
+            ...entry,
+            vaultRootKey,
+            syncKey,
+          });
+        }
+        return;
       }
+      const normalized = normalizeSourceSyncState(state);
+      const nextKeys = new Set(
+        Object.keys(normalized.entries).map((syncKey) =>
+          resolveStateEntryKey(vaultRootKey, syncKey),
+        ),
+      );
       for (const row of await store.entries()) {
         if (row.value.vaultRootKey === vaultRootKey && !nextKeys.has(row.key)) {
           await store.delete(row.key);
         }
       }
+      for (const [syncKey, entry] of Object.entries(normalized.entries)) {
+        await store.register(resolveStateEntryKey(vaultRootKey, syncKey), {
+          ...entry,
+          vaultRootKey,
+          syncKey,
+        });
+      }
     },
   };
 }
@@ -209,7 +247,9 @@ export async function readMemoryWikiSourceSyncState(
   vaultRoot: string,
   store?: MemoryWikiSourceSyncStateStore,
 ): Promise<MemoryWikiImportedSourceState> {
-  return await resolveSourceSyncStore(store).read(vaultRoot);
+  const state = await resolveSourceSyncStore(store).read(vaultRoot);
+  sourceSyncStateChanges.set(state, { upsertKeys: new Set(), deleteKeys: new Set() });
+  return state;
 }
 
 export async function readLegacyMemoryWikiSourceSyncState(
@@ -225,7 +265,19 @@ export async function writeMemoryWikiSourceSyncState(
   state: MemoryWikiImportedSourceState,
   store?: MemoryWikiSourceSyncStateStore,
 ): Promise<void> {
-  await resolveSourceSyncStore(store).write(vaultRoot, state);
+  const changes = sourceSyncStateChanges.get(state);
+  if (changes && changes.upsertKeys.size === 0 && changes.deleteKeys.size === 0) {
+    return;
+  }
+  const plan = changes
+    ? {
+        upsertKeys: [...changes.upsertKeys],
+        deleteKeys: [...changes.deleteKeys],
+      }
+    : undefined;
+  await resolveSourceSyncStore(store).write(vaultRoot, state, plan);
+  changes?.upsertKeys.clear();
+  changes?.deleteKeys.clear();
 }
 
 export async function shouldSkipImportedSourceWrite(params: {
@@ -272,6 +324,9 @@ export async function pruneImportedSourceEntries(params: {
     const pageAbsPath = path.join(params.vaultRoot, entry.pagePath);
     await fs.rm(pageAbsPath, { force: true }).catch(() => undefined);
     delete params.state.entries[syncKey];
+    const changes = sourceSyncStateChanges.get(params.state);
+    changes?.upsertKeys.delete(syncKey);
+    changes?.deleteKeys.add(syncKey);
     removedCount += 1;
   }
   return removedCount;
@@ -282,5 +337,19 @@ export function setImportedSourceEntry(params: {
   entry: MemoryWikiImportedSourceStateEntry;
   state: MemoryWikiImportedSourceState;
 }): void {
+  const current = params.state.entries[params.syncKey];
+  if (
+    current?.group === params.entry.group &&
+    current.pagePath === params.entry.pagePath &&
+    current.sourcePath === params.entry.sourcePath &&
+    current.sourceUpdatedAtMs === params.entry.sourceUpdatedAtMs &&
+    current.sourceSize === params.entry.sourceSize &&
+    current.renderFingerprint === params.entry.renderFingerprint
+  ) {
+    return;
+  }
   params.state.entries[params.syncKey] = params.entry;
+  const changes = sourceSyncStateChanges.get(params.state);
+  changes?.deleteKeys.delete(params.syncKey);
+  changes?.upsertKeys.add(params.syncKey);
 }

extensions/memory-wiki/src/source-sync.test.ts

@@ -1,5 +1,10 @@
 // Memory Wiki tests cover source sync plugin behavior.
+import os from "node:os";
+import path from "node:path";
 import { beforeEach, describe, expect, it, vi } from "vitest";
+import type { OpenClawConfig } from "../api.js";
+import { resolveMemoryWikiConfig } from "./config.js";
+import { withMemoryWikiVaultMutation } from "./mutation-coordinator.js";
 import { syncMemoryWikiImportedSources } from "./source-sync.js";
 
 const { syncBridgeMock, syncUnsafeLocalMock, refreshIndexesMock } = vi.hoisted(() => ({
@@ -30,6 +35,33 @@ const bridgeResult = {
   pagePaths: ["sources/alpha.md"],
 };
 
+const refreshResult = {
+  refreshed: true,
+  reason: "import-changed" as const,
+  compile: { updatedFiles: ["index.md", "sources/index.md"] },
+};
+
+const appConfig = {
+  agents: { list: [{ id: "main", default: true }] },
+} as OpenClawConfig;
+
+let vaultCounter = 0;
+
+function createConfig(
+  vaultMode: "bridge" | "unsafe-local" | "isolated" = "bridge",
+  vaultPath = path.join(os.tmpdir(), `memory-wiki-source-sync-${vaultCounter++}`),
+) {
+  return resolveMemoryWikiConfig({ vaultMode, vault: { path: vaultPath } });
+}
+
+function deferred<T>() {
+  let resolve!: (value: T) => void;
+  const promise = new Promise<T>((done) => {
+    resolve = done;
+  });
+  return { promise, resolve };
+}
+
 describe("syncMemoryWikiImportedSources", () => {
   beforeEach(() => {
     syncBridgeMock.mockReset();
@@ -40,20 +72,11 @@ describe("syncMemoryWikiImportedSources", () => {
       ...bridgeResult,
       workspaces: 0,
     });
-    refreshIndexesMock.mockResolvedValue({
-      refreshed: true,
-      reason: "import-changed",
-      compile: { updatedFiles: ["index.md", "sources/index.md"] },
-    });
+    refreshIndexesMock.mockResolvedValue(refreshResult);
   });
 
   it("routes bridge mode through bridge sync and merges refresh results", async () => {
-    const config = { vaultMode: "bridge" } as Parameters<
-      typeof syncMemoryWikiImportedSources
-    >[0]["config"];
-    const appConfig = { agents: { list: [{ id: "main", default: true }] } } as Parameters<
-      typeof syncMemoryWikiImportedSources
-    >[0]["appConfig"];
+    const config = createConfig();
 
     const result = await syncMemoryWikiImportedSources({ config, appConfig });
 
@@ -71,6 +94,172 @@ describe("syncMemoryWikiImportedSources", () => {
     });
   });
 
+  it("shares one full source and index flight across equivalent polls", async () => {
+    const config = createConfig();
+    const bridgeGate = deferred<typeof bridgeResult>();
+    syncBridgeMock.mockReturnValueOnce(bridgeGate.promise);
+
+    const requests = Array.from({ length: 32 }, () =>
+      syncMemoryWikiImportedSources({ config, appConfig }),
+    );
+    await vi.waitFor(() => expect(syncBridgeMock).toHaveBeenCalledTimes(1));
+
+    bridgeGate.resolve(bridgeResult);
+    const results = await Promise.all(requests);
+
+    expect(refreshIndexesMock).toHaveBeenCalledTimes(1);
+    expect(results.every((result) => result === results[0])).toBe(true);
+  });
+
+  it("coalesces separately resolved equivalent configs for one vault", async () => {
+    const vaultPath = path.join(os.tmpdir(), `memory-wiki-source-sync-${vaultCounter++}`);
+    const firstConfig = createConfig("bridge", vaultPath);
+    const secondConfig = createConfig("bridge", vaultPath);
+    const bridgeGate = deferred<typeof bridgeResult>();
+    syncBridgeMock.mockReturnValueOnce(bridgeGate.promise);
+
+    const first = syncMemoryWikiImportedSources({ config: firstConfig, appConfig });
+    await vi.waitFor(() => expect(syncBridgeMock).toHaveBeenCalledTimes(1));
+    const second = syncMemoryWikiImportedSources({ config: secondConfig, appConfig });
+    await Promise.resolve();
+
+    expect(syncBridgeMock).toHaveBeenCalledTimes(1);
+    bridgeGate.resolve(bridgeResult);
+    const [firstResult, secondResult] = await Promise.all([first, second]);
+    expect(refreshIndexesMock).toHaveBeenCalledTimes(1);
+    expect(secondResult).toBe(firstResult);
+  });
+
+  it("keeps sharing the active flight while indexes refresh", async () => {
+    const config = createConfig();
+    const refreshGate = deferred<typeof refreshResult>();
+    refreshIndexesMock.mockReturnValueOnce(refreshGate.promise);
+
+    const first = syncMemoryWikiImportedSources({ config, appConfig });
+    await vi.waitFor(() => expect(refreshIndexesMock).toHaveBeenCalledTimes(1));
+    const followers = Array.from({ length: 32 }, () =>
+      syncMemoryWikiImportedSources({ config, appConfig }),
+    );
+    await Promise.resolve();
+
+    expect(syncBridgeMock).toHaveBeenCalledTimes(1);
+    expect(refreshIndexesMock).toHaveBeenCalledTimes(1);
+    refreshGate.resolve(refreshResult);
+    const [firstResult, ...followerResults] = await Promise.all([first, ...followers]);
+    expect(followerResults.every((result) => result === firstResult)).toBe(true);
+  });
+
+  it("clears completed flights so later polls observe new source state", async () => {
+    const config = createConfig();
+    const secondResult = { ...bridgeResult, updatedCount: 3 };
+    syncBridgeMock.mockResolvedValueOnce(bridgeResult).mockResolvedValueOnce(secondResult);
+
+    await syncMemoryWikiImportedSources({ config, appConfig });
+    await expect(syncMemoryWikiImportedSources({ config, appConfig })).resolves.toMatchObject({
+      updatedCount: 3,
+    });
+
+    expect(syncBridgeMock).toHaveBeenCalledTimes(2);
+    expect(refreshIndexesMock).toHaveBeenCalledTimes(2);
+  });
+
+  it("waits for an existing vault mutation before starting source sync", async () => {
+    const config = createConfig();
+    const blockerEntered = deferred<void>();
+    const blockerGate = deferred<void>();
+    const blocker = withMemoryWikiVaultMutation(config.vault.path, async () => {
+      blockerEntered.resolve(undefined);
+      await blockerGate.promise;
+    });
+    await blockerEntered.promise;
+
+    const sync = syncMemoryWikiImportedSources({ config, appConfig });
+    await Promise.resolve();
+    expect(syncBridgeMock).not.toHaveBeenCalled();
+
+    blockerGate.resolve(undefined);
+    await blocker;
+    await sync;
+    expect(syncBridgeMock).toHaveBeenCalledTimes(1);
+  });
+
+  it("serializes different config snapshots for the same vault", async () => {
+    const config = createConfig();
+    const firstAppConfig = {
+      agents: { list: [{ id: "main", default: true }] },
+      update: { channel: "stable" },
+    } as OpenClawConfig;
+    const secondAppConfig = {
+      agents: { list: [{ id: "main", default: true }] },
+      update: { channel: "beta" },
+    } as OpenClawConfig;
+    const firstGate = deferred<typeof bridgeResult>();
+    const secondResult = { ...bridgeResult, pagePaths: ["sources/beta.md"] };
+    syncBridgeMock.mockReturnValueOnce(firstGate.promise).mockResolvedValueOnce(secondResult);
+
+    const first = syncMemoryWikiImportedSources({ config, appConfig: firstAppConfig });
+    await vi.waitFor(() => expect(syncBridgeMock).toHaveBeenCalledTimes(1));
+    const second = syncMemoryWikiImportedSources({ config, appConfig: secondAppConfig });
+    await Promise.resolve();
+    expect(syncBridgeMock).toHaveBeenCalledTimes(1);
+
+    firstGate.resolve(bridgeResult);
+    await first;
+    await vi.waitFor(() => expect(syncBridgeMock).toHaveBeenCalledTimes(2));
+    await expect(second).resolves.toMatchObject({ pagePaths: ["sources/beta.md"] });
+    expect(refreshIndexesMock).toHaveBeenCalledTimes(2);
+  });
+
+  it("serializes bridge and unsafe-local syncs for one vault", async () => {
+    const vaultPath = path.join(os.tmpdir(), `memory-wiki-source-sync-${vaultCounter++}`);
+    const bridgeConfig = createConfig("bridge", vaultPath);
+    const unsafeConfig = createConfig("unsafe-local", vaultPath);
+    const bridgeGate = deferred<typeof bridgeResult>();
+    syncBridgeMock.mockReturnValueOnce(bridgeGate.promise);
+
+    const bridge = syncMemoryWikiImportedSources({ config: bridgeConfig, appConfig });
+    await vi.waitFor(() => expect(syncBridgeMock).toHaveBeenCalledTimes(1));
+    const unsafe = syncMemoryWikiImportedSources({ config: unsafeConfig, appConfig });
+    await Promise.resolve();
+    expect(syncUnsafeLocalMock).not.toHaveBeenCalled();
+
+    bridgeGate.resolve(bridgeResult);
+    await bridge;
+    await vi.waitFor(() => expect(syncUnsafeLocalMock).toHaveBeenCalledTimes(1));
+    await unsafe;
+    expect(refreshIndexesMock).toHaveBeenCalledTimes(2);
+  });
+
+  it("allows different agent vaults to sync in parallel", async () => {
+    const firstConfig = createConfig();
+    const secondConfig = createConfig();
+    const firstGate = deferred<typeof bridgeResult>();
+    const secondResult = { ...bridgeResult, pagePaths: ["sources/other.md"] };
+    syncBridgeMock.mockReturnValueOnce(firstGate.promise).mockResolvedValueOnce(secondResult);
+
+    const first = syncMemoryWikiImportedSources({ config: firstConfig, appConfig });
+    await vi.waitFor(() => expect(syncBridgeMock).toHaveBeenCalledTimes(1));
+    const second = syncMemoryWikiImportedSources({ config: secondConfig, appConfig });
+    await vi.waitFor(() => expect(syncBridgeMock).toHaveBeenCalledTimes(2));
+
+    await expect(second).resolves.toMatchObject({ pagePaths: ["sources/other.md"] });
+    firstGate.resolve(bridgeResult);
+    await first;
+  });
+
+  it("clears failed flights so a later sync can retry", async () => {
+    const config = createConfig();
+    syncBridgeMock.mockRejectedValueOnce(new Error("bridge unavailable"));
+
+    await expect(syncMemoryWikiImportedSources({ config, appConfig })).rejects.toThrow(
+      "bridge unavailable",
+    );
+    await expect(syncMemoryWikiImportedSources({ config, appConfig })).resolves.toMatchObject({
+      pagePaths: bridgeResult.pagePaths,
+    });
+    expect(syncBridgeMock).toHaveBeenCalledTimes(2);
+  });
+
   it("routes unsafe-local mode through unsafe-local sync", async () => {
     const unsafeLocalResult = {
       ...bridgeResult,
@@ -83,9 +272,7 @@ describe("syncMemoryWikiImportedSources", () => {
       refreshed: false,
       reason: "auto-compile-disabled",
     });
-    const config = { vaultMode: "unsafe-local" } as Parameters<
-      typeof syncMemoryWikiImportedSources
-    >[0]["config"];
+    const config = createConfig("unsafe-local");
 
     const result = await syncMemoryWikiImportedSources({ config });
 
@@ -104,9 +291,7 @@ describe("syncMemoryWikiImportedSources", () => {
   });
 
   it("returns a no-op sync result outside imported-source modes", async () => {
-    const config = { vaultMode: "isolated" } as Parameters<
-      typeof syncMemoryWikiImportedSources
-    >[0]["config"];
+    const config = createConfig("isolated");
 
     const result = await syncMemoryWikiImportedSources({ config });
 

extensions/memory-wiki/src/source-sync.ts

@@ -6,6 +6,10 @@ import {
   type RefreshMemoryWikiIndexesResult,
 } from "./compile.js";
 import type { ResolvedMemoryWikiConfig } from "./config.js";
+import {
+  resolveMemoryWikiVaultMutationKey,
+  withMemoryWikiVaultMutation,
+} from "./mutation-coordinator.js";
 import { syncMemoryWikiUnsafeLocalSources } from "./unsafe-local.js";
 
 export type MemoryWikiImportedSourceSyncResult = BridgeMemoryWikiResult & {
@@ -14,10 +18,35 @@ export type MemoryWikiImportedSourceSyncResult = BridgeMemoryWikiResult & {
   indexRefreshReason: RefreshMemoryWikiIndexesResult["reason"];
 };
 
-export async function syncMemoryWikiImportedSources(params: {
+type SyncMemoryWikiImportedSourcesParams = {
   config: ResolvedMemoryWikiConfig;
   appConfig?: OpenClawConfig;
-}): Promise<MemoryWikiImportedSourceSyncResult> {
+};
+
+type ActiveImportedSourceSync = {
+  requestKey: string;
+  appConfig?: OpenClawConfig;
+  promise: Promise<MemoryWikiImportedSourceSyncResult>;
+};
+
+const activeImportedSourceSyncs = new Map<string, ActiveImportedSourceSync[]>();
+
+function resolveImportedSourceSyncRequestKey(
+  params: SyncMemoryWikiImportedSourcesParams,
+  vaultKey: string,
+): string {
+  return JSON.stringify({
+    ...params.config,
+    vault: {
+      ...params.config.vault,
+      path: vaultKey,
+    },
+  });
+}
+
+async function syncMemoryWikiImportedSourcesOnce(
+  params: SyncMemoryWikiImportedSourcesParams,
+): Promise<MemoryWikiImportedSourceSyncResult> {
   let syncResult: BridgeMemoryWikiResult;
   if (params.config.vaultMode === "bridge") {
     syncResult = await syncMemoryWikiBridgeSources(params);
@@ -45,3 +74,42 @@ export async function syncMemoryWikiImportedSources(params: {
     indexRefreshReason: refreshResult.reason,
   };
 }
+
+export async function syncMemoryWikiImportedSources(
+  params: SyncMemoryWikiImportedSourcesParams,
+): Promise<MemoryWikiImportedSourceSyncResult> {
+  const vaultKey = await resolveMemoryWikiVaultMutationKey(params.config.vault.path);
+  const requestKey = resolveImportedSourceSyncRequestKey(params, vaultKey);
+  const active = activeImportedSourceSyncs.get(vaultKey) ?? [];
+  const matching = active.find(
+    (entry) => entry.requestKey === requestKey && entry.appConfig === params.appConfig,
+  );
+  if (matching) {
+    return await matching.promise;
+  }
+
+  // Equivalent polls share the whole source-and-index flight. Different
+  // snapshots still queue on the common vault transaction boundary.
+  const promise = withMemoryWikiVaultMutation(params.config.vault.path, () =>
+    syncMemoryWikiImportedSourcesOnce(params),
+  );
+  const entry: ActiveImportedSourceSync = {
+    requestKey,
+    ...(params.appConfig ? { appConfig: params.appConfig } : {}),
+    promise,
+  };
+  active.push(entry);
+  activeImportedSourceSyncs.set(vaultKey, active);
+
+  try {
+    return await promise;
+  } finally {
+    const index = active.indexOf(entry);
+    if (index >= 0) {
+      active.splice(index, 1);
+    }
+    if (active.length === 0 && activeImportedSourceSyncs.get(vaultKey) === active) {
+      activeImportedSourceSyncs.delete(vaultKey);
+    }
+  }
+}