fix(memory): preserve vector data and harden persistence

.changeset/quiet-memories-upsert.md

@@ -0,0 +1,5 @@
+---
+'@agentskit/memory': patch
+---
+
+Fix memory backend edge cases: preserve Redis vector bytes, prefer stable Vectra IDs where supported, protect adapter-owned metadata fields, keep file and localStorage fallback writes atomic, make SQLite KV eviction transactional when the opener supports transactions, preserve same-event-loop localStorage writes and expiration purges, bound vector KV TTL recall to `k` and reject foreign collection rows, aggregate erasure failures, and retry failed lazy initialization after best-effort cleanup of failed internally owned resources. Mongo Atlas uses ordered ID upserts when supported by the injected collection, replacing canonical fields while preserving unrelated host fields.

.github/workflows/ci.yml

@@ -101,7 +101,6 @@ jobs:
           fi
 
       - name: Build all packages
-        if: steps.package-scope.outputs.run_all == 'true'
         run: pnpm exec turbo run build --filter='./packages/*' --concurrency=2
 
       - name: Build affected packages and dependents

packages/memory/README.md

@@ -116,14 +116,42 @@ persistence succeeds.
 - `milvusVectorStore` — Milvus/Zilliz REST API.
 - `mongoAtlasVectorStore` — MongoDB Atlas Vector Search with an injected collection.
 
+Mongo Atlas writes use ordered `_id` upserts when the collection provides
+`bulkWrite` or `updateOne`. Repeating an ID replaces the adapter-owned
+`content`, embedding, and `metadata` fields while preserving unrelated fields
+on the Mongo document; duplicate IDs in one batch resolve to the last document
+in input order. Older insert-only shims keep their insert behavior and cannot
+promise repeated-ID replacement.
+
+The Vectra file adapter uses stable native IDs when the installed Vectra
+version provides `upsertItem`. Older versions use insert-only writes, so
+legacy random-ID duplicates are not automatically removed. Adapter-owned
+`_id` and `content` metadata values take precedence over caller metadata with
+those names. Core `VectorSearchOptions.filter` is not yet translated by every
+remote vector adapter; consult the adapter implementation before relying on a
+filter to separate results.
+
 Vector adapters follow ADR 0003: a result is included only when its score is
 strictly greater than `threshold`. Remote HTTP adapters accept `timeoutMs` and
 `maxResponseBytes`, defaulting to 15 seconds and 2 MiB; both require positive
 safe integers. Injected `fetch` and caller `signal` are also supported. Timeout
 and response-limit failures use `AK_MEMORY_REMOTE_HTTP`. File KV writes are
 atomic and serialized across instances in one process. They are not a
 multi-process coordination primitive; use SQLite, Redis, or another external
-store when several processes write the same file.
+store when several processes write the same file. SQLite KV writes include
+eviction in a transaction when the configured SQLite opener exposes native
+transactions; simpler injected openers retain their non-transactional path.
+File rename publication is atomic on the same filesystem, but writes are not
+fsync-backed and the in-process queue does not coordinate separate processes.
+The localStorage backend uses synchronous storage read-modify-write and
+expiration purges for same-event-loop calls, but does not coordinate writes
+across browser tabs.
+
+Vector KV TTL hides expired entries from `get` and `recall` but does not
+physically delete their vector rows. TTL recall uses bounded over-fetch and
+returns at most `k`; it cannot guarantee that it finds every live row if more
+than the over-fetch window is occupied by expired or foreign-collection
+results. `maxMessages` eviction is not enforced by this vector KV adapter.
 
 Same 3-method `VectorStore` contract — swap without touching agent code.
 

packages/memory/src/file-vector.ts

@@ -29,6 +29,7 @@ interface VectraIndex {
   endUpdate(): Promise<void>
   cancelUpdate(): void
   insertItem(item: { vector: number[]; metadata: Record<string, unknown> }): Promise<unknown>
+  upsertItem?(item: { id: string; vector: number[]; metadata: Record<string, unknown> }): Promise<unknown>
   // vectra >=0.15 inserts a BM25 `query` string between vector and topK; pass "" for pure vector search.
   queryItems(vector: number[], query: string, topK: number): Promise<Array<{ score: number; item: { metadata: Record<string, unknown> } }>>
   listItems(): Promise<Array<{ id: string; metadata: Record<string, unknown> }>>;
@@ -59,10 +60,12 @@ function createVectraStore(dirPath: string): VectorStore {
       await idx.beginUpdate()
       try {
         for (const doc of docs) {
-          await idx.insertItem({
-            vector: doc.vector,
-            metadata: { _id: doc.id, ...doc.metadata },
-          })
+          const metadata = { ...doc.metadata, _id: doc.id }
+          if (idx.upsertItem) {
+            await idx.upsertItem({ id: doc.id, vector: doc.vector, metadata })
+          } else {
+            await idx.insertItem({ vector: doc.vector, metadata })
+          }
         }
         await idx.endUpdate()
       } catch (err) {
@@ -113,7 +116,7 @@ export function fileVectorMemory(config: FileVectorMemoryConfig): VectorMemory {
       await store.upsert(docs.map(doc => ({
         id: doc.id,
         vector: doc.embedding,
-        metadata: { content: doc.content, ...doc.metadata },
+        metadata: { ...doc.metadata, content: doc.content },
       })))
     },
     async search(embedding, options) {

packages/memory/src/forget.ts

@@ -48,6 +48,15 @@ export interface ForgetSubjectResult {
   evidenceHash: string
 }
 
+function failedReport(backend: string, reason: string): ForgetReport {
+  return {
+    backend,
+    deletedCount: 0,
+    at: new Date().toISOString(),
+    failures: [{ id: '*', reason }],
+  }
+}
+
 function isForgettable(value: unknown): value is ForgettableMemory {
   return (
     !!value &&
@@ -86,7 +95,11 @@ export async function forgetSubject(
         : 'unknown')
       continue
     }
-    reports.push(await memory.forgetSubject(subjectId))
+    try {
+      reports.push(await memory.forgetSubject(subjectId))
+    } catch (err) {
+      reports.push(failedReport(memory.__agentskitBackend, err instanceof Error ? err.message : String(err)))
+    }
   }
   const totalDeleted = reports.reduce((sum, r) => sum + r.deletedCount, 0)
   const evidenceHash = await hash(
@@ -120,8 +133,14 @@ export function makeForgettable<M extends object>(
   return Object.assign(memory, {
     __agentskitBackend: options.backend,
     forgetSubject: async (subjectId: string): Promise<ForgetReport> => {
-      const ids = await options.listIds(subjectId)
       const failures: Array<{ id: string; reason: string }> = []
+      let ids: string[]
+      try {
+        ids = await options.listIds(subjectId)
+      } catch (err) {
+        const reason = err instanceof Error ? err.message : String(err)
+        return failedReport(options.backend, reason)
+      }
       try {
         await options.deleteIds(ids)
       } catch (err) {

packages/memory/src/kv-store-basic.ts

@@ -27,6 +27,32 @@ const enqueueFileWrite = (path: string, task: () => Promise<void>): Promise<void
   return next
 }
 
+const writeFileAtomically = async (path: string, contents: string): Promise<void> => {
+  await mkdir(dirname(path), { recursive: true })
+  const tempPath = join(dirname(path), `.${basename(path)}.${randomBytes(8).toString('hex')}.tmp`)
+  try {
+    await writeFile(tempPath, contents, { encoding: 'utf8', mode: 0o600 })
+    await rename(tempPath, path)
+  } finally {
+    await unlink(tempPath).catch(() => {})
+  }
+}
+
+const removeExpiredFileEntry = (
+  path: string,
+  key: string,
+  ttlSeconds: number | undefined,
+  load: () => Promise<Map<string, KvEntry>>,
+  persist: (map: Map<string, KvEntry>) => Promise<void>,
+): Promise<void> => enqueueFileWrite(path, async () => {
+  const map = await load()
+  const entry = map.get(key)
+  if (entry && isExpired(entry, ttlSeconds, Date.now())) {
+    map.delete(key)
+    await persist(map)
+  }
+})
+
 /** Creates an in-memory key-value store with optional TTL and size limits.
  * @param config Backend and retention settings.
  * @returns A key-value store backed by a process-local map.
@@ -35,20 +61,19 @@ const enqueueFileWrite = (path: string, task: () => Promise<void>): Promise<void
 export const createInMemoryStore = (config: InMemoryKvConfig): AgentskitMemoryStore => {
   validateKvRetention(config)
   const store = new Map<string, KvEntry>()
-  const now = () => Date.now()
   return {
     id: 'in-memory',
     async get(key) {
       const entry = store.get(key)
       if (!entry) return undefined
-      if (isExpired(entry, config.ttlSeconds, now())) {
+      if (isExpired(entry, config.ttlSeconds, Date.now())) {
         store.delete(key)
         return undefined
       }
       return entry.value
     },
     async set(key, value) {
-      store.set(key, { value, insertedAt: now() })
+      store.set(key, { value, insertedAt: Date.now() })
       enforceMaxMessages(store, config.maxMessages)
     },
   }
@@ -72,14 +97,7 @@ export const createFileStore = (config: FileKvConfig): AgentskitMemoryStore => {
   }
 
   const persist = async (map: Map<string, KvEntry>): Promise<void> => {
-    await mkdir(dirname(path), { recursive: true })
-    const tempPath = join(dirname(path), `.${basename(path)}.${randomBytes(8).toString('hex')}.tmp`)
-    try {
-      await writeFile(tempPath, JSON.stringify(Object.fromEntries(map), null, 2), { encoding: 'utf8', mode: 0o600 })
-      await rename(tempPath, path)
-    } finally {
-      try { await unlink(tempPath) } catch { /* rename already published it */ }
-    }
+    await writeFileAtomically(path, JSON.stringify(Object.fromEntries(map), null, 2))
   }
 
   return {
@@ -89,14 +107,7 @@ export const createFileStore = (config: FileKvConfig): AgentskitMemoryStore => {
       const entry = map.get(key)
       if (!entry) return undefined
       if (isExpired(entry, config.ttlSeconds, Date.now())) {
-        await enqueueFileWrite(path, async () => {
-          const latest = await load()
-          const current = latest.get(key)
-          if (current && isExpired(current, config.ttlSeconds, Date.now())) {
-            latest.delete(key)
-            await persist(latest)
-          }
-        })
+        await removeExpiredFileEntry(path, key, config.ttlSeconds, load, persist)
         return undefined
       }
       return entry.value
@@ -151,49 +162,38 @@ export const createLocalStorageStore = ({
     }
   }
 
-  const load = async (): Promise<Map<string, KvEntry>> =>
-    storage ? mapFromJson(storage.getItem(key)) : loadFromFile()
-
+  const loadSync = (): Map<string, KvEntry> => mapFromJson(storage!.getItem(key))
   const persist = async (map: Map<string, KvEntry>): Promise<void> => {
-    const raw = JSON.stringify(Object.fromEntries(map), null, 2)
-    if (storage) {
-      storage.setItem(key, raw)
-      return
-    }
-    await mkdir(dirname(filePath), { recursive: true })
-    await writeFile(filePath, raw, { encoding: 'utf8', mode: 0o600 })
+    await writeFileAtomically(filePath, JSON.stringify(Object.fromEntries(map), null, 2))
+  }
+
+  const persistSync = (map: Map<string, KvEntry>): void => {
+    storage!.setItem(key, JSON.stringify(Object.fromEntries(map), null, 2))
   }
 
   return {
     id: storage ? `localstorage:${key}` : `localstorage-file:${filePath}:${key}`,
     async get(itemKey) {
-      const map = await load()
+      const map = storage ? loadSync() : await loadFromFile()
       const entry = map.get(itemKey)
       if (!entry) return undefined
       if (isExpired(entry, config.ttlSeconds, Date.now())) {
         if (storage) {
           map.delete(itemKey)
-          await persist(map)
+          persistSync(map)
         } else {
-          await enqueueFileWrite(filePath, async () => {
-            const latest = await loadFromFile()
-            const current = latest.get(itemKey)
-            if (current && isExpired(current, config.ttlSeconds, Date.now())) {
-              latest.delete(itemKey)
-              await persist(latest)
-            }
-          })
+          await removeExpiredFileEntry(filePath, itemKey, config.ttlSeconds, loadFromFile, persist)
         }
         return undefined
       }
       return entry.value
     },
     async set(itemKey, value) {
       if (storage) {
-        const map = await load()
+        const map = loadSync()
         map.set(itemKey, { value, insertedAt: Date.now() })
         enforceMaxMessages(map, config.maxMessages)
-        await persist(map)
+        persistSync(map)
         return
       }
       await enqueueFileWrite(filePath, async () => {

packages/memory/src/kv-store-sqlite.ts

@@ -51,6 +51,14 @@ export const createSqliteStore = ({ config, open }: CreateSqliteStoreOpts): Agen
     }
   }
 
+  const write = db.transaction ? db.transaction((key: string, value: string, now: number) => {
+    setStmt.run(key, value, now)
+    enforce()
+  }) : (key: string, value: string, now: number): void => {
+    setStmt.run(key, value, now)
+    enforce()
+  }
+
   return {
     id: `sqlite:${config.path}`,
     async get(key) {
@@ -63,8 +71,7 @@ export const createSqliteStore = ({ config, open }: CreateSqliteStoreOpts): Agen
       return JSON.parse(row.value) as unknown
     },
     async set(key, value) {
-      setStmt.run(key, JSON.stringify(value), Date.now())
-      enforce()
+      write(key, JSON.stringify(value), Date.now())
     },
   }
 }

packages/memory/src/kv-store-types.ts

@@ -20,8 +20,7 @@ export interface KvEntry {
 }
 
 export const isExpired = (entry: KvEntry, ttlSeconds: number | undefined, now: number): boolean => {
-  if (ttlSeconds === undefined) return false
-  return now - entry.insertedAt > ttlSeconds * 1000
+  return ttlSeconds !== undefined && now - entry.insertedAt > ttlSeconds * 1000
 }
 
 export const enforceMaxMessages = (map: Map<string, KvEntry>, maxMessages: number | undefined): void => {
@@ -115,6 +114,7 @@ export interface SqliteStmt {
 export interface SqliteLike {
   exec(sql: string): void
   prepare(sql: string): SqliteStmt
+  transaction?<T extends (...args: never[]) => unknown>(fn: T): T
 }
 
 /** Opens a SQLite database at the given path. */

packages/memory/src/kv-store-vector.ts

@@ -30,6 +30,11 @@ export const createVectorStore = ({
 }: CreateVectorStoreOpts): AgentskitMemoryStore & { recall(query: string, k?: number): Promise<readonly unknown[]> } => {
   validateKvRetention(config)
   const collection = config.collection
+  const isLive = (metadata: Record<string, unknown>, now: number) => {
+    const insertedAt = metadata['__insertedAt']
+    return metadata['__collection'] === collection &&
+      (typeof insertedAt !== 'number' || !isExpired({ value: undefined, insertedAt }, config.ttlSeconds, now))
+  }
 
   const embedOne = async (text: string): Promise<number[]> => {
     const [vec] = await embedder.embed([text])
@@ -46,15 +51,12 @@ export const createVectorStore = ({
     id: `vector:${config.provider}:${collection}`,
     async get(key) {
       const vec = await embedOne(key)
-      const hits = await vectorStore.query(vec, 1, { __collection: collection, __key: key })
-      const hit = hits[0]
-      if (hit === undefined) return undefined
-      if (hit.metadata['__key'] !== key) return undefined
-      const insertedAt = hit.metadata['__insertedAt']
-      if (typeof insertedAt === 'number' && isExpired({ value: undefined, insertedAt }, config.ttlSeconds, Date.now())) {
-        return undefined
-      }
-      return hit.metadata['__value']
+      const hits = await vectorStore.query(vec, config.ttlSeconds === undefined ? 1 : 100, { __collection: collection, __key: key })
+      const now = Date.now()
+      const hit = hits.find(candidate => {
+        return candidate.metadata['__key'] === key && isLive(candidate.metadata, now)
+      })
+      return hit?.metadata['__value']
     },
     async set(key, value) {
       const vec = await embedOne(key)
@@ -67,16 +69,15 @@ export const createVectorStore = ({
       ])
     },
     async recall(query, k = 5) {
+      if (k <= 0) return []
       const vec = await embedOne(query)
-      const hits = await vectorStore.query(vec, k, { __collection: collection })
+      const hits = await vectorStore.query(vec, k + (config.ttlSeconds === undefined ? 0 : 100), { __collection: collection })
       const now = Date.now()
       const results: unknown[] = []
       for (const hit of hits) {
-        const insertedAt = hit.metadata['__insertedAt']
-        if (typeof insertedAt === 'number' && isExpired({ value: undefined, insertedAt }, config.ttlSeconds, now)) {
-          continue
-        }
+        if (!isLive(hit.metadata, now)) continue
         results.push(hit.metadata['__value'])
+        if (results.length >= k) break
       }
       return results
     },

packages/memory/src/redis-chat.ts

@@ -37,7 +37,12 @@ export function redisChatMemory(config: RedisChatMemoryConfig): ChatMemory {
 
   const getClient = (): Promise<RedisClientAdapter> => {
     if (config.client) return Promise.resolve(config.client)
-    if (!clientPromise) clientPromise = createRedisClientAdapter(config.url)
+    if (!clientPromise) {
+      clientPromise = createRedisClientAdapter(config.url).catch(error => {
+        clientPromise = null
+        throw error
+      })
+    }
     return clientPromise
   }