diff --git a/extensions/memory-wiki/src/ingest.test.ts b/extensions/memory-wiki/src/ingest.test.ts index 40a68de7594..5383d384328 100644 --- a/extensions/memory-wiki/src/ingest.test.ts +++ b/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((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 | 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] : [])]); + } + }); }); diff --git a/extensions/memory-wiki/src/ingest.ts b/extensions/memory-wiki/src/ingest.ts index 0813943562f..e00606d611a 100644 --- a/extensions/memory-wiki/src/ingest.ts +++ b/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 { 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 { + // 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), + ); +}