diff --git a/extensions/memory-wiki/doctor-contract-api.test.ts b/extensions/memory-wiki/doctor-contract-api.test.ts new file mode 100644 index 000000000000..76eb162ca617 --- /dev/null +++ b/extensions/memory-wiki/doctor-contract-api.test.ts @@ -0,0 +1,190 @@ +// Memory Wiki tests cover doctor migration of legacy source sync state. +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 { + createPluginStateKeyedStoreForTests, + resetPluginStateStoreForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { stateMigrations } from "./doctor-contract-api.js"; +import { + createMemoryWikiSourceSyncStateStore, + readMemoryWikiSourceSyncState, + resolveMemoryWikiSourceSyncStatePath, +} from "./src/source-sync-state.js"; + +const tempDirs: string[] = []; + +async function makeTempDir(): Promise { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "memory-wiki-doctor-")); + tempDirs.push(dir); + return dir; +} + +function migrationParams(params: { stateDir: string; vaultRoot: string }) { + const env = { ...process.env, HOME: params.stateDir, OPENCLAW_STATE_DIR: params.stateDir }; + return { + config: { + plugins: { + entries: { + "memory-wiki": { + config: { + vault: { path: params.vaultRoot }, + }, + }, + }, + }, + }, + env, + stateDir: params.stateDir, + oauthDir: path.join(params.stateDir, "credentials"), + context: { + openPluginStateKeyedStore: (options: OpenKeyedStoreOptions) => + createPluginStateKeyedStoreForTests("memory-wiki", { ...options, env }), + }, + }; +} + +describe("memory-wiki doctor source sync migration", () => { + beforeEach(() => { + resetPluginStateStoreForTests(); + }); + + afterEach(async () => { + resetPluginStateStoreForTests(); + await Promise.all( + tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true })), + ); + }); + + it("detects and migrates legacy source-sync.json into plugin state", async () => { + const stateDir = await makeTempDir(); + const vaultRoot = path.join(stateDir, "vault"); + const legacyPath = resolveMemoryWikiSourceSyncStatePath(vaultRoot); + await fs.mkdir(path.dirname(legacyPath), { recursive: true }); + await fs.writeFile( + legacyPath, + `${JSON.stringify({ + version: 1, + entries: { + alpha: { + group: "bridge", + pagePath: "sources/alpha.md", + sourcePath: "/tmp/alpha.md", + sourceUpdatedAtMs: 100, + sourceSize: 200, + renderFingerprint: "alpha", + }, + }, + })}\n`, + ); + const params = migrationParams({ stateDir, vaultRoot }); + const migration = stateMigrations[0]; + + await expect(migration.detectLegacyState(params)).resolves.toEqual({ + preview: [expect.stringContaining("Memory Wiki source sync:")], + }); + + await expect(migration.migrateLegacyState(params)).resolves.toEqual({ + changes: [ + "Migrated Memory Wiki source sync -> plugin state (1 imported, 0 existing)", + expect.stringContaining("Archived Memory Wiki source-sync legacy source ->"), + ], + warnings: [], + }); + const store = createMemoryWikiSourceSyncStateStore(params.context.openPluginStateKeyedStore); + await expect(readMemoryWikiSourceSyncState(vaultRoot, store)).resolves.toEqual({ + version: 1, + entries: { + alpha: { + group: "bridge", + pagePath: "sources/alpha.md", + sourcePath: "/tmp/alpha.md", + sourceUpdatedAtMs: 100, + sourceSize: 200, + renderFingerprint: "alpha", + }, + }, + }); + await expect(fs.stat(legacyPath)).rejects.toMatchObject({ code: "ENOENT" }); + await expect(fs.stat(`${legacyPath}.migrated`)).resolves.toBeDefined(); + }); + + it("merges legacy entries with existing plugin state before archiving", async () => { + const stateDir = await makeTempDir(); + const vaultRoot = path.join(stateDir, "vault"); + const legacyPath = resolveMemoryWikiSourceSyncStatePath(vaultRoot); + await fs.mkdir(path.dirname(legacyPath), { recursive: true }); + await fs.writeFile( + legacyPath, + `${JSON.stringify({ + version: 1, + entries: { + stale: { + group: "bridge", + pagePath: "sources/stale.md", + sourcePath: "/tmp/stale.md", + sourceUpdatedAtMs: 10, + sourceSize: 20, + renderFingerprint: "stale", + }, + current: { + group: "bridge", + pagePath: "sources/current-old.md", + sourcePath: "/tmp/current-old.md", + sourceUpdatedAtMs: 30, + sourceSize: 40, + renderFingerprint: "old", + }, + }, + })}\n`, + ); + const params = migrationParams({ stateDir, vaultRoot }); + const store = createMemoryWikiSourceSyncStateStore(params.context.openPluginStateKeyedStore); + await store.write(vaultRoot, { + version: 1, + entries: { + current: { + group: "bridge", + pagePath: "sources/current.md", + sourcePath: "/tmp/current.md", + sourceUpdatedAtMs: 50, + sourceSize: 60, + renderFingerprint: "current", + }, + }, + }); + + await expect(stateMigrations[0].migrateLegacyState(params)).resolves.toEqual({ + changes: [ + "Migrated Memory Wiki source sync -> plugin state (1 imported, 1 existing)", + expect.stringContaining("Archived Memory Wiki source-sync legacy source ->"), + ], + warnings: [], + }); + await expect(readMemoryWikiSourceSyncState(vaultRoot, store)).resolves.toEqual({ + version: 1, + entries: { + stale: { + group: "bridge", + pagePath: "sources/stale.md", + sourcePath: "/tmp/stale.md", + sourceUpdatedAtMs: 10, + sourceSize: 20, + renderFingerprint: "stale", + }, + current: { + group: "bridge", + pagePath: "sources/current.md", + sourcePath: "/tmp/current.md", + sourceUpdatedAtMs: 50, + sourceSize: 60, + renderFingerprint: "current", + }, + }, + }); + await expect(fs.stat(legacyPath)).rejects.toMatchObject({ code: "ENOENT" }); + }); +}); diff --git a/extensions/memory-wiki/doctor-contract-api.ts b/extensions/memory-wiki/doctor-contract-api.ts index ca7dd33e08e7..2c9acdd09c9b 100644 --- a/extensions/memory-wiki/doctor-contract-api.ts +++ b/extensions/memory-wiki/doctor-contract-api.ts @@ -1,2 +1,139 @@ -// Memory Wiki API module exposes the plugin public contract. +// Memory Wiki doctor contract migrates shipped source-sync state. +import fs from "node:fs/promises"; +import type { OpenClawConfig } from "openclaw/plugin-sdk/plugin-entry"; +import type { PluginDoctorStateMigration } from "openclaw/plugin-sdk/runtime-doctor"; +import { resolveMemoryWikiConfig, type MemoryWikiPluginConfig } from "./src/config.js"; export { legacyConfigRules, normalizeCompatibilityConfig } from "./src/config-compat.js"; +import { + createMemoryWikiSourceSyncStateStore, + MEMORY_WIKI_SOURCE_SYNC_STATE_MAX_ENTRIES, + MEMORY_WIKI_SOURCE_SYNC_STATE_NAMESPACE, + readLegacyMemoryWikiSourceSyncState, + resolveMemoryWikiSourceSyncStatePath, + writeMemoryWikiSourceSyncState, +} from "./src/source-sync-state.js"; + +function isRecord(value: unknown): value is Record { + return Boolean(value && typeof value === "object" && !Array.isArray(value)); +} + +function resolveHomeDir(env: NodeJS.ProcessEnv): string | undefined { + return env.HOME?.trim() || env.USERPROFILE?.trim() || undefined; +} + +function readConfiguredPluginConfig(config: OpenClawConfig): MemoryWikiPluginConfig | undefined { + const entries = config.plugins?.entries; + const pluginEntry = isRecord(entries) ? entries["memory-wiki"] : undefined; + if (!isRecord(pluginEntry) || !isRecord(pluginEntry.config)) { + return undefined; + } + return pluginEntry.config as MemoryWikiPluginConfig; +} + +function resolveConfiguredVaultRoots(params: { + config: OpenClawConfig; + env: NodeJS.ProcessEnv; +}): string[] { + const homeDir = resolveHomeDir(params.env); + const resolved = resolveMemoryWikiConfig(readConfiguredPluginConfig(params.config), { + homedir: homeDir, + }); + return [resolved.vault.path]; +} + +async function fileExists(filePath: string): Promise { + try { + const stat = await fs.stat(filePath); + return stat.isFile(); + } catch { + return false; + } +} + +async function archiveLegacySource(params: { + filePath: string; + changes: string[]; + warnings: string[]; +}): Promise { + const archivedPath = `${params.filePath}.migrated`; + if (await fileExists(archivedPath)) { + params.warnings.push( + `Left migrated Memory Wiki source-sync source in place because ${archivedPath} already exists`, + ); + return; + } + try { + await fs.rename(params.filePath, archivedPath); + params.changes.push(`Archived Memory Wiki source-sync legacy source -> ${archivedPath}`); + } catch (err) { + params.warnings.push(`Failed archiving Memory Wiki source-sync legacy source: ${String(err)}`); + } +} + +export const stateMigrations: PluginDoctorStateMigration[] = [ + { + id: "memory-wiki-source-sync-json-to-plugin-state", + label: "Memory Wiki source sync state", + async detectLegacyState(params) { + const previews: string[] = []; + for (const vaultRoot of resolveConfiguredVaultRoots({ + config: params.config, + env: params.env, + })) { + const filePath = resolveMemoryWikiSourceSyncStatePath(vaultRoot); + const state = await readLegacyMemoryWikiSourceSyncState(vaultRoot); + const count = Object.keys(state.entries).length; + if (count === 0 || !(await fileExists(filePath))) { + continue; + } + previews.push( + `- Memory Wiki source sync: ${filePath} -> plugin state (${MEMORY_WIKI_SOURCE_SYNC_STATE_NAMESPACE}, ${count} entries)`, + ); + } + return previews.length > 0 ? { preview: previews } : null; + }, + async migrateLegacyState(params) { + const changes: string[] = []; + const warnings: string[] = []; + const store = createMemoryWikiSourceSyncStateStore(params.context.openPluginStateKeyedStore); + for (const vaultRoot of resolveConfiguredVaultRoots({ + config: params.config, + env: params.env, + })) { + const filePath = resolveMemoryWikiSourceSyncStatePath(vaultRoot); + if (!(await fileExists(filePath))) { + continue; + } + const state = await readLegacyMemoryWikiSourceSyncState(vaultRoot); + const count = Object.keys(state.entries).length; + if (count === 0) { + continue; + } + const existingState = await store.read(vaultRoot); + const mergedEntries = { + ...state.entries, + ...existingState.entries, + }; + const mergedCount = Object.keys(mergedEntries).length; + if (mergedCount > MEMORY_WIKI_SOURCE_SYNC_STATE_MAX_ENTRIES) { + warnings.push( + `Skipped Memory Wiki source-sync import for ${vaultRoot}: ${mergedCount} entries exceeds ${MEMORY_WIKI_SOURCE_SYNC_STATE_MAX_ENTRIES}`, + ); + continue; + } + await writeMemoryWikiSourceSyncState( + vaultRoot, + { version: 1, entries: mergedEntries }, + store, + ); + const existingCount = Object.keys(existingState.entries).length; + const importedCount = mergedCount - existingCount; + changes.push( + `Migrated Memory Wiki source sync -> plugin state (${importedCount} imported, ${existingCount} existing)`, + ); + await archiveLegacySource({ filePath, changes, warnings }); + } + return { changes, warnings }; + }, + }, +]; diff --git a/extensions/memory-wiki/index.ts b/extensions/memory-wiki/index.ts index ac01329fe38a..99dbbccb2de4 100644 --- a/extensions/memory-wiki/index.ts +++ b/extensions/memory-wiki/index.ts @@ -5,6 +5,10 @@ import { memoryWikiConfigSchema, resolveMemoryWikiConfig } from "./src/config.js import { createWikiCorpusSupplement } from "./src/corpus-supplement.js"; import { registerMemoryWikiGatewayMethods } from "./src/gateway.js"; import { createWikiPromptSectionBuilder } from "./src/prompt-section.js"; +import { + configureMemoryWikiSourceSyncStateStore, + createMemoryWikiSourceSyncStateStore, +} from "./src/source-sync-state.js"; import { createWikiApplyTool, createWikiGetTool, @@ -20,6 +24,9 @@ export default definePluginEntry({ configSchema: memoryWikiConfigSchema, register(api) { const config = resolveMemoryWikiConfig(api.pluginConfig); + configureMemoryWikiSourceSyncStateStore( + createMemoryWikiSourceSyncStateStore(api.runtime.state.openKeyedStore), + ); api.registerMemoryPromptSupplement(createWikiPromptSectionBuilder(config)); api.registerMemoryCorpusSupplement( diff --git a/extensions/memory-wiki/src/bridge.ts b/extensions/memory-wiki/src/bridge.ts index 8ed5de3e0c60..96e1488582e0 100644 --- a/extensions/memory-wiki/src/bridge.ts +++ b/extensions/memory-wiki/src/bridge.ts @@ -19,6 +19,7 @@ import { import { writeImportedSourcePage } from "./source-page-shared.js"; import { resolveArtifactKey } from "./source-path-shared.js"; import { + assertMemoryWikiSourceSyncStateCapacity, pruneImportedSourceEntries, readMemoryWikiSourceSyncState, writeMemoryWikiSourceSyncState, @@ -224,10 +225,15 @@ export async function syncMemoryWikiBridgeSources(params: { } const publicArtifacts = await listActiveMemoryPublicArtifacts({ cfg: params.appConfig }); - const state = await readMemoryWikiSourceSyncState(params.config.vault.path); const results: Array<{ pagePath: string; changed: boolean; created: boolean }> = []; const activeKeys = new Set(); const artifacts = await collectBridgeArtifacts(params.config.bridge, publicArtifacts); + const state = await readMemoryWikiSourceSyncState(params.config.vault.path); + assertMemoryWikiSourceSyncStateCapacity({ + state, + group: "bridge", + incomingCount: artifacts.length, + }); const agentIdsByWorkspace = new Map(); for (const artifact of publicArtifacts) { agentIdsByWorkspace.set(artifact.workspaceDir, artifact.agentIds); diff --git a/extensions/memory-wiki/src/source-sync-state.test.ts b/extensions/memory-wiki/src/source-sync-state.test.ts new file mode 100644 index 000000000000..357c61b10bd0 --- /dev/null +++ b/extensions/memory-wiki/src/source-sync-state.test.ts @@ -0,0 +1,175 @@ +// Memory Wiki tests cover source sync state plugin behavior. +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 { + createPluginStateKeyedStoreForTests, + resetPluginStateStoreForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { + assertMemoryWikiSourceSyncStateCapacity, + configureMemoryWikiSourceSyncStateStore, + createMemoryWikiSourceSyncStateStore, + MEMORY_WIKI_SOURCE_SYNC_STATE_MAX_ENTRIES, + readLegacyMemoryWikiSourceSyncState, + readMemoryWikiSourceSyncState, + resolveMemoryWikiSourceSyncStatePath, + writeMemoryWikiSourceSyncState, +} from "./source-sync-state.js"; + +const tempDirs: string[] = []; + +async function makeTempDir(): Promise { + const dir = await fs.mkdtemp(path.join(os.tmpdir(), "memory-wiki-source-sync-")); + tempDirs.push(dir); + return dir; +} + +function openStore(env: NodeJS.ProcessEnv) { + return createMemoryWikiSourceSyncStateStore((options: OpenKeyedStoreOptions) => + createPluginStateKeyedStoreForTests("memory-wiki", { ...options, env }), + ); +} + +describe("memory wiki source sync state", () => { + beforeEach(() => { + resetPluginStateStoreForTests(); + configureMemoryWikiSourceSyncStateStore(undefined); + }); + + afterEach(async () => { + configureMemoryWikiSourceSyncStateStore(undefined); + resetPluginStateStoreForTests(); + await Promise.all( + tempDirs.splice(0).map((dir) => fs.rm(dir, { recursive: true, force: true })), + ); + }); + + it("persists source sync entries in plugin state", async () => { + const stateDir = await makeTempDir(); + const vaultRoot = path.join(stateDir, "vault"); + const store = openStore({ ...process.env, OPENCLAW_STATE_DIR: stateDir }); + + await writeMemoryWikiSourceSyncState( + vaultRoot, + { + version: 1, + entries: { + alpha: { + group: "bridge", + pagePath: "sources/alpha.md", + sourcePath: "/tmp/source.md", + sourceUpdatedAtMs: 123, + sourceSize: 456, + renderFingerprint: "fingerprint", + }, + }, + }, + store, + ); + + await expect(readMemoryWikiSourceSyncState(vaultRoot, store)).resolves.toEqual({ + version: 1, + entries: { + alpha: { + group: "bridge", + pagePath: "sources/alpha.md", + sourcePath: "/tmp/source.md", + sourceUpdatedAtMs: 123, + sourceSize: 456, + renderFingerprint: "fingerprint", + }, + }, + }); + await expect(fs.stat(resolveMemoryWikiSourceSyncStatePath(vaultRoot))).rejects.toMatchObject({ + code: "ENOENT", + }); + }); + + it("keeps legacy file reads separate for doctor migration", async () => { + const vaultRoot = await makeTempDir(); + const legacyPath = resolveMemoryWikiSourceSyncStatePath(vaultRoot); + await fs.mkdir(path.dirname(legacyPath), { recursive: true }); + await fs.writeFile( + legacyPath, + `${JSON.stringify({ + version: 1, + entries: { + beta: { + group: "unsafe-local", + pagePath: "sources/beta.md", + sourcePath: "/tmp/beta.md", + sourceUpdatedAtMs: 10, + sourceSize: 20, + renderFingerprint: "beta", + }, + }, + })}\n`, + ); + + await expect(readMemoryWikiSourceSyncState(vaultRoot)).resolves.toEqual({ + version: 1, + entries: {}, + }); + await expect(readLegacyMemoryWikiSourceSyncState(vaultRoot)).resolves.toEqual({ + version: 1, + entries: { + beta: { + group: "unsafe-local", + pagePath: "sources/beta.md", + sourcePath: "/tmp/beta.md", + sourceUpdatedAtMs: 10, + sourceSize: 20, + renderFingerprint: "beta", + }, + }, + }); + }); + + it("rejects writes beyond the source-sync state row cap", async () => { + const stateDir = await makeTempDir(); + const vaultRoot = path.join(stateDir, "vault"); + const store = openStore({ ...process.env, OPENCLAW_STATE_DIR: stateDir }); + const entries = Object.fromEntries( + Array.from({ length: MEMORY_WIKI_SOURCE_SYNC_STATE_MAX_ENTRIES + 1 }, (_, 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 expect( + writeMemoryWikiSourceSyncState(vaultRoot, { version: 1, entries }, store), + ).rejects.toThrow("Memory Wiki source sync state exceeds SQLite entry limit"); + }); + + it("rejects projected imports that would exceed the source-sync row cap", () => { + expect(() => + assertMemoryWikiSourceSyncStateCapacity({ + state: { + version: 1, + entries: { + retained: { + group: "unsafe-local", + pagePath: "sources/retained.md", + sourcePath: "/tmp/retained.md", + sourceUpdatedAtMs: 1, + sourceSize: 1, + renderFingerprint: "retained", + }, + }, + }, + group: "bridge", + incomingCount: MEMORY_WIKI_SOURCE_SYNC_STATE_MAX_ENTRIES, + }), + ).toThrow("Memory Wiki source sync state exceeds SQLite entry limit"); + }); +}); diff --git a/extensions/memory-wiki/src/source-sync-state.ts b/extensions/memory-wiki/src/source-sync-state.ts index 1dde9e536033..b4d418dfa67f 100644 --- a/extensions/memory-wiki/src/source-sync-state.ts +++ b/extensions/memory-wiki/src/source-sync-state.ts @@ -1,11 +1,16 @@ // Memory Wiki plugin module implements source sync state behavior. +import { createHash } from "node:crypto"; import fs from "node:fs/promises"; import path from "node:path"; -import { readJsonFileWithFallback, writeJsonFileAtomically } from "openclaw/plugin-sdk/json-store"; +import { readJsonFileWithFallback } from "openclaw/plugin-sdk/json-store"; +import type { + OpenKeyedStoreOptions, + PluginStateKeyedStore, +} from "openclaw/plugin-sdk/plugin-state-runtime"; export type MemoryWikiImportedSourceGroup = "bridge" | "unsafe-local"; -type MemoryWikiImportedSourceStateEntry = { +export type MemoryWikiImportedSourceStateEntry = { group: MemoryWikiImportedSourceGroup; pagePath: string; sourcePath: string; @@ -14,40 +19,213 @@ type MemoryWikiImportedSourceStateEntry = { renderFingerprint: string; }; -type MemoryWikiImportedSourceState = { +export type MemoryWikiImportedSourceState = { version: 1; entries: Record; }; +type MemoryWikiSourceSyncStateStore = { + read: (vaultRoot: string) => Promise; + write: (vaultRoot: string, state: MemoryWikiImportedSourceState) => Promise; +}; + +type MemoryWikiSourceSyncStateRecord = MemoryWikiImportedSourceStateEntry & { + vaultRootKey: string; + syncKey: string; +}; + +export const MEMORY_WIKI_SOURCE_SYNC_STATE_NAMESPACE = "source-sync"; +export const MEMORY_WIKI_SOURCE_SYNC_STATE_MAX_ENTRIES = 20_000; + const EMPTY_STATE: MemoryWikiImportedSourceState = { version: 1, entries: {}, }; -function resolveMemoryWikiSourceSyncStatePath(vaultRoot: string): string { +let configuredSourceSyncStore: MemoryWikiSourceSyncStateStore | undefined; +const memorySourceSyncStateByVault = new Map(); + +export function resolveMemoryWikiSourceSyncStatePath(vaultRoot: string): string { return path.join(vaultRoot, ".openclaw-wiki", "source-sync.json"); } +function cloneSourceSyncState(state: MemoryWikiImportedSourceState): MemoryWikiImportedSourceState { + return { + version: 1, + entries: Object.fromEntries( + Object.entries(state.entries).map(([key, value]) => [key, { ...value }]), + ), + }; +} + +function normalizeSourceSyncState(value: unknown): MemoryWikiImportedSourceState { + if (!value || typeof value !== "object" || Array.isArray(value)) { + return EMPTY_STATE; + } + const parsed = value as Partial; + if (parsed.version !== 1 || !parsed.entries || typeof parsed.entries !== "object") { + return EMPTY_STATE; + } + const entries: Record = {}; + for (const [syncKey, entry] of Object.entries(parsed.entries)) { + if ( + !entry || + typeof entry !== "object" || + Array.isArray(entry) || + (entry.group !== "bridge" && entry.group !== "unsafe-local") || + typeof entry.pagePath !== "string" || + typeof entry.sourcePath !== "string" || + typeof entry.sourceUpdatedAtMs !== "number" || + typeof entry.sourceSize !== "number" || + typeof entry.renderFingerprint !== "string" + ) { + continue; + } + entries[syncKey] = { + group: entry.group, + pagePath: entry.pagePath, + sourcePath: entry.sourcePath, + sourceUpdatedAtMs: entry.sourceUpdatedAtMs, + sourceSize: entry.sourceSize, + renderFingerprint: entry.renderFingerprint, + }; + } + return { version: 1, entries }; +} + +function resolveVaultRootKey(vaultRoot: string): string { + return createHash("sha256").update(path.resolve(vaultRoot), "utf8").digest("hex").slice(0, 32); +} + +function resolveStateEntryKey(vaultRootKey: string, syncKey: string): string { + return createHash("sha256").update(`${vaultRootKey}\0${syncKey}`, "utf8").digest("hex"); +} + +function createMemoryFallbackStateStore(): MemoryWikiSourceSyncStateStore { + return { + async read(vaultRoot) { + const vaultRootKey = resolveVaultRootKey(vaultRoot); + return cloneSourceSyncState(memorySourceSyncStateByVault.get(vaultRootKey) ?? EMPTY_STATE); + }, + async write(vaultRoot, state) { + assertSourceSyncStateWithinLimit(state); + const vaultRootKey = resolveVaultRootKey(vaultRoot); + memorySourceSyncStateByVault.set(vaultRootKey, cloneSourceSyncState(state)); + }, + }; +} + +function assertSourceSyncStateWithinLimit(state: MemoryWikiImportedSourceState): void { + const count = Object.keys(state.entries).length; + if (count > MEMORY_WIKI_SOURCE_SYNC_STATE_MAX_ENTRIES) { + throw new Error( + `Memory Wiki source sync state exceeds SQLite entry limit (${count}/${MEMORY_WIKI_SOURCE_SYNC_STATE_MAX_ENTRIES})`, + ); + } +} + +export function assertMemoryWikiSourceSyncStateCapacity(params: { + state: MemoryWikiImportedSourceState; + group: MemoryWikiImportedSourceGroup; + incomingCount: number; +}): void { + const retainedOtherGroupCount = Object.values(params.state.entries).filter( + (entry) => entry.group !== params.group, + ).length; + const projectedCount = retainedOtherGroupCount + params.incomingCount; + if (projectedCount > MEMORY_WIKI_SOURCE_SYNC_STATE_MAX_ENTRIES) { + throw new Error( + `Memory Wiki source sync state exceeds SQLite entry limit (${projectedCount}/${MEMORY_WIKI_SOURCE_SYNC_STATE_MAX_ENTRIES})`, + ); + } +} + +export function createMemoryWikiSourceSyncStateStore( + openKeyedStore: (options: OpenKeyedStoreOptions) => PluginStateKeyedStore, +): MemoryWikiSourceSyncStateStore { + const openStore = () => + openKeyedStore({ + namespace: MEMORY_WIKI_SOURCE_SYNC_STATE_NAMESPACE, + maxEntries: MEMORY_WIKI_SOURCE_SYNC_STATE_MAX_ENTRIES, + }); + + return { + async read(vaultRoot) { + const vaultRootKey = resolveVaultRootKey(vaultRoot); + const entries: MemoryWikiImportedSourceState["entries"] = {}; + for (const row of await openStore().entries()) { + const value = row.value; + if (value.vaultRootKey !== vaultRootKey || typeof value.syncKey !== "string") { + continue; + } + const normalized = normalizeSourceSyncState({ + version: 1, + entries: { [value.syncKey]: value }, + }); + const entry = normalized.entries[value.syncKey]; + if (entry) { + entries[value.syncKey] = entry; + } + } + return { version: 1, entries }; + }, + async write(vaultRoot, state) { + assertSourceSyncStateWithinLimit(state); + const vaultRootKey = resolveVaultRootKey(vaultRoot); + const store = openStore(); + const normalized = normalizeSourceSyncState(state); + const nextKeys = new Set(); + for (const [syncKey, entry] of Object.entries(normalized.entries)) { + const key = resolveStateEntryKey(vaultRootKey, syncKey); + nextKeys.add(key); + await store.register(key, { + ...entry, + vaultRootKey, + syncKey, + }); + } + for (const row of await store.entries()) { + if (row.value.vaultRootKey === vaultRootKey && !nextKeys.has(row.key)) { + await store.delete(row.key); + } + } + }, + }; +} + +export function configureMemoryWikiSourceSyncStateStore( + store: MemoryWikiSourceSyncStateStore | undefined, +): void { + configuredSourceSyncStore = store; +} + +function resolveSourceSyncStore( + store?: MemoryWikiSourceSyncStateStore, +): MemoryWikiSourceSyncStateStore { + return store ?? configuredSourceSyncStore ?? createMemoryFallbackStateStore(); +} + export async function readMemoryWikiSourceSyncState( vaultRoot: string, + store?: MemoryWikiSourceSyncStateStore, +): Promise { + return await resolveSourceSyncStore(store).read(vaultRoot); +} + +export async function readLegacyMemoryWikiSourceSyncState( + vaultRoot: string, ): Promise { const statePath = resolveMemoryWikiSourceSyncStatePath(vaultRoot); - const { value: parsed } = await readJsonFileWithFallback>( - statePath, - EMPTY_STATE, - ); - return { - version: 1, - entries: { ...parsed.entries }, - }; + const { value: parsed } = await readJsonFileWithFallback(statePath, EMPTY_STATE); + return normalizeSourceSyncState(parsed); } export async function writeMemoryWikiSourceSyncState( vaultRoot: string, state: MemoryWikiImportedSourceState, + store?: MemoryWikiSourceSyncStateStore, ): Promise { - const statePath = resolveMemoryWikiSourceSyncStatePath(vaultRoot); - await writeJsonFileAtomically(statePath, state); + await resolveSourceSyncStore(store).write(vaultRoot, state); } export async function shouldSkipImportedSourceWrite(params: { diff --git a/extensions/memory-wiki/src/test-helpers.ts b/extensions/memory-wiki/src/test-helpers.ts index 091a685c3091..e37cb0c97398 100644 --- a/extensions/memory-wiki/src/test-helpers.ts +++ b/extensions/memory-wiki/src/test-helpers.ts @@ -1,6 +1,7 @@ // Memory Wiki helper module supports test helpers behavior. import fs from "node:fs/promises"; import path from "node:path"; +import type { PluginStateEntry } from "openclaw/plugin-sdk/plugin-state-runtime"; import { createTestPluginApi } from "openclaw/plugin-sdk/plugin-test-api"; import { resolvePreferredOpenClawTmpDir } from "openclaw/plugin-sdk/temp-path"; import { afterEach, vi } from "vitest"; @@ -28,6 +29,46 @@ type MemoryWikiPluginApiHarness = { registerTool: ReturnType; }; +function createMemoryKeyedStore() { + const values = new Map(); + return { + async register(key: string, value: T) { + values.set(key, value); + }, + async registerIfAbsent(key: string, value: T) { + if (values.has(key)) { + return false; + } + values.set(key, value); + return true; + }, + async lookup(key: string) { + return values.get(key); + }, + async consume(key: string) { + const value = values.get(key); + values.delete(key); + return value; + }, + async delete(key: string) { + return values.delete(key); + }, + async entries() { + return [...values.entries()].map( + ([key, value]) => + ({ + key, + value, + createdAt: 0, + }) satisfies PluginStateEntry, + ); + }, + async clear() { + values.clear(); + }, + }; +} + export function createMemoryWikiTestHarness() { const tempDirs: string[] = []; @@ -80,7 +121,11 @@ export function createMemoryWikiTestHarness() { name: "Memory Wiki", source: "test", config: {}, - runtime: {} as OpenClawPluginApi["runtime"], + runtime: { + state: { + openKeyedStore: vi.fn(() => createMemoryKeyedStore()), + }, + } as unknown as OpenClawPluginApi["runtime"], registerCli, registerGatewayMethod, registerMemoryCorpusSupplement, diff --git a/extensions/memory-wiki/src/unsafe-local.ts b/extensions/memory-wiki/src/unsafe-local.ts index c37a004c54b9..8e72f5898f54 100644 --- a/extensions/memory-wiki/src/unsafe-local.ts +++ b/extensions/memory-wiki/src/unsafe-local.ts @@ -15,6 +15,7 @@ import { import { writeImportedSourcePage } from "./source-page-shared.js"; import { resolveArtifactKey } from "./source-path-shared.js"; import { + assertMemoryWikiSourceSyncStateCapacity, pruneImportedSourceEntries, readMemoryWikiSourceSyncState, writeMemoryWikiSourceSyncState, @@ -215,6 +216,11 @@ export async function syncMemoryWikiUnsafeLocalSources( const artifacts = await collectUnsafeLocalArtifacts(config.unsafeLocal.paths); const state = await readMemoryWikiSourceSyncState(config.vault.path); + assertMemoryWikiSourceSyncStateCapacity({ + state, + group: "unsafe-local", + incomingCount: artifacts.length, + }); const activeKeys = new Set(); const results = await Promise.all( artifacts.map(async (artifact) => {