Add memory persistence migration and repository adapter methods

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);

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;
+    },
+  };
+}

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[]> | number[] }} [options] - Optional ingest tuning.
+ * @param {{ maxSummaryLength?: number, embedText?: (value: string) => Promise<number[]> | number[], repository?: typeof memoryRepository }} [options] - Optional ingest tuning.
  * @returns {Promise<import('./schema.js').MemoryRecord[]>} 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<import('./schema.js').MemoryRecord[]>} 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,12 +114,20 @@ 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,
     });
@@ -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<import('./schema.js').MemoryRecord | null>} 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<import('./schema.js').MemoryRecord | null>} 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;
+}

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<unknown> }} repository
+ * @param {import('./schema.js').MemoryRecord} memory
+ */
+export async function insertMemory(repository, memory) {
+  return repository.insertMemory(memory);
+}

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<import('./schema.js').MemoryRecord[]> }} 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 });
+}

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<unknown> }} 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)));
+}