github/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-06-07 22:09:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat (memory): Implement new (opt-in) QMD memory backend

Browse Source
This commit is contained in:
Vignesh Natarajan
2026-01-27 21:57:15 -08:00
committed by Vignesh
parent e9f182def7
commit 5d3af3bc62
24 changed files with 1828 additions and 601 deletions
Download Patch File Download Diff File Expand all files Collapse all files

509
src/memory/manager.ts
View File

@@ -1,25 +1,18 @@
import type { DatabaseSync } from "node:sqlite";
import chokidar, { type FSWatcher } from "chokidar";
import { randomUUID } from "node:crypto";
import fsSync from "node:fs";
import fs from "node:fs/promises";
import path from "node:path";
import type { ResolvedMemorySearchConfig } from "../agents/memory-search.js";
import type { OpenClawConfig } from "../config/config.js";
import type { DatabaseSync } from "node:sqlite";
import chokidar, { type FSWatcher } from "chokidar";
import { resolveAgentDir, resolveAgentWorkspaceDir } from "../agents/agent-scope.js";
import type { ResolvedMemorySearchConfig } from "../agents/memory-search.js";
import { resolveMemorySearchConfig } from "../agents/memory-search.js";
import type { MoltbotConfig } from "../config/config.js";
import { resolveSessionTranscriptsDirForAgent } from "../config/sessions/paths.js";
import { createSubsystemLogger } from "../logging/subsystem.js";
import { onSessionTranscriptUpdate } from "../sessions/transcript-events.js";
import { resolveUserPath } from "../utils.js";
import { runGeminiEmbeddingBatches, type GeminiBatchRequest } from "./batch-gemini.js";
import {
OPENAI_BATCH_ENDPOINT,
type OpenAiBatchRequest,
runOpenAiEmbeddingBatches,
} from "./batch-openai.js";
import { DEFAULT_GEMINI_EMBEDDING_MODEL } from "./embeddings-gemini.js";
import { DEFAULT_OPENAI_EMBEDDING_MODEL } from "./embeddings-openai.js";
import {
createEmbeddingProvider,
type EmbeddingProvider,
@@ -27,7 +20,14 @@ import {
type GeminiEmbeddingClient,
type OpenAiEmbeddingClient,
} from "./embeddings.js";
import { bm25RankToScore, buildFtsQuery, mergeHybridResults } from "./hybrid.js";
import { DEFAULT_GEMINI_EMBEDDING_MODEL } from "./embeddings-gemini.js";
import { DEFAULT_OPENAI_EMBEDDING_MODEL } from "./embeddings-openai.js";
import {
OPENAI_BATCH_ENDPOINT,
type OpenAiBatchRequest,
runOpenAiEmbeddingBatches,
} from "./batch-openai.js";
import { runGeminiEmbeddingBatches, type GeminiBatchRequest } from "./batch-gemini.js";
import {
buildFileEntry,
chunkMarkdown,
@@ -35,26 +35,23 @@ import {
hashText,
isMemoryPath,
listMemoryFiles,
normalizeExtraMemoryPaths,
type MemoryChunk,
type MemoryFileEntry,
normalizeRelPath,
parseEmbedding,
} from "./internal.js";
import { bm25RankToScore, buildFtsQuery, mergeHybridResults } from "./hybrid.js";
import { searchKeyword, searchVector } from "./manager-search.js";
import { ensureMemoryIndexSchema } from "./memory-schema.js";
import { loadSqliteVecExtension } from "./sqlite-vec.js";
import { requireNodeSqlite } from "./sqlite.js";
type MemorySource = "memory" | "sessions";
export type MemorySearchResult = {
path: string;
startLine: number;
endLine: number;
score: number;
snippet: string;
source: MemorySource;
};
import { loadSqliteVecExtension } from "./sqlite-vec.js";
import type {
MemoryProviderStatus,
MemorySearchManager,
MemorySearchResult,
MemorySource,
MemorySyncProgressUpdate,
} from "./types.js";
type MemoryIndexMeta = {
model: string;
@@ -74,12 +71,6 @@ type SessionFileEntry = {
content: string;
};
type MemorySyncProgressUpdate = {
completed: number;
total: number;
label?: string;
};
type MemorySyncProgressState = {
completed: number;
total: number;
@@ -114,9 +105,9 @@ const INDEX_CACHE = new Map<string, MemoryIndexManager>();
const vectorToBlob = (embedding: number[]): Buffer =>
Buffer.from(new Float32Array(embedding).buffer);
export class MemoryIndexManager {
export class MemoryIndexManager implements MemorySearchManager {
private readonly cacheKey: string;
private readonly cfg: OpenClawConfig;
private readonly cfg: MoltbotConfig;
private readonly agentId: string;
private readonly workspaceDir: string;
private readonly settings: ResolvedMemorySearchConfig;
@@ -172,20 +163,16 @@ export class MemoryIndexManager {
private syncing: Promise<void> | null = null;
static async get(params: {
cfg: OpenClawConfig;
cfg: MoltbotConfig;
agentId: string;
}): Promise<MemoryIndexManager | null> {
const { cfg, agentId } = params;
const settings = resolveMemorySearchConfig(cfg, agentId);
if (!settings) {
return null;
}
if (!settings) return null;
const workspaceDir = resolveAgentWorkspaceDir(cfg, agentId);
const key = `${agentId}:${workspaceDir}:${JSON.stringify(settings)}`;
const existing = INDEX_CACHE.get(key);
if (existing) {
return existing;
}
if (existing) return existing;
const providerResult = await createEmbeddingProvider({
config: cfg,
agentDir: resolveAgentDir(cfg, agentId),
@@ -209,7 +196,7 @@ export class MemoryIndexManager {
private constructor(params: {
cacheKey: string;
cfg: OpenClawConfig;
cfg: MoltbotConfig;
agentId: string;
workspaceDir: string;
settings: ResolvedMemorySearchConfig;
@@ -252,19 +239,13 @@ export class MemoryIndexManager {
}
async warmSession(sessionKey?: string): Promise<void> {
if (!this.settings.sync.onSessionStart) {
return;
}
if (!this.settings.sync.onSessionStart) return;
const key = sessionKey?.trim() || "";
if (key && this.sessionWarm.has(key)) {
return;
}
if (key && this.sessionWarm.has(key)) return;
void this.sync({ reason: "session-start" }).catch((err) => {
log.warn(`memory sync failed (session-start): ${String(err)}`);
});
if (key) {
this.sessionWarm.add(key);
}
if (key) this.sessionWarm.add(key);
}
async search(
@@ -282,9 +263,7 @@ export class MemoryIndexManager {
});
}
const cleaned = query.trim();
if (!cleaned) {
return [];
}
if (!cleaned) return [];
const minScore = opts?.minScore ?? this.settings.query.minScore;
const maxResults = opts?.maxResults ?? this.settings.query.maxResults;
const hybrid = this.settings.query.hybrid;
@@ -343,9 +322,7 @@ export class MemoryIndexManager {
query: string,
limit: number,
): Promise<Array<MemorySearchResult & { id: string; textScore: number }>> {
if (!this.fts.enabled || !this.fts.available) {
return [];
}
if (!this.fts.enabled || !this.fts.available) return [];
const sourceFilter = this.buildSourceFilter();
const results = await searchKeyword({
db: this.db,
@@ -397,9 +374,7 @@ export class MemoryIndexManager {
force?: boolean;
progress?: (update: MemorySyncProgressUpdate) => void;
}): Promise<void> {
if (this.syncing) {
return this.syncing;
}
if (this.syncing) return this.syncing;
this.syncing = this.runSync(params).finally(() => {
this.syncing = null;
});
@@ -411,54 +386,13 @@ export class MemoryIndexManager {
from?: number;
lines?: number;
}): Promise<{ text: string; path: string }> {
const rawPath = params.relPath.trim();
if (!rawPath) {
const relPath = normalizeRelPath(params.relPath);
if (!relPath || !isMemoryPath(relPath)) {
throw new Error("path required");
}
const absPath = path.isAbsolute(rawPath)
? path.resolve(rawPath)
: path.resolve(this.workspaceDir, rawPath);
const relPath = path.relative(this.workspaceDir, absPath).replace(/\\/g, "/");
const inWorkspace =
relPath.length > 0 && !relPath.startsWith("..") && !path.isAbsolute(relPath);
const allowedWorkspace = inWorkspace && isMemoryPath(relPath);
let allowedAdditional = false;
if (!allowedWorkspace && this.settings.extraPaths.length > 0) {
const additionalPaths = normalizeExtraMemoryPaths(
this.workspaceDir,
this.settings.extraPaths,
);
for (const additionalPath of additionalPaths) {
try {
const stat = await fs.lstat(additionalPath);
if (stat.isSymbolicLink()) {
continue;
}
if (stat.isDirectory()) {
if (absPath === additionalPath || absPath.startsWith(`${additionalPath}${path.sep}`)) {
allowedAdditional = true;
break;
}
continue;
}
if (stat.isFile()) {
if (absPath === additionalPath && absPath.endsWith(".md")) {
allowedAdditional = true;
break;
}
}
} catch {}
}
}
if (!allowedWorkspace && !allowedAdditional) {
throw new Error("path required");
}
if (!absPath.endsWith(".md")) {
throw new Error("path required");
}
const stat = await fs.lstat(absPath);
if (stat.isSymbolicLink() || !stat.isFile()) {
throw new Error("path required");
const absPath = path.resolve(this.workspaceDir, relPath);
if (!absPath.startsWith(this.workspaceDir)) {
throw new Error("path escapes workspace");
}
const content = await fs.readFile(absPath, "utf-8");
if (!params.from && !params.lines) {
@@ -471,40 +405,7 @@ export class MemoryIndexManager {
return { text: slice.join("\n"), path: relPath };
}
status(): {
files: number;
chunks: number;
dirty: boolean;
workspaceDir: string;
dbPath: string;
provider: string;
model: string;
requestedProvider: string;
sources: MemorySource[];
extraPaths: string[];
sourceCounts: Array<{ source: MemorySource; files: number; chunks: number }>;
cache?: { enabled: boolean; entries?: number; maxEntries?: number };
fts?: { enabled: boolean; available: boolean; error?: string };
fallback?: { from: string; reason?: string };
vector?: {
enabled: boolean;
available?: boolean;
extensionPath?: string;
loadError?: string;
dims?: number;
};
batch?: {
enabled: boolean;
failures: number;
limit: number;
wait: boolean;
concurrency: number;
pollIntervalMs: number;
timeoutMs: number;
lastError?: string;
lastProvider?: string;
};
} {
status(): MemoryProviderStatus {
const sourceFilter = this.buildSourceFilter();
const files = this.db
.prepare(`SELECT COUNT(*) as c FROM files WHERE 1=1${sourceFilter.sql}`)
@@ -518,9 +419,7 @@ export class MemoryIndexManager {
};
const sourceCounts = (() => {
const sources = Array.from(this.sources);
if (sources.length === 0) {
return [];
}
if (sources.length === 0) return [];
const bySource = new Map<MemorySource, { files: number; chunks: number }>();
for (const source of sources) {
bySource.set(source, { files: 0, chunks: 0 });
@@ -545,19 +444,19 @@ export class MemoryIndexManager {
entry.chunks = row.c ?? 0;
bySource.set(row.source, entry);
}
return sources.map((source) => Object.assign({ source }, bySource.get(source)!));
return sources.map((source) => ({ source, ...bySource.get(source)! }));
})();
return {
backend: "builtin",
files: files?.c ?? 0,
chunks: chunks?.c ?? 0,
dirty: this.dirty,
dirty: this.dirty || this.sessionsDirty,
workspaceDir: this.workspaceDir,
dbPath: this.settings.store.path,
provider: this.provider.id,
model: this.provider.model,
requestedProvider: this.requestedProvider,
sources: Array.from(this.sources),
extraPaths: this.settings.extraPaths,
sourceCounts,
cache: this.cache.enabled
? {
@@ -601,9 +500,7 @@ export class MemoryIndexManager {
}
async probeVectorAvailability(): Promise<boolean> {
if (!this.vector.enabled) {
return false;
}
if (!this.vector.enabled) return false;
return this.ensureVectorReady();
}
@@ -618,9 +515,7 @@ export class MemoryIndexManager {
}
async close(): Promise<void> {
if (this.closed) {
return;
}
if (this.closed) return;
this.closed = true;
if (this.watchTimer) {
clearTimeout(this.watchTimer);
@@ -647,9 +542,7 @@ export class MemoryIndexManager {
}
private async ensureVectorReady(dimensions?: number): Promise<boolean> {
if (!this.vector.enabled) {
return false;
}
if (!this.vector.enabled) return false;
if (!this.vectorReady) {
this.vectorReady = this.withTimeout(
this.loadVectorExtension(),
@@ -675,9 +568,7 @@ export class MemoryIndexManager {
}
private async loadVectorExtension(): Promise<boolean> {
if (this.vector.available !== null) {
return this.vector.available;
}
if (this.vector.available !== null) return this.vector.available;
if (!this.vector.enabled) {
this.vector.available = false;
return false;
@@ -687,9 +578,7 @@ export class MemoryIndexManager {
? 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");
}
if (!loaded.ok) throw new Error(loaded.error ?? "unknown sqlite-vec load error");
this.vector.extensionPath = loaded.extensionPath;
this.vector.available = true;
return true;
@@ -703,9 +592,7 @@ export class MemoryIndexManager {
}
private ensureVectorTable(dimensions: number): void {
if (this.vector.dims === dimensions) {
return;
}
if (this.vector.dims === dimensions) return;
if (this.vector.dims && this.vector.dims !== dimensions) {
this.dropVectorTable();
}
@@ -729,9 +616,7 @@ export class MemoryIndexManager {
private buildSourceFilter(alias?: string): { sql: string; params: MemorySource[] } {
const sources = Array.from(this.sources);
if (sources.length === 0) {
return { sql: "", params: [] };
}
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 };
@@ -750,9 +635,7 @@ export class MemoryIndexManager {
}
private seedEmbeddingCache(sourceDb: DatabaseSync): void {
if (!this.cache.enabled) {
return;
}
if (!this.cache.enabled) return;
try {
const rows = sourceDb
.prepare(
@@ -767,9 +650,7 @@ export class MemoryIndexManager {
dims: number | null;
updated_at: number;
}>;
if (!rows.length) {
return;
}
if (!rows.length) return;
const insert = this.db.prepare(
`INSERT INTO ${EMBEDDING_CACHE_TABLE} (provider, model, provider_key, hash, embedding, dims, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?)
@@ -846,26 +727,12 @@ export class MemoryIndexManager {
}
private ensureWatcher() {
if (!this.sources.has("memory") || !this.settings.sync.watch || this.watcher) {
return;
}
const additionalPaths = normalizeExtraMemoryPaths(this.workspaceDir, this.settings.extraPaths)
.map((entry) => {
try {
const stat = fsSync.lstatSync(entry);
return stat.isSymbolicLink() ? null : entry;
} catch {
return null;
}
})
.filter((entry): entry is string => Boolean(entry));
const watchPaths = new Set<string>([
if (!this.sources.has("memory") || !this.settings.sync.watch || this.watcher) return;
const watchPaths = [
path.join(this.workspaceDir, "MEMORY.md"),
path.join(this.workspaceDir, "memory.md"),
path.join(this.workspaceDir, "memory"),
...additionalPaths,
]);
this.watcher = chokidar.watch(Array.from(watchPaths), {
];
this.watcher = chokidar.watch(watchPaths, {
ignoreInitial: true,
awaitWriteFinish: {
stabilityThreshold: this.settings.sync.watchDebounceMs,
@@ -882,26 +749,18 @@ export class MemoryIndexManager {
}
private ensureSessionListener() {
if (!this.sources.has("sessions") || this.sessionUnsubscribe) {
return;
}
if (!this.sources.has("sessions") || this.sessionUnsubscribe) return;
this.sessionUnsubscribe = onSessionTranscriptUpdate((update) => {
if (this.closed) {
return;
}
if (this.closed) return;
const sessionFile = update.sessionFile;
if (!this.isSessionFileForAgent(sessionFile)) {
return;
}
if (!this.isSessionFileForAgent(sessionFile)) return;
this.scheduleSessionDirty(sessionFile);
});
}
private scheduleSessionDirty(sessionFile: string) {
this.sessionPendingFiles.add(sessionFile);
if (this.sessionWatchTimer) {
return;
}
if (this.sessionWatchTimer) return;
this.sessionWatchTimer = setTimeout(() => {
this.sessionWatchTimer = null;
void this.processSessionDeltaBatch().catch((err) => {
@@ -911,17 +770,13 @@ export class MemoryIndexManager {
}
private async processSessionDeltaBatch(): Promise<void> {
if (this.sessionPendingFiles.size === 0) {
return;
}
if (this.sessionPendingFiles.size === 0) return;
const pending = Array.from(this.sessionPendingFiles);
this.sessionPendingFiles.clear();
let shouldSync = false;
for (const sessionFile of pending) {
const delta = await this.updateSessionDelta(sessionFile);
if (!delta) {
continue;
}
if (!delta) continue;
const bytesThreshold = delta.deltaBytes;
const messagesThreshold = delta.deltaMessages;
const bytesHit =
@@ -930,9 +785,7 @@ export class MemoryIndexManager {
messagesThreshold <= 0
? delta.pendingMessages > 0
: delta.pendingMessages >= messagesThreshold;
if (!bytesHit && !messagesHit) {
continue;
}
if (!bytesHit && !messagesHit) continue;
this.sessionsDirtyFiles.add(sessionFile);
this.sessionsDirty = true;
delta.pendingBytes =
@@ -955,9 +808,7 @@ export class MemoryIndexManager {
pendingMessages: number;
} | null> {
const thresholds = this.settings.sync.sessions;
if (!thresholds) {
return null;
}
if (!thresholds) return null;
let stat: { size: number };
try {
stat = await fs.stat(sessionFile);
@@ -1008,9 +859,7 @@ export class MemoryIndexManager {
}
private async countNewlines(absPath: string, start: number, end: number): Promise<number> {
if (end <= start) {
return 0;
}
if (end <= start) return 0;
const handle = await fs.open(absPath, "r");
try {
let offset = start;
@@ -1019,13 +868,9 @@ export class MemoryIndexManager {
while (offset < end) {
const toRead = Math.min(buffer.length, end - offset);
const { bytesRead } = await handle.read(buffer, 0, toRead, offset);
if (bytesRead <= 0) {
break;
}
if (bytesRead <= 0) break;
for (let i = 0; i < bytesRead; i += 1) {
if (buffer[i] === 10) {
count += 1;
}
if (buffer[i] === 10) count += 1;
}
offset += bytesRead;
}
@@ -1037,18 +882,14 @@ export class MemoryIndexManager {
private resetSessionDelta(absPath: string, size: number): void {
const state = this.sessionDeltas.get(absPath);
if (!state) {
return;
}
if (!state) return;
state.lastSize = size;
state.pendingBytes = 0;
state.pendingMessages = 0;
}
private isSessionFileForAgent(sessionFile: string): boolean {
if (!sessionFile) {
return false;
}
if (!sessionFile) return false;
const sessionsDir = resolveSessionTranscriptsDirForAgent(this.agentId);
const resolvedFile = path.resolve(sessionFile);
const resolvedDir = path.resolve(sessionsDir);
@@ -1057,9 +898,7 @@ export class MemoryIndexManager {
private ensureIntervalSync() {
const minutes = this.settings.sync.intervalMinutes;
if (!minutes || minutes <= 0 || this.intervalTimer) {
return;
}
if (!minutes || minutes <= 0 || this.intervalTimer) return;
const ms = minutes * 60 * 1000;
this.intervalTimer = setInterval(() => {
void this.sync({ reason: "interval" }).catch((err) => {
@@ -1069,12 +908,8 @@ export class MemoryIndexManager {
}
private scheduleWatchSync() {
if (!this.sources.has("memory") || !this.settings.sync.watch) {
return;
}
if (this.watchTimer) {
clearTimeout(this.watchTimer);
}
if (!this.sources.has("memory") || !this.settings.sync.watch) return;
if (this.watchTimer) clearTimeout(this.watchTimer);
this.watchTimer = setTimeout(() => {
this.watchTimer = null;
void this.sync({ reason: "watch" }).catch((err) => {
@@ -1087,19 +922,11 @@ export class MemoryIndexManager {
params?: { reason?: string; force?: boolean },
needsFullReindex = false,
) {
if (!this.sources.has("sessions")) {
return false;
}
if (params?.force) {
return true;
}
if (!this.sources.has("sessions")) return false;
if (params?.force) return true;
const reason = params?.reason;
if (reason === "session-start" || reason === "watch") {
return false;
}
if (needsFullReindex) {
return true;
}
if (reason === "session-start" || reason === "watch") return false;
if (needsFullReindex) return true;
return this.sessionsDirty && this.sessionsDirtyFiles.size > 0;
}
@@ -1107,7 +934,7 @@ export class MemoryIndexManager {
needsFullReindex: boolean;
progress?: MemorySyncProgressState;
}) {
const files = await listMemoryFiles(this.workspaceDir, this.settings.extraPaths);
const files = await listMemoryFiles(this.workspaceDir);
const fileEntries = await Promise.all(
files.map(async (file) => buildFileEntry(file, this.workspaceDir)),
);
@@ -1156,9 +983,7 @@ export class MemoryIndexManager {
.prepare(`SELECT path FROM files WHERE source = ?`)
.all("memory") as Array<{ path: string }>;
for (const stale of staleRows) {
if (activePaths.has(stale.path)) {
continue;
}
if (activePaths.has(stale.path)) continue;
this.db.prepare(`DELETE FROM files WHERE path = ? AND source = ?`).run(stale.path, "memory");
try {
this.db
@@ -1253,9 +1078,7 @@ export class MemoryIndexManager {
.prepare(`SELECT path FROM files WHERE source = ?`)
.all("sessions") as Array<{ path: string }>;
for (const stale of staleRows) {
if (activePaths.has(stale.path)) {
continue;
}
if (activePaths.has(stale.path)) continue;
this.db
.prepare(`DELETE FROM files WHERE path = ? AND source = ?`)
.run(stale.path, "sessions");
@@ -1287,9 +1110,7 @@ export class MemoryIndexManager {
total: 0,
label: undefined,
report: (update) => {
if (update.label) {
state.label = update.label;
}
if (update.label) state.label = update.label;
const label =
update.total > 0 && state.label
? `${state.label} ${update.completed}/${update.total}`
@@ -1400,12 +1221,8 @@ export class MemoryIndexManager {
private async activateFallbackProvider(reason: string): Promise<boolean> {
const fallback = this.settings.fallback;
if (!fallback || fallback === "none" || fallback === this.provider.id) {
return false;
}
if (this.fallbackFrom) {
return false;
}
if (!fallback || fallback === "none" || fallback === this.provider.id) return false;
if (this.fallbackFrom) return false;
const fallbackFrom = this.provider.id as "openai" | "gemini" | "local";
const fallbackModel =
@@ -1557,9 +1374,7 @@ export class MemoryIndexManager {
const row = this.db.prepare(`SELECT value FROM meta WHERE key = ?`).get(META_KEY) as
| { value: string }
| undefined;
if (!row?.value) {
return null;
}
if (!row?.value) return null;
try {
return JSON.parse(row.value) as MemoryIndexMeta;
} catch {
@@ -1606,26 +1421,16 @@ export class MemoryIndexManager {
const normalized = this.normalizeSessionText(content);
return normalized ? normalized : null;
}
if (!Array.isArray(content)) {
return null;
}
if (!Array.isArray(content)) return null;
const parts: string[] = [];
for (const block of content) {
if (!block || typeof block !== "object") {
continue;
}
if (!block || typeof block !== "object") continue;
const record = block as { type?: unknown; text?: unknown };
if (record.type !== "text" || typeof record.text !== "string") {
continue;
}
if (record.type !== "text" || typeof record.text !== "string") continue;
const normalized = this.normalizeSessionText(record.text);
if (normalized) {
parts.push(normalized);
}
}
if (parts.length === 0) {
return null;
if (normalized) parts.push(normalized);
}
if (parts.length === 0) return null;
return parts.join(" ");
}
@@ -1636,9 +1441,7 @@ export class MemoryIndexManager {
const lines = raw.split("\n");
const collected: string[] = [];
for (const line of lines) {
if (!line.trim()) {
continue;
}
if (!line.trim()) continue;
let record: unknown;
try {
record = JSON.parse(line);
@@ -1655,16 +1458,10 @@ export class MemoryIndexManager {
const message = (record as { message?: unknown }).message as
| { role?: unknown; content?: unknown }
| undefined;
if (!message || typeof message.role !== "string") {
continue;
}
if (message.role !== "user" && message.role !== "assistant") {
continue;
}
if (!message || typeof message.role !== "string") continue;
if (message.role !== "user" && message.role !== "assistant") continue;
const text = this.extractSessionText(message.content);
if (!text) {
continue;
}
if (!text) continue;
const label = message.role === "user" ? "User" : "Assistant";
collected.push(`${label}: ${text}`);
}
@@ -1684,9 +1481,7 @@ export class MemoryIndexManager {
}
private estimateEmbeddingTokens(text: string): number {
if (!text) {
return 0;
}
if (!text) return 0;
return Math.ceil(text.length / EMBEDDING_APPROX_CHARS_PER_TOKEN);
}
@@ -1719,27 +1514,17 @@ export class MemoryIndexManager {
}
private loadEmbeddingCache(hashes: string[]): Map<string, number[]> {
if (!this.cache.enabled) {
return new Map();
}
if (hashes.length === 0) {
return new Map();
}
if (!this.cache.enabled) return new Map();
if (hashes.length === 0) return new Map();
const unique: string[] = [];
const seen = new Set<string>();
for (const hash of hashes) {
if (!hash) {
continue;
}
if (seen.has(hash)) {
continue;
}
if (!hash) continue;
if (seen.has(hash)) continue;
seen.add(hash);
unique.push(hash);
}
if (unique.length === 0) {
return new Map();
}
if (unique.length === 0) return new Map();
const out = new Map<string, number[]>();
const baseParams = [this.provider.id, this.provider.model, this.providerKey];
@@ -1761,12 +1546,8 @@ export class MemoryIndexManager {
}
private upsertEmbeddingCache(entries: Array<{ hash: string; embedding: number[] }>): void {
if (!this.cache.enabled) {
return;
}
if (entries.length === 0) {
return;
}
if (!this.cache.enabled) return;
if (entries.length === 0) return;
const now = Date.now();
const stmt = this.db.prepare(
`INSERT INTO ${EMBEDDING_CACHE_TABLE} (provider, model, provider_key, hash, embedding, dims, updated_at)\n` +
@@ -1791,20 +1572,14 @@ export class MemoryIndexManager {
}
private pruneEmbeddingCacheIfNeeded(): void {
if (!this.cache.enabled) {
return;
}
if (!this.cache.enabled) return;
const max = this.cache.maxEntries;
if (!max || max <= 0) {
return;
}
if (!max || max <= 0) return;
const row = this.db.prepare(`SELECT COUNT(*) as c FROM ${EMBEDDING_CACHE_TABLE}`).get() as
| { c: number }
| undefined;
const count = row?.c ?? 0;
if (count <= max) {
return;
}
if (count <= max) return;
const excess = count - max;
this.db
.prepare(
@@ -1819,9 +1594,7 @@ export class MemoryIndexManager {
}
private async embedChunksInBatches(chunks: MemoryChunk[]): Promise<number[][]> {
if (chunks.length === 0) {
return [];
}
if (chunks.length === 0) return [];
const cached = this.loadEmbeddingCache(chunks.map((chunk) => chunk.hash));
const embeddings: number[][] = Array.from({ length: chunks.length }, () => []);
const missing: Array<{ index: number; chunk: MemoryChunk }> = [];
@@ -1836,9 +1609,7 @@ export class MemoryIndexManager {
}
}
if (missing.length === 0) {
return embeddings;
}
if (missing.length === 0) return embeddings;
const missingChunks = missing.map((m) => m.chunk);
const batches = this.buildEmbeddingBatches(missingChunks);
@@ -1864,7 +1635,7 @@ export class MemoryIndexManager {
if (this.provider.id === "openai" && this.openAi) {
const entries = Object.entries(this.openAi.headers)
.filter(([key]) => key.toLowerCase() !== "authorization")
.toSorted(([a], [b]) => a.localeCompare(b))
.sort(([a], [b]) => a.localeCompare(b))
.map(([key, value]) => [key, value]);
return hashText(
JSON.stringify({
@@ -1881,7 +1652,7 @@ export class MemoryIndexManager {
const lower = key.toLowerCase();
return lower !== "authorization" && lower !== "x-goog-api-key";
})
.toSorted(([a], [b]) => a.localeCompare(b))
.sort(([a], [b]) => a.localeCompare(b))
.map(([key, value]) => [key, value]);
return hashText(
JSON.stringify({
@@ -1918,9 +1689,7 @@ export class MemoryIndexManager {
if (!openAi) {
return this.embedChunksInBatches(chunks);
}
if (chunks.length === 0) {
return [];
}
if (chunks.length === 0) return [];
const cached = this.loadEmbeddingCache(chunks.map((chunk) => chunk.hash));
const embeddings: number[][] = Array.from({ length: chunks.length }, () => []);
const missing: Array<{ index: number; chunk: MemoryChunk }> = [];
@@ -1935,9 +1704,7 @@ export class MemoryIndexManager {
}
}
if (missing.length === 0) {
return embeddings;
}
if (missing.length === 0) return embeddings;
const requests: OpenAiBatchRequest[] = [];
const mapping = new Map<string, { index: number; hash: string }>();
@@ -1972,17 +1739,13 @@ export class MemoryIndexManager {
}),
fallback: async () => await this.embedChunksInBatches(chunks),
});
if (Array.isArray(batchResult)) {
return batchResult;
}
if (Array.isArray(batchResult)) return batchResult;
const byCustomId = batchResult;
const toCache: Array<{ hash: string; embedding: number[] }> = [];
for (const [customId, embedding] of byCustomId.entries()) {
const mapped = mapping.get(customId);
if (!mapped) {
continue;
}
if (!mapped) continue;
embeddings[mapped.index] = embedding;
toCache.push({ hash: mapped.hash, embedding });
}
@@ -1999,9 +1762,7 @@ export class MemoryIndexManager {
if (!gemini) {
return this.embedChunksInBatches(chunks);
}
if (chunks.length === 0) {
return [];
}
if (chunks.length === 0) return [];
const cached = this.loadEmbeddingCache(chunks.map((chunk) => chunk.hash));
const embeddings: number[][] = Array.from({ length: chunks.length }, () => []);
const missing: Array<{ index: number; chunk: MemoryChunk }> = [];
@@ -2016,9 +1777,7 @@ export class MemoryIndexManager {
}
}
if (missing.length === 0) {
return embeddings;
}
if (missing.length === 0) return embeddings;
const requests: GeminiBatchRequest[] = [];
const mapping = new Map<string, { index: number; hash: string }>();
@@ -2050,17 +1809,13 @@ export class MemoryIndexManager {
}),
fallback: async () => await this.embedChunksInBatches(chunks),
});
if (Array.isArray(batchResult)) {
return batchResult;
}
if (Array.isArray(batchResult)) return batchResult;
const byCustomId = batchResult;
const toCache: Array<{ hash: string; embedding: number[] }> = [];
for (const [customId, embedding] of byCustomId.entries()) {
const mapped = mapping.get(customId);
if (!mapped) {
continue;
}
if (!mapped) continue;
embeddings[mapped.index] = embedding;
toCache.push({ hash: mapped.hash, embedding });
}
@@ -2069,9 +1824,7 @@ export class MemoryIndexManager {
}
private async embedBatchWithRetry(texts: string[]): Promise<number[][]> {
if (texts.length === 0) {
return [];
}
if (texts.length === 0) return [];
let attempt = 0;
let delayMs = EMBEDDING_RETRY_BASE_DELAY_MS;
while (true) {
@@ -2133,9 +1886,7 @@ export class MemoryIndexManager {
timeoutMs: number,
message: string,
): Promise<T> {
if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) {
return await promise;
}
if (!Number.isFinite(timeoutMs) || timeoutMs <= 0) return await promise;
let timer: NodeJS.Timeout | null = null;
const timeoutPromise = new Promise<never>((_, reject) => {
timer = setTimeout(() => reject(new Error(message)), timeoutMs);
@@ -2143,16 +1894,12 @@ export class MemoryIndexManager {
try {
return (await Promise.race([promise, timeoutPromise])) as T;
} finally {
if (timer) {
clearTimeout(timer);
}
if (timer) clearTimeout(timer);
}
}
private async runWithConcurrency<T>(tasks: Array<() => Promise<T>>, limit: number): Promise<T[]> {
if (tasks.length === 0) {
return [];
}
if (tasks.length === 0) return [];
const resolvedLimit = Math.max(1, Math.min(limit, tasks.length));
const results: T[] = Array.from({ length: tasks.length });
let next = 0;
@@ -2160,14 +1907,10 @@ export class MemoryIndexManager {
const workers = Array.from({ length: resolvedLimit }, async () => {
while (true) {
if (firstError) {
return;
}
if (firstError) return;
const index = next;
next += 1;
if (index >= tasks.length) {
return;
}
if (index >= tasks.length) return;
try {
results[index] = await tasks[index]();
} catch (err) {
@@ -2178,9 +1921,7 @@ export class MemoryIndexManager {
});
await Promise.allSettled(workers);
if (firstError) {
throw firstError;
}
if (firstError) throw firstError;
return results;
}