import { randomUUID } from "node:crypto"; import fsSync from "node:fs"; import fs from "node:fs/promises"; import path from "node:path"; import type { DatabaseSync } from "node:sqlite"; import chokidar, { FSWatcher } from "chokidar"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; import { classifyMemoryMultimodalPath } from "openclaw/plugin-sdk/memory-core-host-engine-embeddings"; import { createSubsystemLogger, onSessionTranscriptUpdate, resolveAgentDir, resolveSessionTranscriptsDirForAgent, resolveUserPath, type OpenClawConfig, type ResolvedMemorySearchConfig, } from "openclaw/plugin-sdk/memory-core-host-engine-foundation"; import { buildSessionEntry, isSessionArchiveArtifactName, isUsageCountedSessionTranscriptFileName, listSessionFilesForAgent, sessionPathForFile, } from "openclaw/plugin-sdk/memory-core-host-engine-qmd"; import { buildFileEntry, ensureMemoryIndexSchema, isFileMissingError, listMemoryFiles, loadSqliteVecExtension, normalizeExtraMemoryPaths, runWithConcurrency, type MemorySource, type MemorySyncProgressUpdate, } from "openclaw/plugin-sdk/memory-core-host-engine-storage"; import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coerce-runtime"; import { createEmbeddingProvider, type EmbeddingProvider, type EmbeddingProviderId, type EmbeddingProviderRuntime, } from "./embeddings.js"; import { runMemoryAtomicReindex } from "./manager-atomic-reindex.js"; import { closeMemoryDatabase, openMemoryDatabaseAtPath } from "./manager-db.js"; import { isMemoryEmbeddingOperationError } from "./manager-embedding-errors.js"; import { applyMemoryFallbackProviderState, resolveMemoryFallbackProviderRequest, resolveFallbackCurrentProviderId, type MemoryProviderLifecycleState, } from "./manager-provider-state.js"; import { resolveConfiguredScopeHash, resolveConfiguredSourcesForMeta, shouldRunFullMemoryReindex, type MemoryIndexMeta, } from "./manager-reindex-state.js"; import { shouldSyncSessionsForReindex } from "./manager-session-reindex.js"; import { resolveMemorySessionStartupDirtyFiles, resolveMemorySessionSyncPlan, type MemorySessionStartupFileState, } from "./manager-session-sync-state.js"; import { loadMemorySourceFileState, resolveMemorySourceExistingHash, } from "./manager-source-state.js"; import { runMemoryTargetedSessionSync } from "./manager-targeted-sync.js"; import { recordMemoryWatchEventPath, settleMemoryWatchEventPaths, type MemoryWatchEventStats, type MemoryWatchSettleQueue, } from "./watch-settle.js"; type MemorySyncProgressState = { completed: number; total: number; label?: string; report: (update: MemorySyncProgressUpdate) => void; }; type MemoryIndexEntry = { path: string; absPath: string; mtimeMs: number; size: number; hash: string; content?: string; }; const META_KEY = "memory_index_meta_v1"; const VECTOR_TABLE = "chunks_vec"; const FTS_TABLE = "chunks_fts"; const EMBEDDING_CACHE_TABLE = "embedding_cache"; const SESSION_DIRTY_DEBOUNCE_MS = 5000; const SESSION_DELTA_READ_CHUNK_BYTES = 64 * 1024; const SESSION_SYNC_YIELD_EVERY = 10; const VECTOR_LOAD_TIMEOUT_MS = 30_000; const IGNORED_MEMORY_WATCH_DIR_NAMES = new Set([ ".git", "node_modules", ".pnpm-store", ".venv", "venv", ".tox", "__pycache__", ]); const log = createSubsystemLogger("memory"); const TEST_MEMORY_WATCH_FACTORY_KEY = Symbol.for("openclaw.test.memoryWatchFactory"); const TEST_MEMORY_NATIVE_WATCH_FACTORY_KEY = Symbol.for("openclaw.test.memoryNativeWatchFactory"); type NativeMemoryWatchPair = { dir: string; main: fsSync.FSWatcher; parent: fsSync.FSWatcher | null; }; function resolveMemoryWatchFactory(): typeof chokidar.watch { if (process.env.VITEST === "true" || process.env.NODE_ENV === "test") { const override = (globalThis as Record)[TEST_MEMORY_WATCH_FACTORY_KEY]; if (typeof override === "function") { return override as typeof chokidar.watch; } } return chokidar.watch.bind(chokidar); } function resolveMemoryNativeWatchFactory(): typeof fsSync.watch { if (process.env.VITEST === "true" || process.env.NODE_ENV === "test") { const override = (globalThis as Record)[ TEST_MEMORY_NATIVE_WATCH_FACTORY_KEY ]; if (typeof override === "function") { return override as typeof fsSync.watch; } } return fsSync.watch.bind(fsSync); } function shouldIgnoreMemoryWatchPath( watchPath: string, stats?: { isDirectory?: () => boolean }, multimodalSettings?: ResolvedMemorySearchConfig["multimodal"], ): boolean { const normalized = path.normalize(watchPath); const parts = normalized .split(path.sep) .map((segment) => normalizeLowercaseStringOrEmpty(segment)); if (parts.some((segment) => IGNORED_MEMORY_WATCH_DIR_NAMES.has(segment))) { return true; } if (stats?.isDirectory?.()) { return false; } if (!stats) { return false; } const extension = normalizeLowercaseStringOrEmpty(path.extname(normalized)); if (extension.length === 0 || extension === ".md") { return false; } if (!multimodalSettings) { return true; } return classifyMemoryMultimodalPath(normalized, multimodalSettings) === null; } export function runDetachedMemorySync(sync: () => Promise, reason: "interval" | "watch") { void sync().catch((err) => { log.warn(`memory sync failed (${reason}): ${String(err)}`); }); } function createSessionSyncYield(total: number): () => Promise { let completed = 0; return async () => { completed += 1; if (completed < total && completed % SESSION_SYNC_YIELD_EVERY === 0) { await new Promise((resolve) => { setImmediate(resolve); }); } }; } export abstract class MemoryManagerSyncOps { protected abstract readonly cfg: OpenClawConfig; protected abstract readonly agentId: string; protected abstract readonly workspaceDir: string; protected abstract readonly settings: ResolvedMemorySearchConfig; protected provider: EmbeddingProvider | null = null; protected fallbackFrom?: EmbeddingProviderId; protected abstract providerUnavailableReason?: string; protected abstract providerLifecycle: MemoryProviderLifecycleState; protected providerRuntime?: EmbeddingProviderRuntime; protected abstract batch: { enabled: boolean; wait: boolean; concurrency: number; pollIntervalMs: number; timeoutMs: number; }; protected readonly sources: Set = new Set(); protected providerKey: string | null = null; protected abstract readonly vector: { enabled: boolean; available: boolean | null; semanticAvailable?: boolean; extensionPath?: string; loadError?: string; dims?: number; }; protected readonly fts: { enabled: boolean; available: boolean; loadError?: string; } = { enabled: false, available: false }; protected vectorReady: Promise | null = null; protected watcher: FSWatcher | null = null; private nativeMemoryWatchPairs: NativeMemoryWatchPair[] = []; protected watchTimer: NodeJS.Timeout | null = null; protected sessionWatchTimer: NodeJS.Timeout | null = null; protected sessionUnsubscribe: (() => void) | null = null; protected fallbackReason?: string; protected intervalTimer: NodeJS.Timeout | null = null; protected closed = false; protected dirty = false; protected pendingWatchPaths: MemoryWatchSettleQueue = new Map(); protected sessionsDirty = false; protected sessionsDirtyFiles = new Set(); protected sessionPendingFiles = new Set(); protected sessionDeltas = new Map< string, { lastSize: number; pendingBytes: number; pendingMessages: number } >(); protected vectorDegradedWriteWarningShown = false; private lastMetaSerialized: string | null = null; protected abstract readonly cache: { enabled: boolean; maxEntries?: number }; protected abstract db: DatabaseSync; protected abstract computeProviderKey(): string; protected abstract sync(params?: { reason?: string; force?: boolean; forceSessions?: boolean; sessionFile?: string; progress?: (update: MemorySyncProgressUpdate) => void; }): Promise; protected abstract withTimeout( promise: Promise, timeoutMs: number, message: string, ): Promise; protected abstract getIndexConcurrency(): number; protected abstract pruneEmbeddingCacheIfNeeded(): void; protected abstract resetProviderInitializationForRetry(): void; protected abstract indexFile( entry: MemoryIndexEntry, options: { source: MemorySource; content?: string }, ): Promise; protected resetVectorState(): void { this.vectorReady = null; this.vector.available = null; this.vector.semanticAvailable = undefined; this.vector.loadError = undefined; this.vector.dims = undefined; this.vectorDegradedWriteWarningShown = false; } protected async ensureVectorReady(dimensions?: number): Promise { if (!this.vector.enabled) { return false; } if (!this.vectorReady) { this.vectorReady = this.withTimeout( this.loadVectorExtension(), VECTOR_LOAD_TIMEOUT_MS, `sqlite-vec load timed out after ${Math.round(VECTOR_LOAD_TIMEOUT_MS / 1000)}s`, ); } let ready = false; try { ready = (await this.vectorReady) || false; } catch (err) { const message = formatErrorMessage(err); this.vector.available = false; this.vector.loadError = message; this.vectorReady = null; log.warn(`sqlite-vec unavailable: ${message}`); return false; } if (ready && typeof dimensions === "number" && dimensions > 0) { this.ensureVectorTable(dimensions); } return ready; } private async loadVectorExtension(): Promise { if (this.vector.available !== null) { return this.vector.available; } if (!this.vector.enabled) { this.vector.available = false; return false; } try { const resolvedPath = this.vector.extensionPath?.trim() ? resolveUserPath(this.vector.extensionPath) : undefined; const loaded = await loadSqliteVecExtension({ db: this.db, extensionPath: resolvedPath }); if (!loaded.ok) { throw new Error(loaded.error ?? "unknown sqlite-vec load error"); } this.vector.extensionPath = loaded.extensionPath; this.vector.available = true; return true; } catch (err) { const message = formatErrorMessage(err); this.vector.available = false; this.vector.loadError = message; log.warn(`sqlite-vec unavailable: ${message}`); return false; } } private ensureVectorTable(dimensions: number): void { if (this.vector.dims === dimensions) { return; } if (this.vector.dims && this.vector.dims !== dimensions) { this.dropVectorTable(); } this.db.exec( `CREATE VIRTUAL TABLE IF NOT EXISTS ${VECTOR_TABLE} USING vec0(\n` + ` id TEXT PRIMARY KEY,\n` + ` embedding FLOAT[${dimensions}]\n` + `)`, ); this.vector.dims = dimensions; } private dropVectorTable(): void { try { this.db.exec(`DROP TABLE IF EXISTS ${VECTOR_TABLE}`); } catch (err) { const message = formatErrorMessage(err); log.debug(`Failed to drop ${VECTOR_TABLE}: ${message}`); } } protected buildSourceFilter( alias?: string, sourcesOverride?: MemorySource[], ): { sql: string; params: MemorySource[] } { const sources = sourcesOverride ?? Array.from(this.sources); if (sources.length === 0) { return { sql: "", params: [] }; } const column = alias ? `${alias}.source` : "source"; const placeholders = sources.map(() => "?").join(", "); return { sql: ` AND ${column} IN (${placeholders})`, params: sources }; } protected openDatabase(): DatabaseSync { const dbPath = resolveUserPath(this.settings.store.path); return openMemoryDatabaseAtPath(dbPath, this.settings.store.vector.enabled); } private async seedEmbeddingCache(sourceDb: DatabaseSync): Promise { if (!this.cache.enabled) { return; } let transactionStarted = false; try { const rows = sourceDb .prepare( `SELECT provider, model, provider_key, hash, embedding, dims, updated_at FROM ${EMBEDDING_CACHE_TABLE}`, ) .iterate() as IterableIterator<{ provider: string; model: string; provider_key: string; hash: string; embedding: string; dims: number | null; updated_at: number; }>; // Keep gateway health probes responsive while rebuilding large caches. const SEED_EMBEDDING_YIELD_EVERY = 1000; let rowCount = 0; let insert: ReturnType | null = null; for (const row of rows) { if (!insert) { insert = this.db.prepare( `INSERT INTO ${EMBEDDING_CACHE_TABLE} (provider, model, provider_key, hash, embedding, dims, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(provider, model, provider_key, hash) DO UPDATE SET embedding=excluded.embedding, dims=excluded.dims, updated_at=excluded.updated_at`, ); this.db.exec("BEGIN"); transactionStarted = true; } insert.run( row.provider, row.model, row.provider_key, row.hash, row.embedding, row.dims, row.updated_at, ); rowCount += 1; if (rowCount % SEED_EMBEDDING_YIELD_EVERY === 0) { await new Promise((resolve) => { setImmediate(resolve); }); } } if (transactionStarted) { this.db.exec("COMMIT"); } } catch (err) { if (transactionStarted) { try { this.db.exec("ROLLBACK"); } catch {} } throw err; } } protected ensureSchema() { const result = ensureMemoryIndexSchema({ db: this.db, embeddingCacheTable: EMBEDDING_CACHE_TABLE, cacheEnabled: this.cache.enabled, ftsTable: FTS_TABLE, ftsEnabled: this.fts.enabled, ftsTokenizer: this.settings.store.fts.tokenizer, }); this.fts.available = result.ftsAvailable; if (result.ftsError) { this.fts.loadError = result.ftsError; // Only warn when hybrid search is enabled; otherwise this is expected noise. if (this.fts.enabled) { log.warn(`fts unavailable: ${result.ftsError}`); } } } protected ensureWatcher() { if (!this.sources.has("memory") || !this.settings.sync.watch) { return; } if (this.watcher || this.nativeMemoryWatchPairs.length > 0) { // Already initialized — preserve idempotence. return; } // Core paths preserve original symlink-follow behavior (chokidar/fs.watch // resolve through symlinks by default); extraPaths preserves the original // explicit symlink-skip policy. const fileWatchPaths = new Set([path.join(this.workspaceDir, "MEMORY.md")]); const dirWatchPaths = new Set([path.join(this.workspaceDir, "memory")]); const additionalPaths = normalizeExtraMemoryPaths(this.workspaceDir, this.settings.extraPaths); for (const entry of additionalPaths) { try { const stat = fsSync.lstatSync(entry); if (stat.isSymbolicLink()) { continue; } if (stat.isDirectory()) { dirWatchPaths.add(entry); continue; } if ( stat.isFile() && (normalizeLowercaseStringOrEmpty(entry).endsWith(".md") || classifyMemoryMultimodalPath(entry, this.settings.multimodal) !== null) ) { fileWatchPaths.add(entry); } } catch { // Skip missing/unreadable additional paths. } } const markDirty = (watchPath?: string, stats?: MemoryWatchEventStats) => { recordMemoryWatchEventPath(this.pendingWatchPaths, watchPath, stats); this.dirty = true; this.scheduleWatchSync(); }; // Native recursive fs.watch for directory paths — one watcher per // directory on macOS (FSEvents) and Windows (ReadDirectoryChangesW). // Avoids chokidar's per-file fs.watch fan-out that opened ~12k REG FDs // on multi-thousand-`.md` memory trees (issue #86613). // // Linux is intentionally NOT in the native set: Node's // `fs.watch(dir, { recursive: true })` on non-macOS/non-Windows routes // through `internal/fs/recursive_watch`, which walks the tree and // attaches one watcher per entry under the hood. That defeats the // constant-watcher-profile goal of this fix without throwing (so the // creation-failure fallback below would not catch it). Linux paths // therefore go straight to chokidar, matching pre-PR behavior on that // platform. // // On any other native creation failure (e.g. unsupported filesystem, // ERR_FEATURE_UNAVAILABLE_ON_PLATFORM) the directory also falls back to // chokidar so freshness is preserved on the degraded path. const nativeRecursiveSupported = process.platform === "darwin" || process.platform === "win32"; for (const dir of dirWatchPaths) { if (!nativeRecursiveSupported) { fileWatchPaths.add(dir); continue; } if (!this.attachNativeMemoryWatchForDir(dir, markDirty)) { // Native creation failed (dir missing, unsupported FS, throw) — // fall back to chokidar so directory coverage isn't dropped. fileWatchPaths.add(dir); } } if (fileWatchPaths.size > 0) { this.watcher = resolveMemoryWatchFactory()(Array.from(fileWatchPaths), { ignoreInitial: true, ignored: (watchPath, stats) => shouldIgnoreMemoryWatchPath(watchPath, stats, this.settings.multimodal), }); this.watcher.on("add", markDirty); this.watcher.on("change", markDirty); this.watcher.on("unlink", markDirty); this.watcher.on("unlinkDir", markDirty); this.watcher.on("error", (err) => { // File watcher errors (e.g., ENOSPC) should not crash the gateway. // Log the error and continue - memory search still works without auto-sync. const message = err instanceof Error ? err.message : String(err); log.warn(`memory watcher error: ${message}`); }); } } // Attach a native recursive `fs.watch` to `dir` plus a non-recursive // parent-directory watch that detects root-replacement // (`rm -rf memory && mkdir memory`) by inode comparison. Returns true if // the main native watcher attached. Called from ensureWatcher(); also // re-entered from the parent-watch handler on detected replacement. protected attachNativeMemoryWatchForDir( dir: string, markDirty: (watchPath?: string, stats?: MemoryWatchEventStats) => void, ): boolean { if (this.closed) { return false; } let recordedInode: number | null; try { recordedInode = fsSync.statSync(dir).ino; } catch { // Dir doesn't exist; caller will fall back to chokidar. return false; } let mainWatcher: fsSync.FSWatcher; try { mainWatcher = resolveMemoryNativeWatchFactory()( dir, { recursive: true }, (_eventType, filename) => { if (filename == null) { // Node docs: filename may be null on some platforms even when // recursive watching is otherwise supported. Be conservative // and mark broadly dirty rather than dropping the event. markDirty(); return; } const full = path.join(dir, filename); let stats: fsSync.Stats | undefined; try { const s = fsSync.lstatSync(full, { throwIfNoEntry: false }); stats = s ?? undefined; } catch { stats = undefined; } if (shouldIgnoreMemoryWatchPath(full, stats, this.settings.multimodal)) { return; } // Pass stats so the watch-settle queue can debounce rapid // writes; without a snapshot the queue cannot detect stability. markDirty(full, stats); }, ); } catch (err) { log.warn( `failed to start native recursive watcher on ${dir}: ${String(err)}; falling back to chokidar`, ); return false; } const pair: NativeMemoryWatchPair = { dir, main: mainWatcher, parent: null }; mainWatcher.on("error", (err) => { const message = err instanceof Error ? err.message : String(err); log.warn(`memory native watcher error on ${dir}: ${message}`); // Per Node docs the FSWatcher is no longer usable after an error. this.closeNativeMemoryWatchPair(pair); if (this.closed) { return; } // Force a broad re-sync to cover the gap, then restore directory // coverage by reattaching to chokidar so subsequent file changes // still drive watch sync (intervalMinutes defaults to 0; without // a watcher the directory would stop being indexed). markDirty(); this.attachMemoryChokidarFallback(dir, markDirty); }); this.nativeMemoryWatchPairs.push(pair); // Non-recursive parent watcher: catches root-directory replacement so // we can reattach the main watcher on the new inode. Without this, // `rm -rf memory && mkdir memory` would leave the main watcher bound // to the dead inode and silently miss subsequent file changes. try { const parentDir = path.dirname(dir); const baseName = path.basename(dir); const parentWatcher = resolveMemoryNativeWatchFactory()( parentDir, { recursive: false }, (_eventType, filename) => { // Per Node docs `filename` can be null on some platforms even // when the parent watcher is otherwise supported. Treat null // as an unknown event and re-check the watched directory's // inode (clawsweeper review [P2] 5df68c…); otherwise filter // by basename so sibling events don't trigger reattach. if (filename !== null && filename !== baseName) { return; } let currentInode: number | null; try { currentInode = fsSync.statSync(dir).ino; } catch { currentInode = null; } if (currentInode === recordedInode) { return; } // Root was replaced (or removed). Tear down the existing pair // and either reattach (if dir still exists) or fall back to // chokidar (if dir is gone). this.closeNativeMemoryWatchPair(pair); if (this.closed) { return; } markDirty(); if (currentInode !== null) { // Re-attach on the new inode (this also installs a fresh // parent watcher closed over the new recordedInode). If the // helper's own statSync races with the dir disappearing // between our inode check and its own check, it returns // false — fall back to chokidar so coverage isn't lost. if (!this.attachNativeMemoryWatchForDir(dir, markDirty)) { this.attachMemoryChokidarFallback(dir, markDirty); } } else { this.attachMemoryChokidarFallback(dir, markDirty); } }, ); parentWatcher.on("error", (err) => { const message = err instanceof Error ? err.message : String(err); log.warn(`memory native parent watcher error on ${path.dirname(dir)}: ${message}`); try { parentWatcher.close(); } catch { // ignore } this.removeNativeMemoryParentWatch(parentWatcher); if (pair.parent === parentWatcher) { pair.parent = null; } // Main watcher still alive — root-replacement detection is lost // but normal events still flow. No fallback needed. }); pair.parent = parentWatcher; } catch (err) { // Parent watcher couldn't start (e.g. parentDir not accessible). // The main watcher still works for non-replacement events; just // log and continue. log.warn( `memory native parent watcher could not start on ${path.dirname(dir)}: ${String(err)}`, ); } return true; } private closeNativeMemoryWatchPair(pair: NativeMemoryWatchPair): void { try { pair.main.close(); } catch { // ignore close failures } if (pair.parent) { try { pair.parent.close(); } catch { // ignore close failures } pair.parent = null; } this.removeNativeMemoryWatchPair(pair); } protected closeNativeMemoryWatchPairs(): void { while (this.nativeMemoryWatchPairs.length > 0) { const pair = this.nativeMemoryWatchPairs[0]; if (!pair) { return; } this.closeNativeMemoryWatchPair(pair); } } private removeNativeMemoryParentWatch(w: fsSync.FSWatcher): void { for (const pair of this.nativeMemoryWatchPairs) { if (pair.parent === w) { pair.parent = null; return; } } } private removeNativeMemoryWatchPair(pair: NativeMemoryWatchPair): void { const idx = this.nativeMemoryWatchPairs.indexOf(pair); if (idx >= 0) { this.nativeMemoryWatchPairs.splice(idx, 1); } } // Reattach `dir` to chokidar after a native recursive watcher dies, so // subsequent memory changes under `dir` continue to drive watch sync. // Called from the native watcher `error` handler in ensureWatcher(); // factored out so the fallback shape can be unit-tested in isolation. protected attachMemoryChokidarFallback( dir: string, markDirty: (watchPath?: string, stats?: MemoryWatchEventStats) => void, ): void { if (this.closed) { // Manager teardown started — don't create new watcher resources. return; } try { if (this.watcher) { // Existing chokidar watcher (handling MEMORY.md and/or other file // paths) — extend it to cover this directory too. this.watcher.add(dir); return; } // No chokidar watcher exists yet. Spin one up just for this directory // so the periodic-sync gap is closed. this.watcher = resolveMemoryWatchFactory()([dir], { ignoreInitial: true, ignored: (watchPath, stats) => shouldIgnoreMemoryWatchPath(watchPath, stats, this.settings.multimodal), }); this.watcher.on("add", markDirty); this.watcher.on("change", markDirty); this.watcher.on("unlink", markDirty); this.watcher.on("unlinkDir", markDirty); this.watcher.on("error", (err) => { const message = err instanceof Error ? err.message : String(err); log.warn(`memory watcher error: ${message}`); }); } catch (err) { log.warn(`failed to attach chokidar fallback for ${dir}: ${String(err)}`); } } protected ensureSessionListener() { if (!this.sources.has("sessions") || this.sessionUnsubscribe) { return; } this.sessionUnsubscribe = onSessionTranscriptUpdate((update) => { if (this.closed) { return; } const sessionFile = update.sessionFile; if (!this.isSessionFileForAgent(sessionFile)) { return; } this.scheduleSessionDirty(sessionFile); }); } protected ensureSessionStartupCatchup(): void { if (!this.sources.has("sessions")) { return; } void this.runSessionStartupCatchup().catch((err) => { log.warn("memory session startup catch-up failed: " + String(err)); }); } protected async markSessionStartupCatchupDirtyFiles(): Promise { if (!this.sources.has("sessions") || this.closed) { return []; } const files = await listSessionFilesForAgent(this.agentId); if (files.length === 0 || this.closed) { return []; } const existingRows = loadMemorySourceFileState({ db: this.db, source: "sessions", }).rows; const fileStates = ( await runWithConcurrency( files.map((file) => async (): Promise => { try { const stat = await fs.stat(file); if (!stat.isFile()) { return null; } return { absPath: file, path: sessionPathForFile(file), mtimeMs: stat.mtimeMs, size: stat.size, }; } catch (err) { if (isFileMissingError(err)) { return null; } throw err; } }), this.getIndexConcurrency(), ) ).filter((file): file is MemorySessionStartupFileState => file !== null); const dirtyFiles = resolveMemorySessionStartupDirtyFiles({ files: fileStates, existingRows }); if (dirtyFiles.length === 0 || this.closed) { return dirtyFiles; } for (const file of dirtyFiles) { this.sessionsDirtyFiles.add(file); } this.sessionsDirty = true; return dirtyFiles; } protected async runSessionStartupCatchup(): Promise { const dirtyFiles = await this.markSessionStartupCatchupDirtyFiles(); if (dirtyFiles.length === 0 || this.closed) { return dirtyFiles; } void this.sync({ reason: "session-startup-catchup" }).catch((err) => { log.warn("memory sync failed (session-startup-catchup): " + String(err)); }); return dirtyFiles; } private scheduleSessionDirty(sessionFile: string) { this.sessionPendingFiles.add(sessionFile); if (this.sessionWatchTimer) { return; } this.sessionWatchTimer = setTimeout(() => { this.sessionWatchTimer = null; void this.processSessionDeltaBatch().catch((err) => { log.warn(`memory session delta failed: ${String(err)}`); }); }, SESSION_DIRTY_DEBOUNCE_MS); } private async processSessionDeltaBatch(): Promise { if (this.sessionPendingFiles.size === 0) { return; } const pending = Array.from(this.sessionPendingFiles); this.sessionPendingFiles.clear(); let shouldSync = false; for (const sessionFile of pending) { // Usage-counted session archives (`.jsonl.reset.` and // `.jsonl.deleted.`) are one-shot mutation events: the file is // written once by the archive rotation and then never touched again. // They carry no incremental `append` semantics, so the delta-bytes / // delta-messages thresholds (designed for live transcripts accumulating // appended messages) cannot gate them correctly — a short archive // below the threshold would simply never reindex. Mark them dirty // directly and skip the delta accounting. const baseName = path.basename(sessionFile); if ( isSessionArchiveArtifactName(baseName) && isUsageCountedSessionTranscriptFileName(baseName) ) { this.sessionsDirtyFiles.add(sessionFile); this.sessionsDirty = true; shouldSync = true; continue; } const delta = await this.updateSessionDelta(sessionFile); if (!delta) { continue; } const bytesThreshold = delta.deltaBytes; const messagesThreshold = delta.deltaMessages; const bytesHit = bytesThreshold <= 0 ? delta.pendingBytes > 0 : delta.pendingBytes >= bytesThreshold; const messagesHit = messagesThreshold <= 0 ? delta.pendingMessages > 0 : delta.pendingMessages >= messagesThreshold; if (!bytesHit && !messagesHit) { continue; } this.sessionsDirtyFiles.add(sessionFile); this.sessionsDirty = true; delta.pendingBytes = bytesThreshold > 0 ? Math.max(0, delta.pendingBytes - bytesThreshold) : 0; delta.pendingMessages = messagesThreshold > 0 ? Math.max(0, delta.pendingMessages - messagesThreshold) : 0; shouldSync = true; } if (shouldSync) { void this.sync({ reason: "session-delta" }).catch((err) => { log.warn(`memory sync failed (session-delta): ${String(err)}`); }); } } private async updateSessionDelta(sessionFile: string): Promise<{ deltaBytes: number; deltaMessages: number; pendingBytes: number; pendingMessages: number; } | null> { const thresholds = this.settings.sync.sessions; if (!thresholds) { return null; } let stat: { size: number }; try { stat = await fs.stat(sessionFile); } catch { return null; } const size = stat.size; let state = this.sessionDeltas.get(sessionFile); if (!state) { state = { lastSize: 0, pendingBytes: 0, pendingMessages: 0 }; this.sessionDeltas.set(sessionFile, state); } const deltaBytes = Math.max(0, size - state.lastSize); if (deltaBytes === 0 && size === state.lastSize) { return { deltaBytes: thresholds.deltaBytes, deltaMessages: thresholds.deltaMessages, pendingBytes: state.pendingBytes, pendingMessages: state.pendingMessages, }; } if (size < state.lastSize) { state.lastSize = size; state.pendingBytes += size; const shouldCountMessages = thresholds.deltaMessages > 0 && (thresholds.deltaBytes <= 0 || state.pendingBytes < thresholds.deltaBytes); if (shouldCountMessages) { state.pendingMessages += await this.countNewlines(sessionFile, 0, size); } } else { state.pendingBytes += deltaBytes; const shouldCountMessages = thresholds.deltaMessages > 0 && (thresholds.deltaBytes <= 0 || state.pendingBytes < thresholds.deltaBytes); if (shouldCountMessages) { state.pendingMessages += await this.countNewlines(sessionFile, state.lastSize, size); } state.lastSize = size; } this.sessionDeltas.set(sessionFile, state); return { deltaBytes: thresholds.deltaBytes, deltaMessages: thresholds.deltaMessages, pendingBytes: state.pendingBytes, pendingMessages: state.pendingMessages, }; } private async countNewlines(absPath: string, start: number, end: number): Promise { if (end <= start) { return 0; } let handle; try { handle = await fs.open(absPath, "r"); } catch (err) { if (isFileMissingError(err)) { return 0; } throw err; } try { let offset = start; let count = 0; const buffer = Buffer.alloc(SESSION_DELTA_READ_CHUNK_BYTES); while (offset < end) { const toRead = Math.min(buffer.length, end - offset); const { bytesRead } = await handle.read(buffer, 0, toRead, offset); if (bytesRead <= 0) { break; } for (let i = 0; i < bytesRead; i += 1) { if (buffer[i] === 10) { count += 1; } } offset += bytesRead; } return count; } finally { await handle.close(); } } private resetSessionDelta(absPath: string, size: number): void { const state = this.sessionDeltas.get(absPath); if (!state) { return; } state.lastSize = size; state.pendingBytes = 0; state.pendingMessages = 0; } private isSessionFileForAgent(sessionFile: string): boolean { if (!sessionFile) { return false; } const sessionsDir = resolveSessionTranscriptsDirForAgent(this.agentId); const resolvedFile = path.resolve(sessionFile); const resolvedDir = path.resolve(sessionsDir); return resolvedFile.startsWith(`${resolvedDir}${path.sep}`); } private normalizeTargetSessionFiles(sessionFiles?: string[]): Set | null { if (!sessionFiles || sessionFiles.length === 0) { return null; } const normalized = new Set(); for (const sessionFile of sessionFiles) { const trimmed = sessionFile.trim(); if (!trimmed) { continue; } const resolved = path.resolve(trimmed); if (this.isSessionFileForAgent(resolved)) { normalized.add(resolved); } } return normalized.size > 0 ? normalized : null; } protected ensureIntervalSync() { const minutes = this.settings.sync.intervalMinutes; if (!minutes || minutes <= 0 || this.intervalTimer) { return; } const ms = minutes * 60 * 1000; this.intervalTimer = setInterval(() => { runDetachedMemorySync(() => this.sync({ reason: "interval" }), "interval"); }, ms); } private scheduleWatchSync() { if (!this.sources.has("memory") || !this.settings.sync.watch) { return; } if (this.watchTimer) { clearTimeout(this.watchTimer); } this.watchTimer = setTimeout(() => { this.watchTimer = null; runDetachedMemorySync(async () => { if (this.closed) { return; } if (!(await settleMemoryWatchEventPaths(this.pendingWatchPaths))) { if (!this.closed) { this.scheduleWatchSync(); } return; } if (this.closed) { return; } await this.sync({ reason: "watch" }); }, "watch"); }, this.settings.sync.watchDebounceMs); } private shouldSyncSessions( params?: { reason?: string; force?: boolean; sessionFiles?: string[] }, needsFullReindex = false, ) { return shouldSyncSessionsForReindex({ hasSessionSource: this.sources.has("sessions"), sessionsDirty: this.sessionsDirty, dirtySessionFileCount: this.sessionsDirtyFiles.size, sync: params, needsFullReindex, }); } private async syncMemoryFiles(params: { needsFullReindex: boolean; progress?: MemorySyncProgressState; }) { const deleteFileByPathAndSource = this.db.prepare( `DELETE FROM files WHERE path = ? AND source = ?`, ); const deleteChunksByPathAndSource = this.db.prepare( `DELETE FROM chunks WHERE path = ? AND source = ?`, ); const deleteVectorRowsByPathAndSource = this.vector.enabled && this.vector.available ? this.db.prepare( `DELETE FROM ${VECTOR_TABLE} WHERE id IN (SELECT id FROM chunks WHERE path = ? AND source = ?)`, ) : null; const deleteFtsRowsByPathAndSource = this.fts.enabled && this.fts.available ? this.db.prepare(`DELETE FROM ${FTS_TABLE} WHERE path = ? AND source = ?`) : null; const files = await listMemoryFiles( this.workspaceDir, this.settings.extraPaths, this.settings.multimodal, ); const fileEntries = ( await runWithConcurrency( files.map( (file) => async () => await buildFileEntry(file, this.workspaceDir, this.settings.multimodal), ), this.getIndexConcurrency(), ) ).filter((entry): entry is MemoryIndexEntry => entry !== null); log.debug("memory sync: indexing memory files", { files: fileEntries.length, needsFullReindex: params.needsFullReindex, batch: this.batch.enabled, concurrency: this.getIndexConcurrency(), }); const existingState = loadMemorySourceFileState({ db: this.db, source: "memory", }); const existingRows = existingState.rows; const existingHashes = existingState.hashes; const activePaths = new Set(fileEntries.map((entry) => entry.path)); if (params.progress) { params.progress.total += fileEntries.length; params.progress.report({ completed: params.progress.completed, total: params.progress.total, label: this.batch.enabled ? "Indexing memory files (batch)..." : "Indexing memory files…", }); } const tasks = fileEntries.map((entry) => async () => { if (!params.needsFullReindex && existingHashes.get(entry.path) === entry.hash) { if (params.progress) { params.progress.completed += 1; params.progress.report({ completed: params.progress.completed, total: params.progress.total, }); } return; } await this.indexFile(entry, { source: "memory" }); if (params.progress) { params.progress.completed += 1; params.progress.report({ completed: params.progress.completed, total: params.progress.total, }); } }); await runWithConcurrency(tasks, this.getIndexConcurrency()); for (const stale of existingRows) { if (activePaths.has(stale.path)) { continue; } deleteFileByPathAndSource.run(stale.path, "memory"); if (deleteVectorRowsByPathAndSource) { try { deleteVectorRowsByPathAndSource.run(stale.path, "memory"); } catch {} } deleteChunksByPathAndSource.run(stale.path, "memory"); if (deleteFtsRowsByPathAndSource) { try { deleteFtsRowsByPathAndSource.run(stale.path, "memory"); } catch {} } } } private async syncSessionFiles(params: { needsFullReindex: boolean; targetSessionFiles?: string[]; progress?: MemorySyncProgressState; }) { const deleteFileByPathAndSource = this.db.prepare( `DELETE FROM files WHERE path = ? AND source = ?`, ); const deleteChunksByPathAndSource = this.db.prepare( `DELETE FROM chunks WHERE path = ? AND source = ?`, ); const deleteVectorRowsByPathAndSource = this.vector.enabled && this.vector.available ? this.db.prepare( `DELETE FROM ${VECTOR_TABLE} WHERE id IN (SELECT id FROM chunks WHERE path = ? AND source = ?)`, ) : null; const deleteFtsRowsByPathSourceAndModel = this.fts.enabled && this.fts.available ? this.db.prepare(`DELETE FROM ${FTS_TABLE} WHERE path = ? AND source = ? AND model = ?`) : null; const targetSessionFiles = params.needsFullReindex ? null : this.normalizeTargetSessionFiles(params.targetSessionFiles); const files = targetSessionFiles ? Array.from(targetSessionFiles) : await listSessionFilesForAgent(this.agentId); const sessionPlan = resolveMemorySessionSyncPlan({ needsFullReindex: params.needsFullReindex, files, targetSessionFiles, sessionsDirtyFiles: this.sessionsDirtyFiles, existingRows: targetSessionFiles ? null : loadMemorySourceFileState({ db: this.db, source: "sessions", }).rows, sessionPathForFile, }); const { activePaths, existingRows, existingHashes, indexAll } = sessionPlan; log.debug("memory sync: indexing session files", { files: files.length, indexAll, dirtyFiles: this.sessionsDirtyFiles.size, targetedFiles: targetSessionFiles?.size ?? 0, batch: this.batch.enabled, concurrency: this.getIndexConcurrency(), }); if (params.progress) { params.progress.total += files.length; params.progress.report({ completed: params.progress.completed, total: params.progress.total, label: this.batch.enabled ? "Indexing session files (batch)..." : "Indexing session files…", }); } const yieldAfterSessionFile = createSessionSyncYield(files.length); const tasks = files.map((absPath) => async () => { try { if (!indexAll && !this.sessionsDirtyFiles.has(absPath)) { if (params.progress) { params.progress.completed += 1; params.progress.report({ completed: params.progress.completed, total: params.progress.total, }); } return; } const entry = await buildSessionEntry(absPath); if (!entry) { if (params.progress) { params.progress.completed += 1; params.progress.report({ completed: params.progress.completed, total: params.progress.total, }); } return; } const existingHash = resolveMemorySourceExistingHash({ db: this.db, source: "sessions", path: entry.path, existingHashes, }); if (!params.needsFullReindex && existingHash === entry.hash) { if (params.progress) { params.progress.completed += 1; params.progress.report({ completed: params.progress.completed, total: params.progress.total, }); } this.resetSessionDelta(absPath, entry.size); return; } await this.indexFile(entry, { source: "sessions", content: entry.content }); this.resetSessionDelta(absPath, entry.size); if (params.progress) { params.progress.completed += 1; params.progress.report({ completed: params.progress.completed, total: params.progress.total, }); } } finally { await yieldAfterSessionFile(); } }); await runWithConcurrency(tasks, this.getIndexConcurrency()); if (activePaths === null) { // Targeted syncs only refresh the requested transcripts and should not // prune unrelated session rows without a full directory enumeration. return; } const staleRows = existingRows ?? []; const yieldAfterStaleSessionRow = createSessionSyncYield(staleRows.length); for (const stale of staleRows) { try { if (activePaths.has(stale.path)) { continue; } deleteFileByPathAndSource.run(stale.path, "sessions"); if (deleteVectorRowsByPathAndSource) { try { deleteVectorRowsByPathAndSource.run(stale.path, "sessions"); } catch {} } deleteChunksByPathAndSource.run(stale.path, "sessions"); if (deleteFtsRowsByPathSourceAndModel) { try { deleteFtsRowsByPathSourceAndModel.run( stale.path, "sessions", this.provider?.model ?? "fts-only", ); } catch {} } } finally { await yieldAfterStaleSessionRow(); } } } private createSyncProgress( onProgress: (update: MemorySyncProgressUpdate) => void, ): MemorySyncProgressState { const state: MemorySyncProgressState = { completed: 0, total: 0, label: undefined, report: (update) => { if (update.label) { state.label = update.label; } const label = update.total > 0 && state.label ? `${state.label} ${update.completed}/${update.total}` : state.label; onProgress({ completed: update.completed, total: update.total, label, }); }, }; return state; } private assertFtsOnlySyncAllowed(): void { if (this.provider) { return; } const existingMeta = this.readMeta(); if ( !existingMeta || existingMeta.model === "fts-only" || !this.settings.provider || this.settings.provider === "none" ) { return; } this.resetProviderInitializationForRetry(); throw new Error( `Memory sync aborted: embedding provider "${this.settings.provider}" is configured but unavailable. ` + `Refusing to run sync in fts-only fallback mode to protect existing vector index (current model: ${existingMeta.model}).`, ); } protected async runSync(params?: { reason?: string; force?: boolean; sessionFiles?: string[]; progress?: (update: MemorySyncProgressUpdate) => void; }) { // Guard: if an embedding provider is configured but currently unavailable, // abort sync to prevent silently degrading an existing semantic vector index // to fts-only and wiping existing semantic vectors. // This only protects existing semantic indexes; fresh or already-fts-only // indexes can safely sync without an embedding provider. this.assertFtsOnlySyncAllowed(); const progress = params?.progress ? this.createSyncProgress(params.progress) : undefined; if (progress) { progress.report({ completed: progress.completed, total: progress.total, label: "Loading vector extension…", }); } const vectorReady = await this.ensureVectorReady(); const meta = this.readMeta(); const configuredSources = resolveConfiguredSourcesForMeta(this.sources); const configuredScopeHash = resolveConfiguredScopeHash({ workspaceDir: this.workspaceDir, extraPaths: this.settings.extraPaths, multimodal: { enabled: this.settings.multimodal.enabled, modalities: this.settings.multimodal.modalities, maxFileBytes: this.settings.multimodal.maxFileBytes, }, }); const targetSessionFiles = this.normalizeTargetSessionFiles(params?.sessionFiles); const hasTargetSessionFiles = targetSessionFiles !== null; if (params?.reason === "cli" && !params.force && !hasTargetSessionFiles) { await this.markSessionStartupCatchupDirtyFiles(); } const targetedSessionSync = await runMemoryTargetedSessionSync({ hasSessionSource: this.sources.has("sessions"), targetSessionFiles, reason: params?.reason, progress: progress ?? undefined, useUnsafeReindex: process.env.OPENCLAW_TEST_FAST === "1" && process.env.OPENCLAW_TEST_MEMORY_UNSAFE_REINDEX === "1", sessionsDirtyFiles: this.sessionsDirtyFiles, syncSessionFiles: async (targetedParams) => { await this.syncSessionFiles(targetedParams); }, shouldFallbackOnError: (err) => this.shouldFallbackOnError(err), activateFallbackProvider: async (reason) => await this.activateFallbackProvider(reason), runSafeReindex: async (reindexParams) => { await this.runSafeReindex(reindexParams); }, runUnsafeReindex: async (reindexParams) => { await this.runUnsafeReindex(reindexParams); }, }); if (targetedSessionSync.handled) { this.sessionsDirty = targetedSessionSync.sessionsDirty; return; } const needsFullReindex = (params?.force && !hasTargetSessionFiles) || shouldRunFullMemoryReindex({ meta, // Also detects provider→FTS-only transitions so orphaned old-model FTS rows are cleaned up. provider: this.provider ? { id: this.provider.id, model: this.provider.model } : null, providerKey: this.providerKey ?? undefined, configuredSources, configuredScopeHash, chunkTokens: this.settings.chunking.tokens, chunkOverlap: this.settings.chunking.overlap, vectorReady, ftsTokenizer: this.settings.store.fts.tokenizer, }); try { if (needsFullReindex) { if ( process.env.OPENCLAW_TEST_FAST === "1" && process.env.OPENCLAW_TEST_MEMORY_UNSAFE_REINDEX === "1" ) { await this.runUnsafeReindex({ reason: params?.reason, force: params?.force, progress: progress ?? undefined, }); } else { await this.runSafeReindex({ reason: params?.reason, force: params?.force, progress: progress ?? undefined, }); } return; } const shouldSyncMemory = this.sources.has("memory") && ((!hasTargetSessionFiles && params?.force) || needsFullReindex || this.dirty); const shouldSyncSessions = this.shouldSyncSessions(params, needsFullReindex); if (shouldSyncMemory) { await this.syncMemoryFiles({ needsFullReindex, progress: progress ?? undefined }); this.dirty = false; } if (shouldSyncSessions) { await this.syncSessionFiles({ needsFullReindex, targetSessionFiles: targetSessionFiles ? Array.from(targetSessionFiles) : undefined, progress: progress ?? undefined, }); this.sessionsDirty = false; this.sessionsDirtyFiles.clear(); } else if (this.sessionsDirtyFiles.size > 0) { this.sessionsDirty = true; } else { this.sessionsDirty = false; } } catch (err) { const reason = formatErrorMessage(err); const activated = this.shouldFallbackOnError(err) && (await this.activateFallbackProvider(reason)); if (activated) { await this.runSafeReindex({ reason: params?.reason ?? "fallback", force: true, progress: progress ?? undefined, }); return; } if (!this.provider && this.fts.enabled && this.shouldFallbackOnError(err)) { log.warn(`memory embeddings unavailable; rebuilding lexical memory index only: ${reason}`); await this.runSafeReindex({ reason: params?.reason ?? "embedding-degraded", force: true, progress: progress ?? undefined, }); return; } throw err; } } protected shouldFallbackOnError(err: unknown): boolean { return isMemoryEmbeddingOperationError(err); } protected resolveBatchConfig(): { enabled: boolean; wait: boolean; concurrency: number; pollIntervalMs: number; timeoutMs: number; } { const batch = this.settings.remote?.batch; const enabled = Boolean(batch?.enabled && this.provider && this.providerRuntime?.batchEmbed); return { enabled, wait: batch?.wait ?? true, concurrency: Math.max(1, batch?.concurrency ?? 2), pollIntervalMs: batch?.pollIntervalMs ?? 2000, timeoutMs: (batch?.timeoutMinutes ?? 60) * 60 * 1000, }; } protected async activateFallbackProvider(reason: string): Promise { const currentProviderId = resolveFallbackCurrentProviderId({ provider: this.provider, lifecycle: this.providerLifecycle, }); const fallbackRequest = resolveMemoryFallbackProviderRequest({ cfg: this.cfg, settings: this.settings, currentProviderId, }); if (!fallbackRequest || !currentProviderId) { return false; } if (this.fallbackFrom) { return false; } const fallbackResult = await createEmbeddingProvider({ config: this.cfg, agentDir: resolveAgentDir(this.cfg, this.agentId), ...fallbackRequest, }); const fallbackState = applyMemoryFallbackProviderState({ current: { provider: this.provider, fallbackFrom: this.fallbackFrom, fallbackReason: this.fallbackReason, providerUnavailableReason: undefined, providerRuntime: this.providerRuntime, lifecycle: this.providerLifecycle, }, fallbackFrom: currentProviderId, reason, result: fallbackResult, }); this.fallbackFrom = fallbackState.fallbackFrom; this.fallbackReason = fallbackState.fallbackReason; this.provider = fallbackState.provider; this.providerRuntime = fallbackState.providerRuntime; this.providerUnavailableReason = fallbackState.providerUnavailableReason; this.providerLifecycle = fallbackState.lifecycle; this.providerKey = this.computeProviderKey(); this.batch = this.resolveBatchConfig(); log.warn(`memory embeddings: switched to fallback provider (${fallbackRequest.provider})`, { reason, }); return true; } protected async runSafeReindex(params: { reason?: string; force?: boolean; progress?: MemorySyncProgressState; }): Promise { this.assertFtsOnlySyncAllowed(); const dbPath = resolveUserPath(this.settings.store.path); const tempDbPath = `${dbPath}.tmp-${randomUUID()}`; const tempDb = openMemoryDatabaseAtPath(tempDbPath, this.settings.store.vector.enabled); const originalDb = this.db; let tempDbClosed = false; let originalDbClosed = false; const originalState = { ftsAvailable: this.fts.available, ftsError: this.fts.loadError, vectorAvailable: this.vector.available, vectorLoadError: this.vector.loadError, vectorDims: this.vector.dims, vectorDegradedWriteWarningShown: this.vectorDegradedWriteWarningShown, vectorReady: this.vectorReady, }; const restoreOriginalState = () => { if (originalDbClosed) { this.db = openMemoryDatabaseAtPath(dbPath, this.settings.store.vector.enabled); } else { this.db = originalDb; } this.fts.available = originalState.ftsAvailable; this.fts.loadError = originalState.ftsError; this.vector.available = originalDbClosed ? null : originalState.vectorAvailable; this.vector.loadError = originalState.vectorLoadError; this.vector.dims = originalState.vectorDims; this.vectorDegradedWriteWarningShown = originalState.vectorDegradedWriteWarningShown; this.vectorReady = originalDbClosed ? null : originalState.vectorReady; }; this.db = tempDb; this.resetVectorState(); this.fts.available = false; this.fts.loadError = undefined; this.ensureSchema(); let nextMeta: MemoryIndexMeta | null = null; try { nextMeta = await runMemoryAtomicReindex({ targetPath: dbPath, tempPath: tempDbPath, beforeTempCleanup: () => { if (!tempDbClosed) { closeMemoryDatabase(tempDb); tempDbClosed = true; } }, build: async () => { await this.seedEmbeddingCache(originalDb); const shouldSyncMemory = this.sources.has("memory"); const shouldSyncSessions = this.shouldSyncSessions( { reason: params.reason, force: params.force }, true, ); if (shouldSyncMemory) { await this.syncMemoryFiles({ needsFullReindex: true, progress: params.progress }); this.dirty = false; } if (shouldSyncSessions) { await this.syncSessionFiles({ needsFullReindex: true, progress: params.progress }); this.sessionsDirty = false; this.sessionsDirtyFiles.clear(); } else if (this.sessionsDirtyFiles.size > 0) { this.sessionsDirty = true; } else { this.sessionsDirty = false; } const meta: MemoryIndexMeta = { model: this.provider?.model ?? "fts-only", provider: this.provider?.id ?? "none", providerKey: this.providerKey!, sources: resolveConfiguredSourcesForMeta(this.sources), scopeHash: resolveConfiguredScopeHash({ workspaceDir: this.workspaceDir, extraPaths: this.settings.extraPaths, multimodal: { enabled: this.settings.multimodal.enabled, modalities: this.settings.multimodal.modalities, maxFileBytes: this.settings.multimodal.maxFileBytes, }, }), chunkTokens: this.settings.chunking.tokens, chunkOverlap: this.settings.chunking.overlap, ftsTokenizer: this.settings.store.fts.tokenizer, }; if (this.vector.available && this.vector.dims) { meta.vectorDims = this.vector.dims; } this.writeMeta(meta); this.pruneEmbeddingCacheIfNeeded?.(); closeMemoryDatabase(tempDb); tempDbClosed = true; closeMemoryDatabase(originalDb); originalDbClosed = true; return meta; }, }); this.db = openMemoryDatabaseAtPath(dbPath, this.settings.store.vector.enabled); this.resetVectorState(); this.ensureSchema(); this.vector.dims = nextMeta?.vectorDims; } catch (err) { try { if (!tempDbClosed && this.db === tempDb) { closeMemoryDatabase(tempDb); tempDbClosed = true; } } catch {} restoreOriginalState(); throw err; } } private async runUnsafeReindex(params: { reason?: string; force?: boolean; progress?: MemorySyncProgressState; }): Promise { // Perf: for test runs, skip atomic temp-db swapping. The index is isolated // under the per-test HOME anyway, and this cuts substantial fs+sqlite churn. this.resetIndex(); const shouldSyncMemory = this.sources.has("memory"); const shouldSyncSessions = this.shouldSyncSessions( { reason: params.reason, force: params.force }, true, ); if (shouldSyncMemory) { await this.syncMemoryFiles({ needsFullReindex: true, progress: params.progress }); this.dirty = false; } if (shouldSyncSessions) { await this.syncSessionFiles({ needsFullReindex: true, progress: params.progress }); this.sessionsDirty = false; this.sessionsDirtyFiles.clear(); } else if (this.sessionsDirtyFiles.size > 0) { this.sessionsDirty = true; } else { this.sessionsDirty = false; } const nextMeta: MemoryIndexMeta = { model: this.provider?.model ?? "fts-only", provider: this.provider?.id ?? "none", providerKey: this.providerKey!, sources: resolveConfiguredSourcesForMeta(this.sources), scopeHash: resolveConfiguredScopeHash({ workspaceDir: this.workspaceDir, extraPaths: this.settings.extraPaths, multimodal: { enabled: this.settings.multimodal.enabled, modalities: this.settings.multimodal.modalities, maxFileBytes: this.settings.multimodal.maxFileBytes, }, }), chunkTokens: this.settings.chunking.tokens, chunkOverlap: this.settings.chunking.overlap, ftsTokenizer: this.settings.store.fts.tokenizer, }; if (this.vector.available && this.vector.dims) { nextMeta.vectorDims = this.vector.dims; } this.writeMeta(nextMeta); this.pruneEmbeddingCacheIfNeeded?.(); } private resetIndex() { this.db.exec(`DELETE FROM files`); this.db.exec(`DELETE FROM chunks`); if (this.fts.enabled && this.fts.available) { try { this.db.exec(`DROP TABLE IF EXISTS ${FTS_TABLE}`); } catch {} } this.ensureSchema(); this.dropVectorTable(); this.vector.dims = undefined; this.sessionsDirtyFiles.clear(); } protected readMeta(): MemoryIndexMeta | null { const row = this.db.prepare(`SELECT value FROM meta WHERE key = ?`).get(META_KEY) as | { value: string } | undefined; if (!row?.value) { this.lastMetaSerialized = null; return null; } try { const parsed = JSON.parse(row.value) as MemoryIndexMeta; this.lastMetaSerialized = row.value; return parsed; } catch { this.lastMetaSerialized = null; return null; } } protected writeMeta(meta: MemoryIndexMeta) { const value = JSON.stringify(meta); if (this.lastMetaSerialized === value) { return; } this.db .prepare( `INSERT INTO meta (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, ) .run(META_KEY, value); this.lastMetaSerialized = value; } }