diff --git a/migrations/create_memories_table.sql b/migrations/create_memories_table.sql new file mode 100644 index 0000000..6fd3d58 --- /dev/null +++ b/migrations/create_memories_table.sql @@ -0,0 +1,46 @@ +CREATE TABLE IF NOT EXISTS memories ( + id UUID PRIMARY KEY, + schema_version INTEGER NOT NULL DEFAULT 1, + agent_id TEXT NOT NULL, + session_id TEXT NOT NULL, + type TEXT NOT NULL, + summary TEXT NOT NULL DEFAULT '', + content JSONB NOT NULL, + tags TEXT[] NOT NULL DEFAULT '{}', + importance DOUBLE PRECISION NOT NULL DEFAULT 0.5, + embedding_id TEXT, + source JSONB NOT NULL DEFAULT '{}'::jsonb, + created_at BIGINT NOT NULL, + last_seen BIGINT NOT NULL, + ttl_days INTEGER, + merged_into UUID REFERENCES memories(id) ON DELETE SET NULL, + pinned BOOLEAN NOT NULL DEFAULT FALSE, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE TABLE IF NOT EXISTS memory_links ( + id BIGSERIAL PRIMARY KEY, + from_memory_id UUID NOT NULL REFERENCES memories(id) ON DELETE CASCADE, + to_memory_id UUID NOT NULL REFERENCES memories(id) ON DELETE CASCADE, + relation_type TEXT NOT NULL DEFAULT 'merged_into', + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (from_memory_id, to_memory_id, relation_type) +); + +CREATE TABLE IF NOT EXISTS memory_admin_flags ( + memory_id UUID PRIMARY KEY REFERENCES memories(id) ON DELETE CASCADE, + pinned BOOLEAN NOT NULL DEFAULT FALSE, + pinned_at TIMESTAMPTZ, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE INDEX IF NOT EXISTS idx_memories_agent_id ON memories (agent_id); +CREATE INDEX IF NOT EXISTS idx_memories_created_at ON memories (created_at DESC); +CREATE INDEX IF NOT EXISTS idx_memories_last_seen ON memories (last_seen DESC); +CREATE INDEX IF NOT EXISTS idx_memories_importance ON memories (importance DESC); +CREATE INDEX IF NOT EXISTS idx_memories_tags_gin ON memories USING GIN (tags); +CREATE INDEX IF NOT EXISTS idx_memories_pinned ON memories (pinned); +CREATE INDEX IF NOT EXISTS idx_memories_merged_into ON memories (merged_into); + +CREATE INDEX IF NOT EXISTS idx_memory_links_relation ON memory_links (relation_type, from_memory_id, to_memory_id); +CREATE INDEX IF NOT EXISTS idx_memory_admin_flags_pinned ON memory_admin_flags (pinned); diff --git a/src/services/memory/db-adapter.js b/src/services/memory/db-adapter.js new file mode 100644 index 0000000..4f770c1 --- /dev/null +++ b/src/services/memory/db-adapter.js @@ -0,0 +1,141 @@ +/** + * @typedef {{ query: (sql: string, params?: unknown[]) => Promise<{ rows?: any[], rowCount?: number }> }} QueryableDb + */ + +/** + * @param {QueryableDb} db + */ +export function createMemoryRepository(db) { + return { + /** + * @param {import('./schema.js').MemoryRecord} memory + */ + async insertMemory(memory) { + const result = await db.query( + `INSERT INTO memories ( + id, schema_version, agent_id, session_id, type, summary, + content, tags, importance, embedding_id, source, + created_at, last_seen, ttl_days, merged_into, pinned + ) VALUES ( + $1, $2, $3, $4, $5, $6, + $7::jsonb, $8::text[], $9, $10, $11::jsonb, + $12, $13, $14, $15, $16 + ) + ON CONFLICT (id) DO UPDATE SET + summary = EXCLUDED.summary, + content = EXCLUDED.content, + tags = EXCLUDED.tags, + importance = EXCLUDED.importance, + embedding_id = EXCLUDED.embedding_id, + source = EXCLUDED.source, + last_seen = EXCLUDED.last_seen, + ttl_days = EXCLUDED.ttl_days, + merged_into = EXCLUDED.merged_into, + pinned = EXCLUDED.pinned, + updated_at = NOW() + RETURNING *`, + [ + memory.id, + memory.schema_version, + memory.agent_id, + memory.session_id, + memory.type, + memory.summary, + JSON.stringify({ text: memory.content }), + memory.tags, + memory.importance, + memory.embedding_id, + JSON.stringify({ value: memory.source }), + memory.created_at, + memory.last_seen, + memory.ttl_days, + memory.merged_into, + memory.pinned, + ] + ); + + return result.rows?.[0] ?? null; + }, + + /** + * @param {string} id + * @param {number} [lastSeen] + */ + async updateMemoryUsage(id, lastSeen = Date.now()) { + const result = await db.query( + `UPDATE memories + SET last_seen = $2, updated_at = NOW() + WHERE id = $1 + RETURNING *`, + [id, lastSeen] + ); + + return result.rows?.[0] ?? null; + }, + + /** + * @param {{ cutoff: number, limit?: number }} options + */ + async listPruneCandidates(options) { + const limit = Number.isFinite(options.limit) ? Number(options.limit) : 500; + const result = await db.query( + `SELECT * + FROM memories + WHERE pinned = FALSE + AND (merged_into IS NOT NULL OR last_seen < $1) + ORDER BY last_seen ASC + LIMIT $2`, + [options.cutoff, limit] + ); + + return result.rows ?? []; + }, + + /** + * @param {string} id + * @param {string} mergedInto + */ + async markMerged(id, mergedInto) { + const result = await db.query( + `WITH updated AS ( + UPDATE memories + SET merged_into = $2, updated_at = NOW() + WHERE id = $1 + RETURNING id + ) + INSERT INTO memory_links (from_memory_id, to_memory_id, relation_type) + SELECT id, $2, 'merged_into' FROM updated + ON CONFLICT (from_memory_id, to_memory_id, relation_type) DO NOTHING + RETURNING from_memory_id`, + [id, mergedInto] + ); + + return (result.rowCount ?? 0) > 0; + }, + + /** + * @param {string} id + * @param {boolean} pinned + */ + async setPinned(id, pinned) { + const result = await db.query( + `WITH updated AS ( + UPDATE memories + SET pinned = $2, updated_at = NOW(), last_seen = $3 + WHERE id = $1 + RETURNING id, pinned + ) + INSERT INTO memory_admin_flags (memory_id, pinned, pinned_at, updated_at) + SELECT id, pinned, CASE WHEN pinned THEN NOW() ELSE NULL END, NOW() FROM updated + ON CONFLICT (memory_id) DO UPDATE SET + pinned = EXCLUDED.pinned, + pinned_at = EXCLUDED.pinned_at, + updated_at = EXCLUDED.updated_at + RETURNING memory_id, pinned`, + [id, pinned, Date.now()] + ); + + return result.rows?.[0] ?? null; + }, + }; +} diff --git a/src/services/memory/index.js b/src/services/memory/index.js index 16043d9..d5d9d67 100644 --- a/src/services/memory/index.js +++ b/src/services/memory/index.js @@ -1,8 +1,8 @@ import { createMemoryCache } from './cache.js'; import { embedText } from './embedder.js'; -import { normalizeEvents } from './ingestor.js'; -import { selectPrunableMemories } from './pruner.js'; -import { filterAndRankMemories } from './retriever.js'; +import { insertMemory, normalizeEvents } from './ingestor.js'; +import { listPruneCandidates, selectPrunableMemories } from './pruner.js'; +import { filterAndRankMemories, updateMemoryUsage } from './retriever.js'; import { startScheduler } from './scheduler.js'; import { createMemoryRecord, validateMemoryItem } from './schema.js'; import { summarizeEvents } from './summarizer.js'; @@ -10,14 +10,45 @@ import { summarizeEvents } from './summarizer.js'; const memoryStore = new Map(); const cache = createMemoryCache(); +const memoryRepository = { + async insertMemory(memory) { + memoryStore.set(memory.id, memory); + return memory; + }, + async updateMemoryUsage(id, lastSeen = Date.now()) { + const existing = memoryStore.get(id); + if (!existing) return null; + const updated = { ...existing, last_seen: lastSeen }; + memoryStore.set(id, updated); + return updated; + }, + async listPruneCandidates({ cutoff }) { + return [...memoryStore.values()].filter((memory) => !memory.pinned && memory.last_seen < cutoff); + }, + async markMerged(id, mergedInto) { + const existing = memoryStore.get(id); + if (!existing) return false; + memoryStore.set(id, { ...existing, merged_into: mergedInto, last_seen: Date.now() }); + return true; + }, + async setPinned(id, pinned) { + const existing = memoryStore.get(id); + if (!existing) return null; + const updated = { ...existing, pinned, last_seen: Date.now() }; + memoryStore.set(id, updated); + return updated; + }, +}; + /** * Ingests agent events and stores them as memory records. * * @param {import('./schema.js').MemoryEvent[]} events - Ordered list of raw events to persist. - * @param {{ maxSummaryLength?: number, embedText?: (value: string) => Promise | number[] }} [options] - Optional ingest tuning. + * @param {{ maxSummaryLength?: number, embedText?: (value: string) => Promise | number[], repository?: typeof memoryRepository }} [options] - Optional ingest tuning. * @returns {Promise} Stored memory records. */ export async function ingestEvents(events, options = {}) { + const repository = options.repository ?? memoryRepository; const normalizedEvents = normalizeEvents(events); const records = []; @@ -52,7 +83,7 @@ export async function ingestEvents(events, options = {}) { throw new TypeError('Memory ingest rejected: invalid durable record format'); } - memoryStore.set(record.id, record); + await insertMemory(repository, record); records.push(record); } @@ -63,17 +94,19 @@ export async function ingestEvents(events, options = {}) { /** * Retrieves top-k memories for an agent query. * - * @param {{ agent_id: string, query?: string, tags?: string[], timeframe?: { from?: number, to?: number }, k?: number }} params - Retrieval filter and ranking inputs. + * @param {{ agent_id: string, query?: string, tags?: string[], timeframe?: { from?: number, to?: number }, k?: number, repository?: typeof memoryRepository }} params - Retrieval filter and ranking inputs. * @returns {Promise} Sorted list of relevant memories. */ export async function getRelevantMemories(params) { - const cacheKey = JSON.stringify(params); + const { repository = memoryRepository, ...queryParams } = params; + const cacheKey = JSON.stringify(queryParams); const cached = cache.get(cacheKey); if (cached) { return cached; } - const ranked = filterAndRankMemories([...memoryStore.values()], params); + const ranked = filterAndRankMemories([...memoryStore.values()], queryParams); + await updateMemoryUsage(repository, ranked.map((memory) => memory.id)); cache.set(cacheKey, ranked); return ranked; } @@ -81,16 +114,24 @@ export async function getRelevantMemories(params) { /** * Runs a memory pruning pass and removes unpinned stale records. * - * @param {{ retentionMs?: number, now?: number, scheduleEveryMs?: number }} [options] - Prune configuration; if scheduleEveryMs is set, returns a scheduler handle. + * @param {{ retentionMs?: number, now?: number, scheduleEveryMs?: number, repository?: typeof memoryRepository }} [options] - Prune configuration; if scheduleEveryMs is set, returns a scheduler handle. * @returns {Promise<{ pruned: import('./schema.js').MemoryRecord[], scheduler?: { stop: () => void } }>} Pruned records and optional scheduler handle. */ export async function runPruneCycle(options = {}) { + const repository = options.repository ?? memoryRepository; const retentionMs = options.retentionMs ?? (1000 * 60 * 60 * 24 * 30); - const prunable = selectPrunableMemories([...memoryStore.values()], { + const dbCandidates = await listPruneCandidates(repository, { retentionMs, now: options.now, }); + const prunable = dbCandidates.length > 0 + ? dbCandidates + : selectPrunableMemories([...memoryStore.values()], { + retentionMs, + now: options.now, + }); + for (const memory of prunable) { memoryStore.delete(memory.id); } @@ -99,7 +140,7 @@ export async function runPruneCycle(options = {}) { const result = { pruned: prunable }; if (Number.isFinite(options.scheduleEveryMs) && options.scheduleEveryMs > 0) { - result.scheduler = startScheduler(() => runPruneCycle({ retentionMs }), { + result.scheduler = startScheduler(() => runPruneCycle({ retentionMs, repository }), { intervalMs: options.scheduleEveryMs, }); } @@ -111,14 +152,10 @@ export async function runPruneCycle(options = {}) { * Pins a memory to prevent pruning and boost retrieval ranking. * * @param {string} id - Memory record identifier. - * @returns {import('./schema.js').MemoryRecord | null} Updated memory record or null when not found. + * @returns {Promise} Updated memory record or null when not found. */ -export function pinMemory(id) { - const memory = memoryStore.get(id); - if (!memory) return null; - - const updated = { ...memory, pinned: true, last_seen: Date.now() }; - memoryStore.set(id, updated); +export async function pinMemory(id) { + const updated = await memoryRepository.setPinned(id, true); cache.clear(); return updated; } @@ -127,14 +164,22 @@ export function pinMemory(id) { * Removes a pin from a memory so it can be pruned normally. * * @param {string} id - Memory record identifier. - * @returns {import('./schema.js').MemoryRecord | null} Updated memory record or null when not found. + * @returns {Promise} Updated memory record or null when not found. */ -export function unpinMemory(id) { - const memory = memoryStore.get(id); - if (!memory) return null; - - const updated = { ...memory, pinned: false, last_seen: Date.now() }; - memoryStore.set(id, updated); +export async function unpinMemory(id) { + const updated = await memoryRepository.setPinned(id, false); cache.clear(); return updated; } + +/** + * Marks one memory as merged into another. + * + * @param {string} id + * @param {string} mergedInto + */ +export async function markMemoryMerged(id, mergedInto) { + const merged = await memoryRepository.markMerged(id, mergedInto); + cache.clear(); + return merged; +} diff --git a/src/services/memory/ingestor.js b/src/services/memory/ingestor.js index 2217e29..c3bfdb8 100644 --- a/src/services/memory/ingestor.js +++ b/src/services/memory/ingestor.js @@ -7,3 +7,11 @@ import { normalizeEvent } from './schema.js'; export function normalizeEvents(events) { return events.map(normalizeEvent); } + +/** + * @param {{ insertMemory: (memory: import('./schema.js').MemoryRecord) => Promise }} repository + * @param {import('./schema.js').MemoryRecord} memory + */ +export async function insertMemory(repository, memory) { + return repository.insertMemory(memory); +} diff --git a/src/services/memory/pruner.js b/src/services/memory/pruner.js index 41e7790..44e90f0 100644 --- a/src/services/memory/pruner.js +++ b/src/services/memory/pruner.js @@ -9,3 +9,14 @@ export function selectPrunableMemories(memories, options) { return memories.filter((memory) => !memory.pinned && memory.last_seen < cutoff); } + + +/** + * @param {{ listPruneCandidates: (options: { cutoff: number, limit?: number }) => Promise }} repository + * @param {{ retentionMs: number, now?: number, limit?: number }} options + */ +export async function listPruneCandidates(repository, options) { + const now = options.now ?? Date.now(); + const cutoff = now - options.retentionMs; + return repository.listPruneCandidates({ cutoff, limit: options.limit }); +} diff --git a/src/services/memory/retriever.js b/src/services/memory/retriever.js index 0f39aac..5bb9031 100644 --- a/src/services/memory/retriever.js +++ b/src/services/memory/retriever.js @@ -41,3 +41,12 @@ export function filterAndRankMemories(memories, params) { .slice(0, k) .map((entry) => entry.memory); } + +/** + * @param {{ updateMemoryUsage: (id: string, lastSeen?: number) => Promise }} repository + * @param {string[]} ids + * @param {number} [lastSeen] + */ +export async function updateMemoryUsage(repository, ids, lastSeen = Date.now()) { + await Promise.all(ids.map((id) => repository.updateMemoryUsage(id, lastSeen))); +}