fix(memory-wiki): prevent source ingest content loss during concurrent vault writes (#104209)

* fix(memory-wiki): serialize source ingest with vault mutations

* test(memory-wiki): make ingest lock proof deterministic

* test(memory-wiki): type queue spy receiver

* test(memory-wiki): keep queue spy lint-safe

* style(memory-wiki): format ingest lock proof

---------

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

extensions/memory-wiki/src/ingest.test.ts

@@ -1,12 +1,22 @@
 // Memory Wiki tests cover ingest plugin behavior.
 import fs from "node:fs/promises";
 import path from "node:path";
-import { describe, expect, it } from "vitest";
+import { KeyedAsyncQueue } from "openclaw/plugin-sdk/keyed-async-queue";
+import { describe, expect, it, vi } from "vitest";
 import { ingestMemoryWikiSource } from "./ingest.js";
+import { withMemoryWikiVaultMutation } from "./mutation-coordinator.js";
 import { createMemoryWikiTestHarness } from "./test-helpers.js";
 
 const { createTempDir, createVault } = createMemoryWikiTestHarness();
 
+function deferred() {
+  let resolve!: () => void;
+  const promise = new Promise<void>((done) => {
+    resolve = done;
+  });
+  return { promise, resolve };
+}
+
 describe("ingestMemoryWikiSource", () => {
   it("copies a local text file into sources markdown", async () => {
     const rootDir = await createTempDir("memory-wiki-ingest-");
@@ -64,4 +74,57 @@ hello from source
       "[meeting notes](sources/meeting-notes.md)",
     );
   });
+
+  it("queues behind a held vault mutation instead of writing mid-transaction", async () => {
+    const rootDir = await createTempDir("memory-wiki-ingest-lock-");
+    const inputPath = path.join(rootDir, "meeting-notes.txt");
+    await fs.writeFile(inputPath, "hello from source\n", "utf8");
+    const { config } = await createVault({
+      rootDir: path.join(rootDir, "vault"),
+    });
+    const pagePath = path.join(config.vault.path, "sources", "meeting-notes.md");
+
+    const lockEntered = deferred();
+    const releaseLock = deferred();
+    const holder = withMemoryWikiVaultMutation(config.vault.path, async () => {
+      lockEntered.resolve();
+      await releaseLock.promise;
+    });
+    await lockEntered.promise;
+
+    const ingestQueued = deferred();
+    const originalEnqueue = Object.getOwnPropertyDescriptor(KeyedAsyncQueue.prototype, "enqueue")
+      ?.value as KeyedAsyncQueue["enqueue"];
+    const enqueueSpy = vi
+      .spyOn(KeyedAsyncQueue.prototype, "enqueue")
+      .mockImplementation(function (this: KeyedAsyncQueue, key, task, hooks) {
+        ingestQueued.resolve();
+        return originalEnqueue.call(this, key, task, hooks);
+      });
+    let ingest: ReturnType<typeof ingestMemoryWikiSource> | undefined;
+    try {
+      ingest = ingestMemoryWikiSource({
+        config,
+        inputPath,
+        nowMs: Date.UTC(2026, 3, 5, 12, 0, 0),
+      });
+      // On fixed code this observes ingest joining the held queue before any
+      // filesystem work. On unfixed code it observes nested compile only after
+      // the source page was already written, so the assertion fails.
+      await ingestQueued.promise;
+      await expect(fs.access(pagePath)).rejects.toThrow();
+
+      releaseLock.resolve();
+      // Completion also proves the nested compile re-enters the held vault
+      // lock reentrantly instead of deadlocking.
+      const result = await ingest;
+      await holder;
+      expect(result.created).toBe(true);
+      await expect(fs.readFile(pagePath, "utf8")).resolves.toContain("hello from source");
+    } finally {
+      releaseLock.resolve();
+      enqueueSpy.mockRestore();
+      await Promise.allSettled([holder, ...(ingest ? [ingest] : [])]);
+    }
+  });
 });

extensions/memory-wiki/src/ingest.ts

@@ -12,6 +12,7 @@ import {
   slugifyWikiPageStem,
   slugifyWikiSegment,
 } from "./markdown.js";
+import { withMemoryWikiVaultMutation } from "./mutation-coordinator.js";
 import { resolveMemoryWikiTimestamp } from "./time.js";
 import { initializeMemoryWikiVault } from "./vault.js";
 
@@ -64,7 +65,7 @@ async function readExistingSourcePage(pagePath: string): Promise<string> {
   throw readError;
 }
 
-export async function ingestMemoryWikiSource(params: {
+async function ingestMemoryWikiSourceUnlocked(params: {
   config: ResolvedMemoryWikiConfig;
   inputPath: string;
   title?: string;
@@ -142,3 +143,17 @@ export async function ingestMemoryWikiSource(params: {
     indexUpdatedFiles: compile.updatedFiles,
   };
 }
+
+export async function ingestMemoryWikiSource(params: {
+  config: ResolvedMemoryWikiConfig;
+  inputPath: string;
+  title?: string;
+  nowMs?: number;
+}): Promise<IngestMemoryWikiSourceResult> {
+  // Ingest read-modify-writes the source page and recompiles the vault; hold
+  // the vault mutation lock across the whole span so it cannot interleave
+  // with the other serialized vault mutators (apply/compile/source-sync).
+  return await withMemoryWikiVaultMutation(params.config.vault.path, () =>
+    ingestMemoryWikiSourceUnlocked(params),
+  );
+}