diff --git a/docs/architecture/agent-memory-system/spec.md b/docs/architecture/agent-memory-system/spec.md index 6ca5b3fcd..5d9a682dd 100644 --- a/docs/architecture/agent-memory-system/spec.md +++ b/docs/architecture/agent-memory-system/spec.md @@ -352,7 +352,9 @@ the model could not understand. A `memory/extract` anchor is written only when a ## 8. Retrieval and scoring `retrieve()` resolves the per-agent retrieval config, fetches `topK*2` candidates from each path, fuses them, -and trims to `topK`. +and trims to `topK`. `searchMemories()` is the only caller allowed to pass a search-only `topKOverride` +(default 50, route max 100); agent-facing recall and prompt injection continue to use the configured +retrieval `topK` and still record access only for recall/injection hits. - **Keyword path.** `agent_memory_fts` (FTS5, BM25-ranked) with the tokenizer chosen at runtime — `trigram` for CJK/substring matching, else `unicode61`. The query is tokenized on whitespace; each term is quoted @@ -368,6 +370,11 @@ and trims to `topK`. and the same row-class exclusions as the keyword path (persona, working, archived, conflicted, superseded) are applied to the matches. If any embedded row's identity is stale (model/dim/fingerprint mismatch), the turn answers **FTS-only** and a non-destructive background re-embed is queued — rows are never deleted. + Query embeddings are tracked per `agentId::embeddingFingerprint` group and then by full normalized query: + identical concurrent queries share one provider call, two distinct fresh queries may run concurrently, and + a third distinct query degrades that turn to FTS-only. The 800 ms soft timeout and 30 s stale replacement + window still apply per caller. Vector matches rejected by the SQLite liveness filter are queued for + fire-and-forget deletion from the current vector store so dead vectors stop occupying candidate slots. **RRF fusion (`fuse`).** Each path contributes `1/(rrfK + rank + 1)` per item, accumulated when a memory appears in both lists. The final order is: @@ -439,16 +446,28 @@ value is absent, so the cooldown survives restarts. Within the cooldown only che no audit row). A missing-model pass is recorded `skipped` and does **not** advance the cooldown (it retries next trigger). -**`mergeNearDuplicates` (budgeted).** Iterates active non-persona rows oldest-first, bounded by **8 LLM -calls** and **24k input tokens** per pass (remainder deferred). For each row it picks the first neighbor with -similarity ≥ **0.85** and runs the same decision prompt; only `UPDATE`/`SUPERSEDE` fold the pair (the -more-recent row normally survives — unless the merged content's provenance key is already owned by a third -row, in which case the pair folds into that owner; importance and confidence only ever rise), so a re-run -converges. +**`mergeNearDuplicates` (budgeted).** Scans only active `embedded` non-persona/non-working rows that match +the current embedding fingerprint/dimension. It uses the row's stored DuckDB vector via +`queryByMemoryId()` rather than re-embedding row content through the provider; the store method reads the +source vector by id and reuses the existing parameterized vector query path, then filters the source id out +of the results. The pass advances an in-memory `{ createdAt, id }` compound cursor so large same-timestamp +windows are not skipped. Each pass is bounded by +**64 stored-vector neighbor scans**, **8 LLM calls**, and **24k input tokens** (remainder deferred). For each +row it picks the first current, live neighbor with similarity ≥ **0.85** and runs the same decision prompt; +only `UPDATE`/`SUPERSEDE` fold the pair (the more-recent row normally survives — unless the merged content's +provenance key is already owned by a third row, in which case the pair folds into that owner; importance and +confidence only ever rise), so a re-run converges. **Non-destructive archival (`archiveStale`).** A row is soft-deleted only when **all** of: `decayScore < -0.05`, `access_count == 0`, age `> 90 days`, still active. Anchors and persona are exempt. `restoreMemory` -reverses it. There is no hard delete on this path. +0.05`, `access_count == 0`, age `> 90 days`, still active. Anchors, persona, and working rows are exempt. +`restoreMemory` reverses it for normal memories. There is no hard delete on this path. Maintenance also runs +a cheap repair that normalizes any legacy persona/working row back to `status='fts_only'` while preserving +embedding refs until a bounded dead-vector sweep deletes the corresponding DuckDB vectors from a successfully +opened current sidecar. The sweep runs only when the current vector store is already warm/ready, filters refs +by both current embedding fingerprint and current ready dimension before applying its batch limit, re-checks +row lifecycle plus dim/model before vector delete, and clears SQLite refs only if the row is still prunable +with the same dim/model after deletion. Old-fingerprint or old-dimension refs remain as traceable metadata +residue; maintenance does not cold-open those sidecars or let them occupy the current cleanup batch. --- @@ -539,12 +558,14 @@ This is where the "stabilization" and "kernel hardening" work concentrates. full file reset (`destroyFile` + recreate), not an in-place migration. - **Transactional vector upsert.** `upsert` wraps delete-then-insert in `BEGIN/COMMIT` with `ROLLBACK` on error, so there is no "deleted-but-not-inserted" hole. -- **Archive / forget / restore lifecycle.** Agent-facing `forgetMemory` is a soft archive: it marks the row - `archived`, leaves any existing vector in place, and relies on recall's status filters plus SQLite - re-checks to keep archived facts out of results. It only requires a managed agent, so users can forget - while memory is disabled. `restoreMemory` re-marks the row `pending_embedding` and re-embeds it, and remains - gated by `canWriteAgentMemory` (managed agent · memory enabled · not disposed). Permanent UI delete - (`deleteMemory`) hard-deletes the row and best-effort deletes its vector. +- **Archive / forget / restore lifecycle.** Agent-facing `forgetMemory` is a soft archive: it marks normal + rows `archived` and leaves recall correctness to status filters plus SQLite re-checks while dead-vector + pruning removes obsolete sidecar entries over time. It only requires a managed agent, so users can forget + while memory is disabled. `restoreMemory` re-marks a normal archived row `pending_embedding` and re-embeds + it, and remains gated by `canWriteAgentMemory` (managed agent · memory enabled · not disposed). Generic + archive/forget/restore refuse `persona` and `working` rows; persona lifecycle is controlled by + persona-specific routes and working rows are kernel-owned. Permanent UI delete (`deleteMemory`) hard-deletes + the row and best-effort deletes its vector. - **Embedding-drain config guard.** A background embedding drain captures the embedding identity it started with; before writing vectors, and before a reindex reset, it re-checks the agent's current `memoryEmbedding` fingerprint and discards the batch if the config changed mid-flight, so a stale drain can never write @@ -593,8 +614,9 @@ inspect `result.ok`, not `isError`. Hard infra failures throw. `memory.approvePersonaDraft`, `memory.rejectPersonaDraft`, `memory.setPersonaAnchor`, `memory.listAuditEvents`, `memory.listViewManifests`. -- `memory.search` is read-only: it caps the result count but cannot widen the agent's configured `topK`, and - does not bump `access_count`. +- `memory.search` is read-only: it uses a search-only depth override (default 50, route max 100) so the + Memory Manager can return more than the agent's recall `topK`, excludes persona/working rows at the SQL + search layer before applying result limits, and it does not bump `access_count`. - `memory.add` accepts optional `category`, runs the decision ring, and writes a `memory/add` user audit row. - `memory.getSourceSpan` resolves a memory's `source_entry_ids` to readable role/content via the effective tape view (powers the lineage UI). @@ -699,8 +721,9 @@ Coverage mirrors source under `test/main/**` (and `test/renderer/**` for UI), pi nDCG). The real-DB FTS5/trigram (CJK) eval runs only when native `better-sqlite3` is loadable (force with -`DEEPCHAT_REQUIRE_NATIVE_SQLITE=1`); it is not wired to a CI Action. Run before merge: `pnpm run typecheck`, -`pnpm test`, `pnpm run lint`, `pnpm run format:check`. +`DEEPCHAT_REQUIRE_NATIVE_SQLITE=1`); it is not wired to a CI Action. Run before merge through mise: +`mise exec -- pnpm run typecheck`, `mise exec -- pnpm test`, `mise exec -- pnpm run lint`, +`mise exec -- pnpm run format:check`. --- @@ -716,10 +739,12 @@ The real-DB FTS5/trigram (CJK) eval runs only when native `better-sqlite3` is lo category. - **Memory-management skill is opt-in.** The bundled skill is discoverable and has no `allowedTools`, but it is not auto-pinned into every conversation to avoid permanent prompt cost. -- **DuckDB disk reclaim.** Per-memory hard delete does not shrink the file; only a whole-store reset reclaims - (no `VACUUM`), so the file grows between resets. Permanent `deleteMemory` removes the row's vector from the - cached/opened store under the per-agent lock; `forgetMemory` is a soft archive and may leave an orphan - vector, which is harmless because recall excludes archived rows and re-checks the authoritative SQLite row. +- **DuckDB disk reclaim.** Per-memory hard delete and dead-vector pruning remove vector rows but do not shrink + the DuckDB file; only a whole-store reset reclaims disk space (no `VACUUM`), so the file can still grow + between resets. Archived/superseded vectors are correctness-safe because recall re-checks SQLite, but they + are not quality-neutral: if left in the HNSW result window they crowd out live candidates. Inline prune and + the warm-store-only, fingerprint+dimension-aware, lifecycle-guarded maintenance sweep remove current-sidecar + dead vectors over time. Management search disables inline prune so read-only search does not write DuckDB. - **FTS5 native dependency.** Under vitest with an unloadable native ABI, the real FTS5/trigram eval skips; CI needs a working native build to exercise it. - **Vector query threshold.** `MemoryVectorStore.query` does not apply a distance cutoff itself; the @@ -754,6 +779,10 @@ The real-DB FTS5/trigram (CJK) eval runs only when native `better-sqlite3` is lo | `CONSOLIDATION_COOLDOWN_MS` | 6h | LLM-backed pass cooldown (restart-durable) | | `CONSOLIDATION_MAX_LLM_CALLS` / tokens | 8 / 24000 | `mergeNearDuplicates` budget | | `CONSOLIDATION_MERGE_SIMILARITY` | 0.85 | near-duplicate merge threshold | +| `CONSOLIDATION_MAX_NEIGHBOR_SCANS` | 64 | stored-vector neighbor scans per consolidation pass | +| `VECTOR_PRUNE_BATCH_LIMIT` | 256 | prunable archived/superseded/internal-kind vector refs per maintenance pass | +| `RECALL_QUERY_EMBEDDING_TIMEOUT_MS` / stale / max concurrent | 800ms / 30s / 2 | foreground query-embedding soft timeout and per-agent+model cap | +| `MEMORY_SEARCH_DEFAULT_LIMIT` | 50 | default management search depth | | `WORKING_BLOB_TOKEN_LIMIT` | 400 | working-memory blob size | | persona thresholds | ≥ 3 memories · importance ≥ 5.0 · changeRatio > 0.6 → needsReview | guarded persona evolution | | reflection thresholds | ≥ 3 memories · importance ≥ 5.0 · reflection importance 0.8 | offline reflection | diff --git a/docs/issues/memory-recall-hot-path-keyword-recall/spec.md b/docs/issues/memory-recall-hot-path-keyword-recall/spec.md index 1cc0d927f..52ada5e41 100644 --- a/docs/issues/memory-recall-hot-path-keyword-recall/spec.md +++ b/docs/issues/memory-recall-hot-path-keyword-recall/spec.md @@ -19,7 +19,9 @@ Memory recall must not add unbounded first-token latency, and FTS-only recall mu - Recall keyword selection is corpus-aware: query terms are extracted without a static stopword list, terms with no active corpus hits are dropped, and high-frequency terms are filtered when better lower-frequency terms exist. - Mixed ASCII/code/CJK query term extraction preserves original query order before applying the candidate cap, so earlier CJK terms cannot be starved by later ASCII/code tokens. - Corpus-aware term stats are collected with one bounded aggregate query per recall, not one query per candidate term. -- At most one tracked warm query-embedding entry per agent/model is active; later turns skip vector recall while that entry is fresh, and stale entries older than 30 seconds can be replaced without aborting old provider requests. +- Superseded by `docs/issues/memory-audit-hardening/` F5: warm query embeddings are now tracked per + agent/model group and full normalized query, with identical-query sharing and at most two fresh + distinct entries per group. The 30-second stale replacement behavior remains. - `memory.search` management search keeps the existing all-term semantics. - Access counter updates happen in one repository call for a recalled result set. - Settings explain the degraded FTS-only state when memory is enabled without an embedding model and warn when extraction falls back to the chat model. @@ -38,7 +40,9 @@ Memory recall must not add unbounded first-token latency, and FTS-only recall mu - Do not implement #19 query expansion or #20 reranker. - Do not implement #21 policy port or deduplicate `deriveRecall`. - Do not maintain a static recall stopword list. -- Do not add DuckDB vacuum/orphan vector cleanup. +- Do not add DuckDB `VACUUM`. The earlier "no orphan vector cleanup" non-goal is superseded by + `docs/issues/memory-audit-hardening/` F4, which adds inline and maintenance dead-vector pruning + without disk-space compaction. - Do not auto-select models or change memory defaults. ## Open Questions diff --git a/src/main/presenter/memoryPresenter/index.ts b/src/main/presenter/memoryPresenter/index.ts index 6900ee061..667dfc64c 100644 --- a/src/main/presenter/memoryPresenter/index.ts +++ b/src/main/presenter/memoryPresenter/index.ts @@ -83,7 +83,14 @@ export class MemoryPresenter implements MemoryRuntimePort { this.embedding.warmEmbeddingConnection(agentId, embedding), reindexEmbeddings: (agentId, force) => this.reindexEmbeddings(agentId, force), backfillEmbeddings: (agentId) => this.backfillEmbeddings(agentId), - isReindexing: (agentId) => this.embedding.isReindexing(agentId) + isReindexing: (agentId) => this.embedding.isReindexing(agentId), + deletePrunableVectorsForMemoryIds: (agentId, embedding, dimensions, memoryIds) => + this.vectorStore.deletePrunableVectorsForMemoryIds( + agentId, + embedding, + dimensions, + memoryIds + ) }) this.reflection = new ReflectionService(this.runtime, { @@ -104,8 +111,17 @@ export class MemoryPresenter implements MemoryRuntimePort { }) maintenanceService = new MaintenanceService(this.runtime, this.rows, { - retrieve: (agentId, query, now, recordAccessHits) => - this.retrieval.retrieve(agentId, query, now, recordAccessHits), + queryNeighborsByMemoryId: (agentId, embedding, dimensions, memoryId, topK) => + this.vectorStore.queryNeighborsByMemoryId(agentId, embedding, dimensions, memoryId, topK), + getWarmVectorStoreDimension: (agentId, embedding) => + this.vectorStore.getWarmVectorStoreDimension(agentId, embedding), + deletePrunableVectorsForMemoryIds: (agentId, embedding, dimensions, memoryIds) => + this.vectorStore.deletePrunableVectorsForMemoryIds( + agentId, + embedding, + dimensions, + memoryIds + ), syncWorkingMemoryAfterMutation: (agentId) => this.workingMemory.syncWorkingMemoryAfterMutation(agentId), triggerEmbedding: (agentId) => this.embedding.processPendingEmbeddings(agentId), diff --git a/src/main/presenter/memoryPresenter/infra/memoryVectorStore.ts b/src/main/presenter/memoryPresenter/infra/memoryVectorStore.ts index 7e7345eef..c5a4b8e68 100644 --- a/src/main/presenter/memoryPresenter/infra/memoryVectorStore.ts +++ b/src/main/presenter/memoryPresenter/infra/memoryVectorStore.ts @@ -274,6 +274,14 @@ export class MemoryVectorStore implements IMemoryVectorStore { } } + private distanceFunction(): string { + return this.metric === 'ip' + ? 'array_negative_inner_product' + : this.metric === 'cosine' + ? 'array_cosine_distance' + : 'array_distance' + } + async upsert(records: MemoryVectorRecord[]): Promise { if (!records.length) return await this.connection.run('BEGIN TRANSACTION;') @@ -299,12 +307,7 @@ export class MemoryVectorStore implements IMemoryVectorStore { embedding: number[], options: MemoryVectorQueryOptions ): Promise { - const fn = - this.metric === 'ip' - ? 'array_negative_inner_product' - : this.metric === 'cosine' - ? 'array_cosine_distance' - : 'array_distance' + const fn = this.distanceFunction() const sql = ` SELECT memory_id, ${fn}(embedding, ?) AS distance FROM ${this.vectorTable} @@ -322,6 +325,22 @@ export class MemoryVectorStore implements IMemoryVectorStore { })) } + async queryByMemoryId( + memoryId: string, + options: MemoryVectorQueryOptions + ): Promise { + const reader = await this.connection.runAndReadAll( + `SELECT embedding FROM ${this.vectorTable} WHERE memory_id = ? LIMIT 1;`, + [memoryId] + ) + const source = reader.getRowObjectsJson()[0]?.embedding + if (!Array.isArray(source)) return [] + const embedding = source.map(Number).filter((value) => Number.isFinite(value)) + if (embedding.length !== source.length || embedding.length === 0) return [] + const matches = await this.query(embedding, { topK: options.topK + 1 }) + return matches.filter((match) => match.memoryId !== memoryId).slice(0, options.topK) + } + async deleteByMemoryIds(memoryIds: string[]): Promise { if (!memoryIds.length) return const placeholders = memoryIds.map(() => '?').join(', ') diff --git a/src/main/presenter/memoryPresenter/infra/vectorStoreManager.ts b/src/main/presenter/memoryPresenter/infra/vectorStoreManager.ts index ebcee7437..1a71d0a47 100644 --- a/src/main/presenter/memoryPresenter/infra/vectorStoreManager.ts +++ b/src/main/presenter/memoryPresenter/infra/vectorStoreManager.ts @@ -1,6 +1,6 @@ import logger from '@shared/logger' -import type { IMemoryVectorStore } from '../types' +import type { IMemoryVectorStore, MemoryVectorMatch } from '../types' import { embeddingFingerprint, type MemoryModelRef, type MemoryRuntimeContext } from '../context' import type { VectorStoreRetrievalPort } from '../ports' @@ -40,6 +40,17 @@ export class VectorStoreManager implements VectorStoreRetrievalPort { return readyIdentity.startsWith(`${this.warmupKey(agentId, embedding)}::`) } + getWarmVectorStoreDimension(agentId: string, embedding: MemoryModelRef): number | null { + const readyIdentity = this.vectorStoreReady.get(agentId) + if (!readyIdentity) return null + if (this.vectorStoreIdentities.get(agentId) !== readyIdentity) return null + if (!this.vectorStores.has(agentId)) return null + const prefix = `${this.warmupKey(agentId, embedding)}::` + if (!readyIdentity.startsWith(prefix)) return null + const dimensions = Number(readyIdentity.slice(prefix.length)) + return Number.isFinite(dimensions) && dimensions > 0 ? Math.floor(dimensions) : null + } + markReady(agentId: string, embedding: MemoryModelRef, dimensions: number): void { this.vectorStoreReady.set(agentId, this.cacheKey(agentId, embedding, dimensions)) } @@ -159,6 +170,65 @@ export class VectorStoreManager implements VectorStoreRetrievalPort { }) } + async deletePrunableVectorsForMemoryIds( + agentId: string, + embedding: MemoryModelRef, + dimensions: number, + memoryIds: string[] + ): Promise { + if (!memoryIds.length) return [] + return this.withAgentLock(agentId, async (locked) => { + if (this.ctx.isDisposed) return [] + const store = await locked.open(embedding, dimensions) + if (!store.isUsable()) { + this.clearReady(agentId) + return [] + } + const fingerprint = embeddingFingerprint(embedding.providerId, embedding.modelId) + const prunableIds = this.ctx.deps.repository.filterPrunableVectorRefs( + agentId, + memoryIds, + dimensions, + fingerprint + ) + if (!prunableIds.length) return [] + try { + await store.deleteByMemoryIds(prunableIds) + return prunableIds + } catch (error) { + logger.warn(`[Memory] vector prune failed: ${String(error)}`) + return [] + } + }) + } + + async deleteVectorsForMemoryIdsForEmbedding( + agentId: string, + embedding: MemoryModelRef, + dimensions: number, + memoryIds: string[] + ): Promise { + return this.deletePrunableVectorsForMemoryIds(agentId, embedding, dimensions, memoryIds) + } + + async queryNeighborsByMemoryId( + agentId: string, + embedding: MemoryModelRef, + dimensions: number, + memoryId: string, + topK: number + ): Promise { + return this.withAgentLock(agentId, async (locked) => { + if (this.ctx.isDisposed) return [] + const store = await locked.open(embedding, dimensions) + if (!store.isUsable()) { + this.clearReady(agentId) + return [] + } + return store.queryByMemoryId(memoryId, { topK }) + }) + } + getLockInFlight(): Promise[] { return [...this.vectorStoreLocks.values()] } diff --git a/src/main/presenter/memoryPresenter/runtimeConstants.ts b/src/main/presenter/memoryPresenter/runtimeConstants.ts index f28e14b9e..bda8d2a28 100644 --- a/src/main/presenter/memoryPresenter/runtimeConstants.ts +++ b/src/main/presenter/memoryPresenter/runtimeConstants.ts @@ -22,6 +22,8 @@ export const CONSOLIDATION_COOLDOWN_MS = 6 * 60 * 60 * 1000 export const CONSOLIDATION_MAX_LLM_CALLS = 8 export const CONSOLIDATION_MAX_INPUT_TOKENS = 24000 export const CONSOLIDATION_MERGE_SIMILARITY = 0.85 +export const CONSOLIDATION_MAX_NEIGHBOR_SCANS = 64 +export const VECTOR_PRUNE_BATCH_LIMIT = 256 export const MAINTENANCE_START_DELAY_MS = 60 * 1000 export const STARTUP_PREWARM_DELAY_MS = 3 * 1000 @@ -38,3 +40,5 @@ export const MEMORY_CREATED_IDS_EVENT_LIMIT = 50 export const RECALL_QUERY_EMBEDDING_TIMEOUT_MS = 800 export const RECALL_QUERY_EMBEDDING_STALE_MS = 30 * 1000 +export const RECALL_QUERY_EMBEDDING_MAX_CONCURRENT = 2 +export const MEMORY_SEARCH_DEFAULT_LIMIT = 50 diff --git a/src/main/presenter/memoryPresenter/services/maintenanceService.ts b/src/main/presenter/memoryPresenter/services/maintenanceService.ts index 9bfd9f4fd..19b658ab8 100644 --- a/src/main/presenter/memoryPresenter/services/maintenanceService.ts +++ b/src/main/presenter/memoryPresenter/services/maintenanceService.ts @@ -2,6 +2,7 @@ import logger from '@shared/logger' import { isAgentMemoryCategory } from '@shared/types/agent-memory' import { ARCHIVE_AGE_MS, ARCHIVE_DECAY_THRESHOLD } from '../core/lifecycle' +import { decayScore, distanceToSimilarity } from '../core/scoring' import { ADD_DECISION, buildDecisionPrompt, @@ -10,27 +11,35 @@ import { } from '../core/decision' import { normalizeMemoryCandidate } from '../core/candidates' import { estimateTokens } from '../core/injectionPort' -import { decayScore } from '../core/scoring' import { CONSOLIDATION_COOLDOWN_MS, CONSOLIDATION_IDLE_MS, CONSOLIDATION_MAX_INPUT_TOKENS, CONSOLIDATION_MAX_LLM_CALLS, + CONSOLIDATION_MAX_NEIGHBOR_SCANS, CONSOLIDATION_MERGE_SIMILARITY, + DECISION_NEIGHBOR_TOP_S, MAINTENANCE_START_DELAY_MS, STARTUP_ARM_STAGGER_MS, STARTUP_PREWARM_DELAY_MS, - STARTUP_PREWARM_STAGGER_MS + STARTUP_PREWARM_STAGGER_MS, + VECTOR_PRUNE_BATCH_LIMIT } from '../runtimeConstants' -import { isSafeAgentId, type MemoryRecallItem } from '../types' +import { + isSafeAgentId, + type AgentMemoryRow, + type ConsolidationScanCursor, + type MemoryPersonaDraftResult, + type MemoryReflectionResult +} from '../types' import type { MemoryRowMutations } from './rowMutations' -import { type MemoryModelRef, type MemoryRuntimeContext } from '../context' -import type { MemoryPersonaDraftResult, MemoryReflectionResult } from '../types' +import { embeddingFingerprint, type MemoryModelRef, type MemoryRuntimeContext } from '../context' export class MaintenanceService { private readonly consolidationTimers = new Map() private readonly consolidationTimerDueAt = new Map() private readonly lastConsolidationAt = new Map() + private readonly consolidationScanCursor = new Map() private readonly consolidationRuns = new Set>() private maintenanceStartTimer: NodeJS.Timeout | null = null private prewarmStartTimer: NodeJS.Timeout | null = null @@ -41,12 +50,20 @@ export class MaintenanceService { private readonly ctx: MemoryRuntimeContext, private readonly rows: MemoryRowMutations, private readonly ports: { - retrieve: ( + queryNeighborsByMemoryId: ( + agentId: string, + embedding: MemoryModelRef, + dimensions: number, + memoryId: string, + topK: number + ) => Promise> + getWarmVectorStoreDimension: (agentId: string, embedding: MemoryModelRef) => number | null + deletePrunableVectorsForMemoryIds: ( agentId: string, - query: string, - now: number, - recordAccessHits: boolean - ) => Promise + embedding: MemoryModelRef, + dimensions: number, + memoryIds: string[] + ) => Promise syncWorkingMemoryAfterMutation: (agentId: string) => void triggerEmbedding: (agentId: string) => Promise warmVectorStore: (agentId: string, embedding: MemoryModelRef) => Promise @@ -100,6 +117,7 @@ export class MaintenanceService { this.consolidationTimers.clear() this.consolidationTimerDueAt.clear() this.lastConsolidationAt.clear() + this.consolidationScanCursor.clear() } private shouldArmMaintenance(agentId: string): boolean { @@ -228,13 +246,14 @@ export class MaintenanceService { this.lastConsolidationAt.set(agentId, last) } if (now - last < CONSOLIDATION_COOLDOWN_MS) { - this.runCheapMaintenance(agentId, now, true) + await this.runCheapMaintenance(agentId, now, true) return } - this.runCheapMaintenance(agentId, now, false) + await this.runCheapMaintenance(agentId, now, false) const model = this.ctx.resolveConsolidationModel(agentId) if (!model) { this.archiveStale(agentId, now) + await this.pruneDeadVectors(agentId) this.ctx.writeAudit(agentId, { eventType: 'memory/maintenance_llm', actorType: 'scheduler', @@ -297,6 +316,7 @@ export class MaintenanceService { if (!this.ctx.canWriteAgentMemory(agentId)) return this.refreshDecayScores(agentId, now) this.archiveStale(agentId, now) + await this.pruneDeadVectors(agentId) this.ports.syncWorkingMemoryAfterMutation(agentId) this.stampConsolidation(agentId, now) this.ctx.writeAudit(agentId, { @@ -322,57 +342,81 @@ export class MaintenanceService { model: MemoryModelRef ): Promise { const embedding = this.ctx.deps.resolveAgentConfig(agentId)?.memoryEmbedding - if (embedding?.providerId && embedding?.modelId) { - await this.ports.warmVectorStore(agentId, { - providerId: embedding.providerId, - modelId: embedding.modelId + if (!embedding?.providerId || !embedding?.modelId) return false + const currentEmbedding = { providerId: embedding.providerId, modelId: embedding.modelId } + const fingerprint = embeddingFingerprint(embedding.providerId, embedding.modelId) + const dimensions = this.ctx.deps.repository.getCurrentEmbeddingDimension(agentId, fingerprint) + if (dimensions === null) return false + await this.ports.warmVectorStore(agentId, currentEmbedding) + if (!this.ctx.canWriteAgentMemory(agentId)) return false + + const cursor = this.consolidationScanCursor.get(agentId) + let scanRows = this.ctx.deps.repository.listConsolidationScanRows(agentId, { + embeddingDim: dimensions, + embeddingModel: fingerprint, + after: cursor, + limit: CONSOLIDATION_MAX_NEIGHBOR_SCANS + 1 + }) + if (cursor && scanRows.length === 0) { + scanRows = this.ctx.deps.repository.listConsolidationScanRows(agentId, { + embeddingDim: dimensions, + embeddingModel: fingerprint, + limit: CONSOLIDATION_MAX_NEIGHBOR_SCANS + 1 }) - if (!this.ctx.canWriteAgentMemory(agentId)) return false } - - const active = this.ctx.deps.repository - .listByAgent(agentId) - .filter( - (row) => - !row.superseded_by && - row.kind !== 'persona' && - row.kind !== 'working' && - row.status !== 'archived' && - row.status !== 'conflicted' - ) - .sort((a, b) => a.created_at - b.created_at) + const hasMore = scanRows.length > CONSOLIDATION_MAX_NEIGHBOR_SCANS + scanRows = scanRows.slice(0, CONSOLIDATION_MAX_NEIGHBOR_SCANS) + if (!scanRows.length) { + this.consolidationScanCursor.delete(agentId) + return false + } let calls = 0 let inputTokens = 0 const merged = new Set() + let lastScanned: AgentMemoryRow | null = null let touched = false - for (const row of active) { + for (const row of scanRows) { if (calls >= CONSOLIDATION_MAX_LLM_CALLS || inputTokens >= CONSOLIDATION_MAX_INPUT_TOKENS) { break } + lastScanned = row if (merged.has(row.id)) continue + const source = this.ctx.deps.repository.getById(row.id) + if (!this.isCurrentEmbeddedConsolidationRow(agentId, source, dimensions, fingerprint)) + continue - let hits: MemoryRecallItem[] = [] + let matches: Array<{ memoryId: string; distance: number }> = [] try { - hits = await this.ports.retrieve(agentId, row.content, now, false) + matches = await this.ports.queryNeighborsByMemoryId( + agentId, + currentEmbedding, + dimensions, + source.id, + DECISION_NEIGHBOR_TOP_S + ) } catch { continue } if (!this.ctx.canWriteAgentMemory(agentId)) break - const neighbor = hits.find( - (hit) => - hit.id !== row.id && - !merged.has(hit.id) && - (hit.similarity ?? 0) >= CONSOLIDATION_MERGE_SIMILARITY - ) + let neighbor: AgentMemoryRow | null = null + for (const match of matches) { + if (match.memoryId === source.id || merged.has(match.memoryId)) continue + if (distanceToSimilarity(match.distance) < CONSOLIDATION_MERGE_SIMILARITY) continue + const neighborRow = this.ctx.deps.repository.getById(match.memoryId) + if (!this.isCurrentEmbeddedConsolidationRow(agentId, neighborRow, dimensions, fingerprint)) + continue + neighbor = neighborRow + break + } if (!neighbor) continue const promptCandidate = normalizeMemoryCandidate({ - kind: row.kind === 'episodic' ? 'episodic' : 'semantic', - category: row.category, - content: row.content, - importance: row.importance + kind: source.kind === 'episodic' ? 'episodic' : 'semantic', + category: source.category, + content: source.content, + importance: source.importance }) if (!promptCandidate) continue const prompt = buildDecisionPrompt(promptCandidate, [{ content: neighbor.content }]) @@ -389,10 +433,8 @@ export class MaintenanceService { if (!this.ctx.canWriteAgentMemory(agentId)) break if (decision.decision === 'UPDATE' || decision.decision === 'SUPERSEDE') { - const neighborRow = this.ctx.deps.repository.getById(neighbor.id) - if (!neighborRow) continue const [primary, secondary] = - row.created_at >= neighborRow.created_at ? [row, neighborRow] : [neighborRow, row] + source.created_at >= neighbor.created_at ? [source, neighbor] : [neighbor, source] const mergedContent = decision.mergedContent ?? primary.content const secondaryCategory = isAgentMemoryCategory(secondary.category) ? secondary.category @@ -415,17 +457,101 @@ export class MaintenanceService { merged.add(survivorId) touched = true } - this.ctx.deps.repository.setLastConsolidatedAt(row.id, now) + this.ctx.deps.repository.setLastConsolidatedAt(source.id, now) + } + const windowFullyScanned = + !!lastScanned && + scanRows.length > 0 && + lastScanned.id === scanRows[scanRows.length - 1].id && + lastScanned.created_at === scanRows[scanRows.length - 1].created_at + if (lastScanned && (hasMore || !windowFullyScanned)) { + this.consolidationScanCursor.set(agentId, { + createdAt: lastScanned.created_at, + id: lastScanned.id + }) + } else { + this.consolidationScanCursor.delete(agentId) } return touched } - private runCheapMaintenance(agentId: string, now: number, archive: boolean): void { + private isLiveConsolidationNeighbor( + agentId: string, + row: AgentMemoryRow | undefined + ): row is AgentMemoryRow { + return ( + !!row && + row.agent_id === agentId && + !row.superseded_by && + row.kind !== 'persona' && + row.kind !== 'working' && + row.status !== 'archived' && + row.status !== 'conflicted' + ) + } + + private isCurrentEmbeddedConsolidationRow( + agentId: string, + row: AgentMemoryRow | undefined, + dimensions: number, + fingerprint: string + ): row is AgentMemoryRow { + return ( + this.isLiveConsolidationNeighbor(agentId, row) && + row.status === 'embedded' && + row.embedding_dim === dimensions && + row.embedding_model === fingerprint + ) + } + + private async runCheapMaintenance(agentId: string, now: number, archive: boolean): Promise { this.refreshDecayScores(agentId, now) - if (archive) this.archiveStale(agentId, now) + if (archive) { + this.archiveStale(agentId, now) + await this.pruneDeadVectors(agentId) + } + const repaired = this.ctx.deps.repository.repairInternalKindStatuses(agentId) + if (repaired > 0) { + this.ctx.writeAudit(agentId, { + eventType: 'memory/repair', + actorType: 'scheduler', + status: 'completed', + outputRefs: { repaired } + }) + } this.ports.syncWorkingMemoryAfterMutation(agentId) } + private async pruneDeadVectors(agentId: string): Promise { + const embedding = this.ctx.deps.resolveAgentConfig(agentId)?.memoryEmbedding + if (!embedding?.providerId || !embedding?.modelId) return + const currentEmbedding = { providerId: embedding.providerId, modelId: embedding.modelId } + const dimensions = this.ports.getWarmVectorStoreDimension(agentId, currentEmbedding) + if (dimensions === null) return + const fingerprint = embeddingFingerprint(embedding.providerId, embedding.modelId) + const refs = this.ctx.deps.repository.listPrunableVectorRefs(agentId, { + limit: VECTOR_PRUNE_BATCH_LIMIT, + embeddingModel: fingerprint, + embeddingDim: dimensions + }) + if (!refs.length) return + if (!this.ctx.canWriteAgentMemory(agentId)) return + const deletedIds = await this.ports.deletePrunableVectorsForMemoryIds( + agentId, + currentEmbedding, + dimensions, + refs.map((ref) => ref.id) + ) + if (deletedIds.length && this.ctx.canWriteAgentMemory(agentId)) { + this.ctx.deps.repository.clearPrunableEmbeddingRefs( + agentId, + deletedIds, + dimensions, + fingerprint + ) + } + } + private stampConsolidation(agentId: string, now: number): void { for (const row of this.ctx.deps.repository.listByAgent(agentId)) { if (row.kind === 'persona') continue @@ -471,6 +597,7 @@ export class MaintenanceService { this.consolidationTimers.delete(agentId) this.consolidationTimerDueAt.delete(agentId) this.lastConsolidationAt.delete(agentId) + this.consolidationScanCursor.delete(agentId) } getInFlight(): Promise[] { @@ -485,10 +612,12 @@ export class MaintenanceService { getMutableRuntimeStateForTests(): { consolidationTimers: Map lastConsolidationAt: Map + consolidationScanCursor: Map } { return { consolidationTimers: this.consolidationTimers, - lastConsolidationAt: this.lastConsolidationAt + lastConsolidationAt: this.lastConsolidationAt, + consolidationScanCursor: this.consolidationScanCursor } } } diff --git a/src/main/presenter/memoryPresenter/services/managementService.ts b/src/main/presenter/memoryPresenter/services/managementService.ts index 8563cb48a..698785c4c 100644 --- a/src/main/presenter/memoryPresenter/services/managementService.ts +++ b/src/main/presenter/memoryPresenter/services/managementService.ts @@ -36,6 +36,10 @@ function toHealthTopAccessedItem( } } +function isInternalMemoryKind(row: AgentMemoryRow): boolean { + return row.kind === 'persona' || row.kind === 'working' +} + export class ManagementService { constructor( private readonly ctx: MemoryRuntimeContext, @@ -55,6 +59,7 @@ export class ManagementService { if (!this.ctx.canWriteAgentMemory(agentId)) return false const row = this.ctx.deps.repository.getById(memoryId) if (!row || row.agent_id !== agentId || row.status !== 'archived') return false + if (isInternalMemoryKind(row)) return false this.ctx.deps.repository.updateStatus(memoryId, 'pending_embedding') this.ports.syncWorkingMemoryAfterMutation(agentId) void this.ports.triggerEmbedding(agentId).catch((error) => { @@ -70,6 +75,7 @@ export class ManagementService { if (!this.ctx.isManagedAgent(agentId)) return false const row = this.ctx.deps.repository.getById(memoryId) if (!row || row.agent_id !== agentId) return false + if (isInternalMemoryKind(row)) return false if (row.status === 'archived') return true this.ctx.deps.repository.archive(row.id, Date.now()) if (this.ctx.isDisposed) return true @@ -84,6 +90,7 @@ export class ManagementService { if (!this.ctx.isManagedAgent(agentId)) return false const row = this.ctx.deps.repository.getById(memoryId) if (!row || row.agent_id !== agentId) return false + if (isInternalMemoryKind(row)) return false const alreadyArchived = row.status === 'archived' if (!alreadyArchived) { this.ctx.deps.repository.archive(row.id, Date.now()) @@ -107,7 +114,9 @@ export class ManagementService { listMemories(agentId: string): AgentMemoryRow[] { this.ctx.assertSafeAgentId(agentId) if (!this.ctx.isManagedAgent(agentId)) return [] - return this.ctx.deps.repository.listByAgent(agentId, { includeArchived: true }) + return this.ctx.deps.repository + .listByAgent(agentId, { includeArchived: true }) + .filter((row) => !isInternalMemoryKind(row)) } getByIds(agentId: string, memoryIds: string[]): AgentMemoryRow[] { diff --git a/src/main/presenter/memoryPresenter/services/retrievalService.ts b/src/main/presenter/memoryPresenter/services/retrievalService.ts index c7c049cd3..a890c0552 100644 --- a/src/main/presenter/memoryPresenter/services/retrievalService.ts +++ b/src/main/presenter/memoryPresenter/services/retrievalService.ts @@ -18,10 +18,17 @@ import { type MemoryInjectionResult } from '../core/injectionPort' import { + MEMORY_SEARCH_DEFAULT_LIMIT, + RECALL_QUERY_EMBEDDING_MAX_CONCURRENT, RECALL_QUERY_EMBEDDING_STALE_MS, RECALL_QUERY_EMBEDDING_TIMEOUT_MS } from '../runtimeConstants' -import type { AgentMemoryRow, MemoryRecallItem, MemorySearchHit } from '../types' +import { + MAX_TOP_K, + type AgentMemoryRow, + type MemoryRecallItem, + type MemorySearchHit +} from '../types' import { embeddingFingerprint, type MemoryModelRef, type MemoryRuntimeContext } from '../context' import type { VectorStoreRetrievalPort, WorkingMemoryReadPort } from '../ports' @@ -32,6 +39,26 @@ type QueryEmbeddingInFlight = { promise: Promise } +function isLiveRecallVectorRow( + agentId: string, + row: AgentMemoryRow | undefined +): row is AgentMemoryRow { + return ( + !!row && + row.agent_id === agentId && + !row.superseded_by && + row.kind !== 'persona' && + row.kind !== 'working' && + row.status !== 'archived' && + row.status !== 'conflicted' + ) +} + +function clampRetrievalTopK(value: number): number { + if (!Number.isFinite(value)) return 1 + return Math.min(MAX_TOP_K, Math.max(1, Math.floor(value))) +} + async function withSoftTimeout( promise: Promise, timeoutMs: number @@ -50,7 +77,7 @@ async function withSoftTimeout( } export class RetrievalService { - private readonly queryEmbeddingInFlight = new Map() + private readonly queryEmbeddingInFlight = new Map>() constructor( private readonly ctx: MemoryRuntimeContext, @@ -66,6 +93,12 @@ export class RetrievalService { reindexEmbeddings: (agentId: string, force?: boolean) => Promise backfillEmbeddings: (agentId: string) => Promise isReindexing: (agentId: string) => boolean + deletePrunableVectorsForMemoryIds: ( + agentId: string, + embedding: MemoryModelRef, + dimensions: number, + memoryIds: string[] + ) => Promise } ) {} @@ -94,18 +127,34 @@ export class RetrievalService { ): Promise | null { const key = `${agentId}::${embeddingFingerprint(embedding.providerId, embedding.modelId)}` const now = Date.now() - const existing = this.queryEmbeddingInFlight.get(key) - if (existing && now - existing.startedAt < RECALL_QUERY_EMBEDDING_STALE_MS) return null - if (existing) { - logger.warn(`[Memory] stale query embedding replaced for ${agentId}`) + let group = this.queryEmbeddingInFlight.get(key) + if (!group) { + group = new Map() + this.queryEmbeddingInFlight.set(key, group) } + let replacedStale = false + for (const [trackedQuery, entry] of group) { + if (now - entry.startedAt >= RECALL_QUERY_EMBEDDING_STALE_MS) { + group.delete(trackedQuery) + replacedStale = true + } + } + if (replacedStale) logger.warn(`[Memory] stale query embedding replaced for ${agentId}`) + + const existing = group.get(query) + if (existing) return existing.promise + if (group.size >= RECALL_QUERY_EMBEDDING_MAX_CONCURRENT) return null const promise = this.ctx.deps.getEmbeddings(embedding.providerId, embedding.modelId, [query]) const entry: QueryEmbeddingInFlight = { startedAt: now, promise } - this.queryEmbeddingInFlight.set(key, entry) + group.set(query, entry) promise .finally(() => { - if (this.queryEmbeddingInFlight.get(key) === entry) this.queryEmbeddingInFlight.delete(key) + const currentGroup = this.queryEmbeddingInFlight.get(key) + if (currentGroup?.get(query) === entry) { + currentGroup.delete(query) + if (currentGroup.size === 0) this.queryEmbeddingInFlight.delete(key) + } }) .catch(() => undefined) return promise @@ -117,9 +166,16 @@ export class RetrievalService { options: { limit?: number } = {} ): Promise { if (!this.ctx.canReadAgentMemory(agentId)) return [] - const hits = await this.retrieve(agentId, query, Date.now(), false) - const limited = - options.limit != null ? hits.slice(0, Math.max(0, Math.floor(options.limit))) : hits + const limit = + options.limit != null + ? Math.min(MAX_TOP_K, Math.max(0, Math.floor(options.limit))) + : MEMORY_SEARCH_DEFAULT_LIMIT + if (limit === 0) return [] + const hits = await this.retrieve(agentId, query, Date.now(), false, { + topKOverride: limit, + enableInlinePrune: false + }) + const limited = hits.slice(0, limit) const results: MemorySearchHit[] = [] for (const hit of limited) { const row = this.ctx.deps.repository.getById(hit.id) @@ -138,6 +194,8 @@ export class RetrievalService { trace?: boolean keywordQuery?: string keywordMatchMode?: 'all' | 'any' + topKOverride?: number + enableInlinePrune?: boolean } = {} ): Promise { if (!this.ctx.canReadAgentMemory(agentId)) return [] @@ -147,7 +205,9 @@ export class RetrievalService { if (!normalizedQuery) return [] const normalizedKeywordQuery = (options.keywordQuery ?? normalizedQuery).trim() - const candidateLimit = topK * 2 + const effectiveTopK = + options.topKOverride !== undefined ? clampRetrievalTopK(options.topKOverride) : topK + const candidateLimit = effectiveTopK * 2 const ftsRows = normalizedKeywordQuery ? this.ctx.deps.repository .search(agentId, normalizedKeywordQuery, candidateLimit, { @@ -213,21 +273,33 @@ export class RetrievalService { this.vectorStore.markReady(agentId, currentEmbedding, vector.length) const matches = await store.query(vector, { topK: candidateLimit }) if (!this.ctx.canReadAgentMemory(agentId)) return [] + const deadVectorIds: string[] = [] for (const match of matches) { const similarity = distanceToSimilarity(match.distance) if (similarity < similarityThreshold) continue const row = this.ctx.deps.repository.getById(match.memoryId) - if ( - !row || - row.superseded_by || - row.kind === 'persona' || - row.kind === 'working' || - row.status === 'archived' || - row.status === 'conflicted' - ) + if (!isLiveRecallVectorRow(agentId, row)) { + deadVectorIds.push(match.memoryId) continue + } vecMatches.push({ row, similarity }) } + if ( + options.enableInlinePrune !== false && + deadVectorIds.length && + this.ctx.canReadAgentMemory(agentId) + ) { + void this.ports + .deletePrunableVectorsForMemoryIds( + agentId, + currentEmbedding, + vector.length, + [...new Set(deadVectorIds)] + ) + .catch((error) => { + logger.warn(`[Memory] inline vector prune failed: ${String(error)}`) + }) + } if (this.ctx.canReadAgentMemory(agentId) && !this.ports.isReindexing(agentId)) { void this.ports.backfillEmbeddings(agentId).catch((error) => { logger.warn(`[Memory] backfill failed for ${agentId}: ${String(error)}`) @@ -254,7 +326,7 @@ export class RetrievalService { } const results = fuse(ftsRows, vecMatches, { - topK, + topK: effectiveTopK, rrfK, weights, now, diff --git a/src/main/presenter/memoryPresenter/types.ts b/src/main/presenter/memoryPresenter/types.ts index 1225b32ec..5aa34a680 100644 --- a/src/main/presenter/memoryPresenter/types.ts +++ b/src/main/presenter/memoryPresenter/types.ts @@ -114,6 +114,32 @@ export interface MemoryRepositoryPort { countByAgent(agentId: string): number hasActiveMemory(agentId: string): boolean listAgentIdsWithMemories(): string[] + listConsolidationScanRows( + agentId: string, + options: { + embeddingDim: number + embeddingModel: string + after?: ConsolidationScanCursor + limit: number + } + ): AgentMemoryRow[] + repairInternalKindStatuses(agentId: string): number + listPrunableVectorRefs( + agentId: string, + options: { limit: number; embeddingModel?: string; embeddingDim?: number } + ): MemoryVectorRef[] + filterPrunableVectorRefs( + agentId: string, + ids: string[], + embeddingDim: number, + embeddingModel: string + ): string[] + clearPrunableEmbeddingRefs( + agentId: string, + ids: string[], + embeddingDim: number, + embeddingModel: string + ): number } export interface RecallKeywordTermStat { @@ -165,10 +191,22 @@ export interface MemoryVectorQueryOptions { threshold?: number } +export interface MemoryVectorRef { + id: string + embeddingDim: number + embeddingModel: string +} + +export interface ConsolidationScanCursor { + createdAt: number + id: string +} + // Vector store port (DuckDB), isolated per agent: one database each, with independent dimensions. export interface IMemoryVectorStore { upsert(records: MemoryVectorRecord[]): Promise query(embedding: number[], options: MemoryVectorQueryOptions): Promise + queryByMemoryId(memoryId: string, options: MemoryVectorQueryOptions): Promise deleteByMemoryIds(memoryIds: string[]): Promise close(): Promise isUsable(): boolean diff --git a/src/main/presenter/sqlitePresenter/tables/agentMemory.ts b/src/main/presenter/sqlitePresenter/tables/agentMemory.ts index e94fa798f..cd7905b48 100644 --- a/src/main/presenter/sqlitePresenter/tables/agentMemory.ts +++ b/src/main/presenter/sqlitePresenter/tables/agentMemory.ts @@ -692,7 +692,7 @@ export class AgentMemoryTable extends BaseTable { AND am.superseded_by IS NULL AND am.status != 'archived' AND am.status != 'conflicted' - AND am.kind != 'working' + AND am.kind NOT IN ('persona', 'working') ORDER BY bm25(agent_memory_fts) LIMIT ?` ) @@ -721,7 +721,7 @@ export class AgentMemoryTable extends BaseTable { AND superseded_by IS NULL AND status != 'archived' AND status != 'conflicted' - AND kind != 'working' + AND kind NOT IN ('persona', 'working') AND (${clauses.join(operator)}) ORDER BY importance DESC, created_at DESC LIMIT ?` @@ -1213,4 +1213,151 @@ export class AgentMemoryTable extends BaseTable { .all() as Array<{ agent_id: string }> return rows.map((row) => row.agent_id) } + + listConsolidationScanRows( + agentId: string, + options: { + embeddingDim: number + embeddingModel: string + after?: { createdAt: number; id: string } + limit: number + } + ): AgentMemoryRow[] { + const cappedLimit = Math.max(0, Math.floor(options.limit)) + if (cappedLimit === 0) return [] + const params: Array = [agentId, options.embeddingDim, options.embeddingModel] + const cursorClause = options.after ? 'AND (created_at > ? OR (created_at = ? AND id > ?))' : '' + if (options.after) { + params.push(options.after.createdAt, options.after.createdAt, options.after.id) + } + params.push(cappedLimit) + return this.db + .prepare( + `SELECT * + FROM agent_memory + WHERE agent_id = ? + AND status = 'embedded' + AND superseded_by IS NULL + AND kind NOT IN ('persona', 'working') + AND embedding_dim = ? + AND embedding_model = ? + ${cursorClause} + ORDER BY created_at ASC, id ASC + LIMIT ?` + ) + .all(...params) as AgentMemoryRow[] + } + + repairInternalKindStatuses(agentId: string): number { + const result = this.db + .prepare( + `UPDATE agent_memory + SET status = 'fts_only' + WHERE agent_id = ? + AND kind IN ('persona', 'working') + AND status != 'fts_only'` + ) + .run(agentId) + return result.changes + } + + listPrunableVectorRefs( + agentId: string, + options: { limit: number; embeddingModel?: string; embeddingDim?: number } + ): Array<{ id: string; embeddingDim: number; embeddingModel: string }> { + const cappedLimit = Math.max(0, Math.floor(options.limit)) + if (cappedLimit === 0) return [] + const params: Array = [agentId] + const embeddingModelClause = options.embeddingModel ? 'AND embedding_model = ?' : '' + if (options.embeddingModel) params.push(options.embeddingModel) + const embeddingDimClause = options.embeddingDim !== undefined ? 'AND embedding_dim = ?' : '' + if (options.embeddingDim !== undefined) params.push(options.embeddingDim) + params.push(cappedLimit) + const rows = this.db + .prepare( + `SELECT id, + embedding_dim AS embeddingDim, + embedding_model AS embeddingModel + FROM agent_memory + WHERE agent_id = ? + AND embedding_id IS NOT NULL + AND embedding_dim IS NOT NULL + AND embedding_dim > 0 + AND embedding_model IS NOT NULL + ${embeddingModelClause} + ${embeddingDimClause} + AND ( + kind IN ('persona', 'working') OR + superseded_by IS NOT NULL OR + status = 'archived' + ) + ORDER BY created_at ASC, id ASC + LIMIT ?` + ) + .all(...params) as Array<{ + id: string + embeddingDim: number + embeddingModel: string + }> + return rows + } + + filterPrunableVectorRefs( + agentId: string, + ids: string[], + embeddingDim: number, + embeddingModel: string + ): string[] { + const uniqueIds = [...new Set(ids.filter((id) => id.trim()))] + if (!uniqueIds.length) return [] + const placeholders = uniqueIds.map(() => '?').join(', ') + const rows = this.db + .prepare( + `SELECT id + FROM agent_memory + WHERE agent_id = ? + AND id IN (${placeholders}) + AND embedding_id IS NOT NULL + AND embedding_dim = ? + AND embedding_model = ? + AND ( + kind IN ('persona', 'working') OR + superseded_by IS NOT NULL OR + status = 'archived' + )` + ) + .all(agentId, ...uniqueIds, embeddingDim, embeddingModel) as Array<{ id: string }> + const liveIds = new Set(rows.map((row) => row.id)) + return uniqueIds.filter((id) => liveIds.has(id)) + } + + clearPrunableEmbeddingRefs( + agentId: string, + ids: string[], + embeddingDim: number, + embeddingModel: string + ): number { + const uniqueIds = [...new Set(ids.filter((id) => id.trim()))] + if (!uniqueIds.length) return 0 + const placeholders = uniqueIds.map(() => '?').join(', ') + const result = this.db + .prepare( + `UPDATE agent_memory + SET embedding_id = NULL, + embedding_dim = NULL, + embedding_model = NULL + WHERE agent_id = ? + AND id IN (${placeholders}) + AND embedding_id IS NOT NULL + AND embedding_dim = ? + AND embedding_model = ? + AND ( + kind IN ('persona', 'working') OR + superseded_by IS NOT NULL OR + status = 'archived' + )` + ) + .run(agentId, ...uniqueIds, embeddingDim, embeddingModel) + return result.changes + } } diff --git a/src/shared/contracts/routes/memory.routes.ts b/src/shared/contracts/routes/memory.routes.ts index 6e2661949..c65269a37 100644 --- a/src/shared/contracts/routes/memory.routes.ts +++ b/src/shared/contracts/routes/memory.routes.ts @@ -348,8 +348,8 @@ export const memorySearchRoute = defineRouteContract({ input: z.object({ agentId: AgentIdSchema, query: z.string(), - // Caps the result count only; it cannot widen the agent's configured topK. - limit: z.number().int().positive().max(500).optional() + // Search-only retrieval depth/result cap. Defaults to 50 and is clamped by the presenter to 100. + limit: z.number().int().positive().max(100).optional() }), output: z.object({ results: z.array(MemorySearchResultSchema) }) }) diff --git a/test/main/presenter/agentMemoryTable.test.ts b/test/main/presenter/agentMemoryTable.test.ts index 8ad346881..c4ee7f1e6 100644 --- a/test/main/presenter/agentMemoryTable.test.ts +++ b/test/main/presenter/agentMemoryTable.test.ts @@ -1229,6 +1229,275 @@ describeIfSqlite('AgentMemoryTable FTS5 + migration', () => { } }) + it('search excludes persona rows before applying the SQL limit', () => { + const db = new DatabaseCtor(':memory:') + try { + const table = new AgentMemoryTableCtor(db) + table.createTable() + for (let index = 0; index < 10; index += 1) { + table.insert({ + id: `persona-${index}`, + agentId: 'a', + kind: 'persona', + content: `redis persona ${index}`, + status: 'fts_only', + createdAt: 100 + index + }) + } + for (let index = 0; index < 2; index += 1) { + table.insert({ + id: `mem-${index}`, + agentId: 'a', + kind: 'semantic', + content: `redis memory ${index}`, + status: 'fts_only', + createdAt: index + }) + } + + const ids = table.search('a', 'redis', 2).map((row) => row.id) + expect(ids).toHaveLength(2) + expect(new Set(ids)).toEqual(new Set(['mem-0', 'mem-1'])) + } finally { + db.close() + } + }) + + it('repairs persona and working rows back to fts_only without clearing embedding refs', () => { + const db = new DatabaseCtor(':memory:') + try { + const table = new AgentMemoryTableCtor(db) + table.createTable() + table.insert({ id: 'persona', agentId: 'a', kind: 'persona', content: 'self model' }) + table.updateStatus('persona', 'pending_embedding', { + embeddingId: 'persona', + embeddingDim: 3, + embeddingModel: 'p:m' + }) + table.insert({ id: 'work', agentId: 'a', kind: 'working', content: 'working blob' }) + table.updateStatus('work', 'embedded', { + embeddingId: 'work', + embeddingDim: 3, + embeddingModel: 'p:m' + }) + table.insert({ id: 'mem', agentId: 'a', kind: 'semantic', content: 'redis memory' }) + table.updateStatus('mem', 'pending_embedding') + + expect(table.repairInternalKindStatuses('a')).toBe(2) + expect(table.getById('persona')).toMatchObject({ + status: 'fts_only', + embedding_id: 'persona', + embedding_dim: 3, + embedding_model: 'p:m' + }) + expect(table.getById('work')).toMatchObject({ + status: 'fts_only', + embedding_id: 'work', + embedding_dim: 3, + embedding_model: 'p:m' + }) + expect(table.getById('mem')?.status).toBe('pending_embedding') + expect(table.repairInternalKindStatuses('a')).toBe(0) + } finally { + db.close() + } + }) + + it('lists prunable vector refs and clears embedding refs without changing lifecycle state', () => { + const db = new DatabaseCtor(':memory:') + try { + const table = new AgentMemoryTableCtor(db) + table.createTable() + table.insert({ + id: 'active', + agentId: 'a', + kind: 'semantic', + content: 'active', + createdAt: 1 + }) + table.updateStatus('active', 'embedded', { + embeddingId: 'active', + embeddingDim: 3, + embeddingModel: 'p:m' + }) + table.insert({ + id: 'persona', + agentId: 'a', + kind: 'persona', + content: 'self', + status: 'fts_only', + createdAt: 2 + }) + table.updateStatus('persona', 'fts_only', { + embeddingId: 'persona', + embeddingDim: 3, + embeddingModel: 'p:m' + }) + table.insert({ + id: 'work', + agentId: 'a', + kind: 'working', + content: 'working', + status: 'fts_only', + createdAt: 3 + }) + table.updateStatus('work', 'fts_only', { + embeddingId: 'work', + embeddingDim: 3, + embeddingModel: 'p:m' + }) + table.insert({ + id: 'archived', + agentId: 'a', + kind: 'semantic', + content: 'archived', + createdAt: 4 + }) + table.updateStatus('archived', 'embedded', { + embeddingId: 'archived', + embeddingDim: 3, + embeddingModel: 'p:m' + }) + table.archive('archived') + table.insert({ + id: 'superseded', + agentId: 'a', + kind: 'semantic', + content: 'superseded', + createdAt: 5 + }) + table.updateStatus('superseded', 'embedded', { + embeddingId: 'superseded', + embeddingDim: 3, + embeddingModel: 'p:m' + }) + table.markSuperseded('superseded', 'active') + + expect(table.listPrunableVectorRefs('a', { limit: 10 })).toEqual([ + { id: 'persona', embeddingDim: 3, embeddingModel: 'p:m' }, + { id: 'work', embeddingDim: 3, embeddingModel: 'p:m' }, + { id: 'archived', embeddingDim: 3, embeddingModel: 'p:m' }, + { id: 'superseded', embeddingDim: 3, embeddingModel: 'p:m' } + ]) + expect( + table.filterPrunableVectorRefs('a', ['active', 'archived', 'superseded'], 3, 'p:m') + ).toEqual(['archived', 'superseded']) + expect( + table.clearPrunableEmbeddingRefs('a', ['active', 'archived', 'superseded'], 3, 'p:m') + ).toBe(2) + expect(table.getById('archived')).toMatchObject({ + status: 'archived', + embedding_id: null, + embedding_dim: null, + embedding_model: null + }) + expect(table.getById('superseded')).toMatchObject({ + superseded_by: 'active', + status: 'embedded', + embedding_id: null, + embedding_dim: null, + embedding_model: null + }) + expect(table.getById('active')?.embedding_id).toBe('active') + } finally { + db.close() + } + }) + + it('filters prunable vector refs by embedding model before applying the limit', () => { + const db = new DatabaseCtor(':memory:') + try { + const table = new AgentMemoryTableCtor(db) + table.createTable() + for (let index = 0; index < 260; index += 1) { + const id = `old-${index}` + table.insert({ + id, + agentId: 'a', + kind: 'semantic', + content: `old archived ${index}`, + createdAt: index + }) + table.updateStatus(id, 'embedded', { + embeddingId: id, + embeddingDim: 3, + embeddingModel: 'old:model' + }) + table.archive(id) + } + for (let index = 0; index < 2; index += 1) { + const id = `current-${index}` + table.insert({ + id, + agentId: 'a', + kind: 'semantic', + content: `current archived ${index}`, + createdAt: 1000 + index + }) + table.updateStatus(id, 'embedded', { + embeddingId: id, + embeddingDim: 3, + embeddingModel: 'p:m' + }) + table.archive(id) + } + + expect( + table.listPrunableVectorRefs('a', { limit: 2, embeddingModel: 'p:m' }).map((ref) => ref.id) + ).toEqual(['current-0', 'current-1']) + } finally { + db.close() + } + }) + + it('filters prunable vector refs by embedding dim before applying the limit', () => { + const db = new DatabaseCtor(':memory:') + try { + const table = new AgentMemoryTableCtor(db) + table.createTable() + for (let index = 0; index < 260; index += 1) { + const id = `old-dim-${index}` + table.insert({ + id, + agentId: 'a', + kind: 'semantic', + content: `old dim archived ${index}`, + createdAt: index + }) + table.updateStatus(id, 'embedded', { + embeddingId: id, + embeddingDim: 8, + embeddingModel: 'p:m' + }) + table.archive(id) + } + for (let index = 0; index < 2; index += 1) { + const id = `current-dim-${index}` + table.insert({ + id, + agentId: 'a', + kind: 'semantic', + content: `current dim archived ${index}`, + createdAt: 1000 + index + }) + table.updateStatus(id, 'embedded', { + embeddingId: id, + embeddingDim: 4, + embeddingModel: 'p:m' + }) + table.archive(id) + } + + expect( + table + .listPrunableVectorRefs('a', { limit: 2, embeddingModel: 'p:m', embeddingDim: 4 }) + .map((ref) => ref.id) + ).toEqual(['current-dim-0', 'current-dim-1']) + } finally { + db.close() + } + }) + it('v32 migration backfills source_entry_ids and embedding_model on a legacy table', () => { const db = new DatabaseCtor(':memory:') try { diff --git a/test/main/presenter/fakes/memoryFakes.ts b/test/main/presenter/fakes/memoryFakes.ts index a3207c3b8..412ffb1bf 100644 --- a/test/main/presenter/fakes/memoryFakes.ts +++ b/test/main/presenter/fakes/memoryFakes.ts @@ -180,6 +180,7 @@ export class FakeRepository implements MemoryRepositoryPort { !row.superseded_by && row.status !== 'archived' && row.status !== 'conflicted' && + row.kind !== 'persona' && row.kind !== 'working' && (matchMode === 'any' ? terms.some((term) => row.content.toLowerCase().includes(term)) @@ -545,6 +546,113 @@ export class FakeRepository implements MemoryRepositoryPort { ) ] } + + listConsolidationScanRows( + agentId: string, + options: { + embeddingDim: number + embeddingModel: string + after?: { createdAt: number; id: string } + limit: number + } + ) { + return [...this.rows.values()] + .filter( + (row) => + row.agent_id === agentId && + row.status === 'embedded' && + row.superseded_by === null && + row.kind !== 'persona' && + row.kind !== 'working' && + row.embedding_dim === options.embeddingDim && + row.embedding_model === options.embeddingModel && + (!options.after || + row.created_at > options.after.createdAt || + (row.created_at === options.after.createdAt && row.id > options.after.id)) + ) + .sort((a, b) => a.created_at - b.created_at || a.id.localeCompare(b.id)) + .slice(0, Math.max(0, Math.floor(options.limit))) + } + + repairInternalKindStatuses(agentId: string) { + let changed = 0 + for (const row of this.rows.values()) { + if (row.agent_id !== agentId || (row.kind !== 'persona' && row.kind !== 'working')) continue + if (row.status === 'fts_only') continue + row.status = 'fts_only' + changed += 1 + } + return changed + } + + private isPrunableVectorRow( + agentId: string, + row: AgentMemoryRow | undefined, + embeddingDim?: number, + embeddingModel?: string + ): row is AgentMemoryRow { + return ( + !!row && + row.agent_id === agentId && + row.embedding_id !== null && + row.embedding_dim !== null && + row.embedding_dim > 0 && + row.embedding_model !== null && + (embeddingDim === undefined || row.embedding_dim === embeddingDim) && + (embeddingModel === undefined || row.embedding_model === embeddingModel) && + (row.kind === 'persona' || + row.kind === 'working' || + row.superseded_by !== null || + row.status === 'archived') + ) + } + + listPrunableVectorRefs( + agentId: string, + options: { limit: number; embeddingModel?: string; embeddingDim?: number } + ) { + return [...this.rows.values()] + .filter((row) => + this.isPrunableVectorRow(agentId, row, options.embeddingDim, options.embeddingModel) + ) + .sort((a, b) => a.created_at - b.created_at || a.id.localeCompare(b.id)) + .slice(0, Math.max(0, Math.floor(options.limit))) + .map((row) => ({ + id: row.id, + embeddingDim: row.embedding_dim!, + embeddingModel: row.embedding_model! + })) + } + + filterPrunableVectorRefs( + agentId: string, + ids: string[], + embeddingDim: number, + embeddingModel: string + ) { + const uniqueIds = [...new Set(ids.filter((id) => id.trim()))] + return uniqueIds.filter((id) => + this.isPrunableVectorRow(agentId, this.rows.get(id), embeddingDim, embeddingModel) + ) + } + + clearPrunableEmbeddingRefs( + agentId: string, + ids: string[], + embeddingDim: number, + embeddingModel: string + ) { + let changed = 0 + for (const id of new Set(ids)) { + const row = this.rows.get(id) + if (!this.isPrunableVectorRow(agentId, row, embeddingDim, embeddingModel)) continue + row.embedding_id = null + row.embedding_dim = null + row.embedding_model = null + changed += 1 + } + return changed + } } export class FakeAuditRepository implements MemoryAuditRepositoryPort { @@ -651,6 +759,16 @@ export class FakeVectorStore implements IMemoryVectorStore { .slice(0, options.topK) } + async queryByMemoryId(memoryId: string, options: { topK: number }): Promise { + const embedding = this.vectors.get(memoryId) + if (!embedding) return [] + return [...this.vectors.entries()] + .filter(([id]) => id !== memoryId) + .map(([id, vec]) => ({ memoryId: id, distance: 1 - cosine(embedding, vec) })) + .sort((a, b) => a.distance - b.distance) + .slice(0, options.topK) + } + async deleteByMemoryIds(memoryIds: string[]) { for (const id of memoryIds) this.vectors.delete(id) } diff --git a/test/main/presenter/memoryExtraction.test.ts b/test/main/presenter/memoryExtraction.test.ts index 7737b61b7..25704fcf8 100644 --- a/test/main/presenter/memoryExtraction.test.ts +++ b/test/main/presenter/memoryExtraction.test.ts @@ -156,6 +156,7 @@ describe('MemoryPresenter.extractAndStore', () => { createVectorStore: async () => ({ upsert: async () => {}, query: async () => [], + queryByMemoryId: async () => [], deleteByMemoryIds: async () => {}, clear: async () => {}, close: async () => {} @@ -210,6 +211,7 @@ describe('MemoryPresenter.extractAndStore', () => { createVectorStore: async () => ({ upsert: async () => {}, query: async () => [], + queryByMemoryId: async () => [], deleteByMemoryIds: async () => {}, clear: async () => {}, close: async () => {} @@ -245,6 +247,7 @@ describe('MemoryPresenter.extractAndStore', () => { createVectorStore: async () => ({ upsert: async () => {}, query: async () => [], + queryByMemoryId: async () => [], deleteByMemoryIds: async () => {}, clear: async () => {}, close: async () => {} @@ -298,6 +301,7 @@ describe('MemoryPresenter.extractAndStore triage gate, cheap model, lineage', () createVectorStore: async () => ({ upsert: async () => {}, query: async () => [], + queryByMemoryId: async () => [], deleteByMemoryIds: async () => {}, close: async () => {}, isUsable: () => true @@ -430,6 +434,7 @@ describe('MemoryPresenter.maybeReflect cheap model', () => { createVectorStore: async () => ({ upsert: async () => {}, query: async () => [], + queryByMemoryId: async () => [], deleteByMemoryIds: async () => {}, close: async () => {}, isUsable: () => true @@ -548,6 +553,7 @@ function makeFakeRepo() { source_session: input.sourceSession ?? null, embedding_id: null, embedding_dim: null, + embedding_model: null, user_scope: null, last_accessed: null, access_count: 0, @@ -604,6 +610,11 @@ function makeFakeRepo() { return n }, countByAgent: (agentId: string) => - [...rows.values()].filter((r) => r.agent_id === agentId).length + [...rows.values()].filter((r) => r.agent_id === agentId).length, + listConsolidationScanRows: () => [], + repairInternalKindStatuses: () => 0, + listPrunableVectorRefs: () => [], + filterPrunableVectorRefs: () => [], + clearPrunableEmbeddingRefs: () => 0 } } diff --git a/test/main/presenter/memoryPresenter.test.ts b/test/main/presenter/memoryPresenter.test.ts index 936fa9681..35359833d 100644 --- a/test/main/presenter/memoryPresenter.test.ts +++ b/test/main/presenter/memoryPresenter.test.ts @@ -1903,7 +1903,7 @@ describe('MemoryPresenter management', () => { } }) - it('skips duplicate query embeddings while a timed-out request is still in flight', async () => { + it('shares identical query embeddings while a timed-out request is still in flight', async () => { vi.useFakeTimers() try { const repo = new FakeRepository() @@ -1937,15 +1937,96 @@ describe('MemoryPresenter management', () => { expect((await first).map((item) => item.id)).toEqual([memoryId]) expect(queryEmbeddingCalls).toBe(1) - const second = await presenter.recall('a', 'redis setup') + const second = presenter.recall('a', 'redis setup') + await vi.advanceTimersByTimeAsync(801) - expect(second.map((item) => item.id)).toEqual([memoryId]) + expect((await second).map((item) => item.id)).toEqual([memoryId]) expect(queryEmbeddingCalls).toBe(1) } finally { vi.useRealTimers() } }) + it('shares identical concurrent query embeddings and returns vector hits to both callers', async () => { + const repo = new FakeRepository() + const store = new FakeVectorStore() + let resolveQueryEmbedding: ((vectors: number[][]) => void) | null = null + const getEmbeddings = vi.fn((_p: string, _m: string, texts: string[]) => { + if (texts[0] === 'redis') { + return new Promise((resolve) => { + resolveQueryEmbedding = resolve + }) + } + return Promise.resolve(texts.map((text) => textToVector(text))) + }) + const presenter = new MemoryPresenter({ + repository: repo, + resolveAgentConfig: () => enabledConfig, + getEmbeddings, + getDimensions: embeddingDimensions, + createVectorStore: async () => store, + resetVectorStore: async () => undefined + }) + const [memoryId] = presenter.writeMemoriesSync([{ kind: 'semantic', content: 'redis setup' }], { + agentId: 'a' + }) + await presenter.processPendingEmbeddings('a') + getEmbeddings.mockClear() + + const first = presenter.recall('a', 'redis') + const second = presenter.recall('a', 'redis') + await waitForMemoryCondition(() => resolveQueryEmbedding !== null) + resolveQueryEmbedding?.([textToVector('redis')]) + + const [firstHits, secondHits] = await Promise.all([first, second]) + expect(getEmbeddings).toHaveBeenCalledTimes(1) + expect(firstHits.find((item) => item.id === memoryId)?.sources?.vec).toBe(true) + expect(secondHits.find((item) => item.id === memoryId)?.sources?.vec).toBe(true) + }) + + it('allows two distinct query embeddings and skips only the third fresh query', async () => { + const repo = new FakeRepository() + const store = new FakeVectorStore() + const pendingQueryEmbeddings: Array<(vectors: number[][]) => void> = [] + let blockQueryEmbedding = false + let queryEmbeddingCalls = 0 + const getEmbeddings = vi.fn((_p: string, _m: string, texts: string[]) => { + if (blockQueryEmbedding && texts[0] !== 'memory warmup') { + queryEmbeddingCalls += 1 + return new Promise((resolve) => pendingQueryEmbeddings.push(resolve)) + } + return Promise.resolve(texts.map((text) => textToVector(text))) + }) + const presenter = new MemoryPresenter({ + repository: repo, + resolveAgentConfig: () => enabledConfig, + getEmbeddings, + getDimensions: embeddingDimensions, + createVectorStore: async () => store, + resetVectorStore: async () => undefined + }) + const [memoryId] = presenter.writeMemoriesSync( + [{ kind: 'semantic', content: 'redis vue setup' }], + { agentId: 'a' } + ) + await presenter.processPendingEmbeddings('a') + + blockQueryEmbedding = true + const first = presenter.recall('a', 'redis setup') + const second = presenter.recall('a', 'vue setup') + await waitForMemoryCondition(() => pendingQueryEmbeddings.length === 2) + const third = await presenter.recall('a', 'setup') + pendingQueryEmbeddings[0]([textToVector('redis setup')]) + pendingQueryEmbeddings[1]([textToVector('vue setup')]) + + const [firstHits, secondHits] = await Promise.all([first, second]) + expect(firstHits.find((item) => item.id === memoryId)?.sources?.vec).toBe(true) + expect(secondHits.find((item) => item.id === memoryId)?.sources?.vec).toBe(true) + expect(third.find((item) => item.id === memoryId)?.sources?.vec).not.toBe(true) + expect(third.map((item) => item.id)).toEqual([memoryId]) + expect(queryEmbeddingCalls).toBe(2) + }) + it('replaces stale query embedding in-flight entries without late-settle deletion', async () => { vi.useFakeTimers() try { @@ -2251,6 +2332,207 @@ describe('MemoryPresenter management', () => { expect((await presenter.recall('a', 'redis')).map((item) => item.id)).toContain(ids[0]) }) + it('inline-prunes vector matches that SQLite rejects as dead', async () => { + const { presenter, repo, store } = makePresenter(enabledConfig) + const [id] = presenter.writeMemoriesSync([{ kind: 'semantic', content: 'redis cache' }], { + agentId: 'a' + }) + await presenter.processPendingEmbeddings('a') + repo.archive(id, Date.now()) + expect(store.vectors.has(id)).toBe(true) + + await presenter.recall('a', 'redis') + + await waitForMemoryCondition(() => !store.vectors.has(id), 'dead vector was not pruned') + }) + + it('does not delete restored vectors from an in-flight inline prune', async () => { + const { presenter, repo, store } = makePresenter(enabledConfig) + const [id] = presenter.writeMemoriesSync([{ kind: 'semantic', content: 'redis cache' }], { + agentId: 'a' + }) + await presenter.processPendingEmbeddings('a') + repo.archive(id, Date.now()) + const originalFilterPrunable = repo.filterPrunableVectorRefs.bind(repo) + const filterPrunable = vi + .spyOn(repo, 'filterPrunableVectorRefs') + .mockImplementation((agentId, ids, embeddingDim, embeddingModel) => { + if (ids.includes(id)) { + repo.updateStatus(id, 'embedded', { + embeddingId: id, + embeddingDim, + embeddingModel + }) + store.vectors.set(id, textToVector('redis cache restored')) + } + return originalFilterPrunable(agentId, ids, embeddingDim, embeddingModel) + }) + const deleteByMemoryIds = vi.spyOn(store, 'deleteByMemoryIds') + + await presenter.recall('a', 'redis') + await waitForMemoryCondition(() => filterPrunable.mock.calls.length > 0) + + expect(deleteByMemoryIds).not.toHaveBeenCalled() + expect(repo.getById(id)).toMatchObject({ + status: 'embedded', + embedding_id: id, + embedding_dim: 4, + embedding_model: 'p:m' + }) + expect(store.vectors.has(id)).toBe(true) + }) + + it('maintenance sweep deletes prunable vectors before clearing embedding refs', async () => { + const generateText = routedLLM({}) + const { presenter, repo, store } = makeLLMPresenter(generateText) + const id = await seedEmbedded(presenter, 'user likes redis') + repo.archive(id, Date.now()) + expect(store.vectors.has(id)).toBe(true) + expect(repo.getById(id)?.embedding_id).toBe(id) + + await presenter.runConsolidationPass('a', 1_000 * DAY) + + expect(store.vectors.has(id)).toBe(false) + expect(repo.getById(id)).toMatchObject({ + status: 'archived', + embedding_id: null, + embedding_dim: null, + embedding_model: null + }) + }) + + it('does not delete restored vectors when a row changes before guarded prune', async () => { + const generateText = routedLLM({}) + const { presenter, repo, store } = makeLLMPresenter(generateText) + const id = await seedEmbedded(presenter, 'user likes redis') + repo.archive(id, Date.now()) + const originalFilterPrunable = repo.filterPrunableVectorRefs.bind(repo) + const deleteByMemoryIds = vi.spyOn(store, 'deleteByMemoryIds') + vi.spyOn(repo, 'filterPrunableVectorRefs').mockImplementation( + (agentId, ids, embeddingDim, embeddingModel) => { + if (ids.includes(id)) { + repo.updateStatus(id, 'embedded', { + embeddingId: id, + embeddingDim, + embeddingModel + }) + store.vectors.set(id, textToVector('user likes redis restored')) + } + return originalFilterPrunable(agentId, ids, embeddingDim, embeddingModel) + } + ) + + await presenter.runConsolidationPass('a', 1_000 * DAY) + + expect(deleteByMemoryIds).not.toHaveBeenCalled() + expect(repo.getById(id)).toMatchObject({ + status: 'embedded', + embedding_id: id, + embedding_dim: 4, + embedding_model: 'p:m' + }) + expect(store.vectors.has(id)).toBe(true) + }) + + it('does not let stale-dimension refs starve current-dimension vector pruning', async () => { + const generateText = routedLLM({}) + const { presenter, repo, store } = makeLLMPresenter(generateText) + const liveId = await seedEmbedded(presenter, 'live current vector') + repo.setAnchor(liveId, true) + for (let index = 0; index < 260; index += 1) { + const id = `old-${index}` + repo.insert({ + id, + agentId: 'a', + kind: 'semantic', + content: `old archived ${index}`, + status: 'embedded', + createdAt: index + }) + repo.updateStatus(id, 'embedded', { + embeddingId: id, + embeddingDim: 8, + embeddingModel: 'p:m' + }) + repo.archive(id, Date.now()) + store.vectors.set(id, textToVector(`old archived ${index}`)) + } + for (let index = 0; index < 2; index += 1) { + const id = `current-${index}` + repo.insert({ + id, + agentId: 'a', + kind: 'semantic', + content: `current archived ${index}`, + status: 'embedded', + createdAt: 1000 + index + }) + repo.updateStatus(id, 'embedded', { + embeddingId: id, + embeddingDim: 4, + embeddingModel: 'p:m' + }) + repo.archive(id, Date.now()) + store.vectors.set(id, textToVector(`current archived ${index}`)) + } + + await presenter.runConsolidationPass('a', 1_000 * DAY) + + expect(store.vectors.has('current-0')).toBe(false) + expect(store.vectors.has('current-1')).toBe(false) + expect(repo.getById('current-0')).toMatchObject({ + embedding_id: null, + embedding_dim: null, + embedding_model: null + }) + expect(repo.getById('old-0')).toMatchObject({ + embedding_id: 'old-0', + embedding_dim: 8, + embedding_model: 'p:m' + }) + expect(store.vectors.has('old-0')).toBe(true) + expect(repo.getById(liveId)?.status).toBe('embedded') + }) + + it('surfaces guarded prune filter failures instead of treating them as no-op deletes', async () => { + const generateText = routedLLM({}) + const { presenter, repo, store } = makeLLMPresenter(generateText) + const id = await seedEmbedded(presenter, 'user likes redis') + repo.archive(id, Date.now()) + vi.spyOn(repo, 'filterPrunableVectorRefs').mockImplementation(() => { + throw new Error('filter failed') + }) + + await expect(presenter.runConsolidationPass('a', 1_000 * DAY)).rejects.toThrow('filter failed') + + expect(repo.getById(id)?.embedding_id).toBe(id) + expect(store.vectors.has(id)).toBe(true) + }) + + it('restores and re-embeds a memory after maintenance prunes its archived vector', async () => { + const generateText = routedLLM({}) + const { presenter, repo, store } = makeLLMPresenter(generateText) + const id = await seedEmbedded(presenter, 'user likes redis') + repo.archive(id, Date.now()) + await presenter.runConsolidationPass('a', 1_000 * DAY) + expect(store.vectors.has(id)).toBe(false) + expect(repo.getById(id)?.embedding_id).toBeNull() + + expect(presenter.restoreMemory('a', id)).toBe(true) + await presenter.processPendingEmbeddings('a') + + expect(repo.getById(id)).toMatchObject({ + status: 'embedded', + embedding_id: id, + embedding_dim: 4, + embedding_model: 'p:m' + }) + expect(store.vectors.has(id)).toBe(true) + expect( + (await presenter.recall('a', 'redis')).find((item) => item.id === id)?.sources?.vec + ).toBe(true) + }) + it('getByIds returns owned memories in input order without status filtering', () => { const { presenter, repo } = makePresenter(enabledConfig) const ids = presenter.writeMemoriesSync( @@ -2313,6 +2595,135 @@ describe('MemoryPresenter management', () => { expect(presenter.restoreMemory('a', id)).toBe(true) expect(repo.getById(id)?.status).toBe('pending_embedding') }) + + it('refuses generic archive/restore/forget for persona and working rows without audit writes', async () => { + const { presenter, repo, auditRepo } = makePresenter(enabledConfig) + repo.insert({ + id: 'persona', + agentId: 'a', + kind: 'persona', + content: 'active self model', + status: 'archived', + personaState: 'active' + }) + repo.insert({ + id: 'working', + agentId: 'a', + kind: 'working', + content: 'working blob', + status: 'fts_only' + }) + + expect(presenter.listMemories('a').map((row) => row.id)).not.toEqual( + expect.arrayContaining(['persona', 'working']) + ) + expect(presenter.getByIds('a', ['persona', 'working']).map((row) => row.id)).toEqual([ + 'persona', + 'working' + ]) + await expect(presenter.forgetMemory('a', 'persona')).resolves.toBe(false) + await expect(presenter.archiveUserMemory('a', 'persona')).resolves.toBe(false) + expect(presenter.restoreMemory('a', 'persona')).toBe(false) + await expect(presenter.forgetMemory('a', 'working')).resolves.toBe(false) + + expect(repo.getById('persona')).toMatchObject({ + status: 'archived', + persona_state: 'active' + }) + expect(repo.getById('working')?.status).toBe('fts_only') + expect(auditRepo.listByAgent('a', { eventType: 'memory/archive' })).toHaveLength(0) + }) + + it('prunes legacy persona and working vectors before repairing their statuses', async () => { + const { presenter, repo, auditRepo, store } = makePresenter(enabledConfig) + presenter.writeMemoriesSync([{ kind: 'semantic', content: 'live current memory' }], { + agentId: 'a' + }) + await presenter.processPendingEmbeddings('a') + repo.insert({ + id: 'persona', + agentId: 'a', + kind: 'persona', + content: 'active self model', + status: 'fts_only', + personaState: 'active' + }) + repo.updateStatus('persona', 'pending_embedding', { + embeddingId: 'persona', + embeddingDim: 4, + embeddingModel: 'p:m' + }) + repo.insert({ + id: 'working', + agentId: 'a', + kind: 'working', + content: 'working blob', + status: 'embedded' + }) + repo.updateStatus('working', 'embedded', { + embeddingId: 'working', + embeddingDim: 4, + embeddingModel: 'p:m' + }) + await store.upsert([ + { memoryId: 'persona', embedding: textToVector('active self model') }, + { memoryId: 'working', embedding: textToVector('working blob') } + ]) + + await presenter.runConsolidationPass('a', 1_000 * DAY) + + expect(repo.getById('persona')).toMatchObject({ + status: 'fts_only', + embedding_id: null, + embedding_dim: null, + embedding_model: null + }) + expect(repo.getById('working')).toMatchObject({ + status: 'fts_only', + embedding_id: null, + embedding_dim: null, + embedding_model: null + }) + expect(store.vectors.has('persona')).toBe(false) + expect(store.vectors.has('working')).toBe(false) + expect(presenter.getHealth('a').embeddings.pending).toBe(0) + expect(presenter.getStatus('a').pendingEmbedding).toBe(0) + expect(auditRepo.listByAgent('a', { eventType: 'memory/repair' })).toHaveLength(1) + }) + + it('keeps internal-kind embedding refs when vector prune fails', async () => { + const { presenter, repo, auditRepo, store } = makePresenter(enabledConfig) + presenter.writeMemoriesSync([{ kind: 'semantic', content: 'live current memory' }], { + agentId: 'a' + }) + await presenter.processPendingEmbeddings('a') + vi.spyOn(store, 'deleteByMemoryIds').mockRejectedValue(new Error('delete failed')) + repo.insert({ + id: 'persona', + agentId: 'a', + kind: 'persona', + content: 'active self model', + status: 'fts_only', + personaState: 'active' + }) + repo.updateStatus('persona', 'pending_embedding', { + embeddingId: 'persona', + embeddingDim: 4, + embeddingModel: 'p:m' + }) + await store.upsert([{ memoryId: 'persona', embedding: textToVector('active self model') }]) + + await presenter.runConsolidationPass('a', 1_000 * DAY) + + expect(repo.getById('persona')).toMatchObject({ + status: 'fts_only', + embedding_id: 'persona', + embedding_dim: 4, + embedding_model: 'p:m' + }) + expect(store.vectors.has('persona')).toBe(true) + expect(auditRepo.listByAgent('a', { eventType: 'memory/repair' })).toHaveLength(1) + }) }) describe('MemoryPresenter.processPendingEmbeddings (batch + fairness)', () => { @@ -2492,6 +2903,7 @@ describe('MemoryPresenter.processPendingEmbeddings (batch + fairness)', () => { }) ), query: async () => [], + queryByMemoryId: async () => [], deleteByMemoryIds: async () => {}, close: async () => {}, isUsable: () => true @@ -2527,6 +2939,7 @@ describe('MemoryPresenter.processPendingEmbeddings (batch + fairness)', () => { throw new Error('INSERT failed') }), query: async () => [], + queryByMemoryId: async () => [], deleteByMemoryIds: async () => {}, close: async () => {}, isUsable: () => true @@ -2569,6 +2982,7 @@ describe('MemoryPresenter.processPendingEmbeddings (batch + fairness)', () => { const unusableStore: IMemoryVectorStore = { upsert: async () => {}, query: async () => [], + queryByMemoryId: async () => [], deleteByMemoryIds: async () => {}, close: async () => {}, isUsable: () => false @@ -3522,6 +3936,7 @@ describe('MemoryPresenter embedding reindex (T5, AC-3.x)', () => { const unusable: IMemoryVectorStore = { upsert: async () => {}, query: async () => [], + queryByMemoryId: async () => [], deleteByMemoryIds: async () => {}, close: async () => {}, isUsable: () => false @@ -3941,12 +4356,13 @@ describe('MemoryPresenter decision ring (T-A1..T-A5)', () => { seedConflicted(repo, 'c1', targetId, 'user prefers valkey') await presenter.runConsolidationPass('a', now) + await presenter.processPendingEmbeddings('a') expect(repo.getById('c1')?.content).toBe('user prefers valkey over redis') expect(repo.getById('c1')?.provenance_key).toBe( buildMemoryProvenanceKey('a', 'semantic', 'user prefers valkey over redis') ) - expect(repo.getById('c1')?.status).toBe('pending_embedding') + expect(repo.getById('c1')?.status).toBe('embedded') expect(repo.getById(targetId)?.status).toBe('archived') }) @@ -4324,15 +4740,51 @@ describe('MemoryPresenter offline consolidation (T-B4..T-B6)', () => { { memoryId: 'old', embedding: textToVector('alpha redis habit') }, { memoryId: 'new', embedding: textToVector('beta redis habit') } ]) + getEmbeddings.mockClear() await presenter.runConsolidationPass('a', now) expect(createVectorStore).toHaveBeenCalledTimes(1) - expect(getEmbeddings).toHaveBeenCalledWith('p', 'm', ['alpha redis habit']) + expect(getEmbeddings.mock.calls.map((call) => call[2])).not.toContainEqual([ + 'alpha redis habit' + ]) + expect(getEmbeddings.mock.calls.map((call) => call[2])).not.toContainEqual(['beta redis habit']) expect(repo.listByAgent('a')).toHaveLength(1) expect(repo.getById('old')?.superseded_by).toBe('new') }) + it('bounds stored-vector consolidation scans and resumes with a compound cursor', async () => { + const generateText = routedLLM({ + decision: '{"decision":"ADD","targetIndex":null,"mergedContent":null}' + }) + const { presenter, repo, store } = makeLLMPresenter(generateText) + const now = 1_000 * DAY + const queryByMemoryId = vi.spyOn(store, 'queryByMemoryId').mockResolvedValue([]) + for (let i = 0; i < 70; i += 1) { + const id = `m-${String(i).padStart(2, '0')}` + repo.insert({ + id, + agentId: 'a', + kind: 'semantic', + content: `memory ${i}`, + status: 'embedded', + createdAt: now - 1000 + }) + repo.updateStatus(id, 'embedded', { + embeddingId: id, + embeddingDim: 4, + embeddingModel: 'p:m' + }) + } + + await presenter.runConsolidationPass('a', now) + expect(queryByMemoryId).toHaveBeenCalledTimes(64) + expect(decisionCalls(generateText)).toBe(0) + + await presenter.runConsolidationPass('a', now + 6 * 60 * 60 * 1000 + 1) + expect(queryByMemoryId).toHaveBeenCalledTimes(70) + }) + it('respects the cooldown: a second pass within the window does no LLM work (T-B5)', async () => { const generateText = routedLLM({ decision: '{"decision":"NOOP","targetIndex":0,"mergedContent":null}' @@ -5707,6 +6159,7 @@ describe('MemoryPresenter lifecycle revival (SDD-8)', () => { resolveQuery = () => resolve([{ memoryId: 'm1', distance: 0.01 }]) }) }), + queryByMemoryId: async () => [], deleteByMemoryIds: async () => {}, close: async () => {}, isUsable: () => true diff --git a/test/main/presenter/memorySearch.test.ts b/test/main/presenter/memorySearch.test.ts index 1ad8037a8..2f8e9d37d 100644 --- a/test/main/presenter/memorySearch.test.ts +++ b/test/main/presenter/memorySearch.test.ts @@ -13,6 +13,12 @@ function seed(content: string, id: string) { } } +function seedRedisRange(repo: { insert(input: ReturnType): unknown }, count: number) { + for (let i = 0; i < count; i += 1) { + repo.insert(seed(`redis memory ${String(i).padStart(2, '0')}`, `m${i}`)) + } +} + describe('MemoryPresenter.searchMemories (read-only facade)', () => { it('surfaces matching rows with their retrieval score', async () => { const { presenter, repo } = makePresenter(enabledConfig) @@ -49,17 +55,60 @@ describe('MemoryPresenter.searchMemories (read-only facade)', () => { ).toEqual(['m1']) }) - it('caps the result count to limit without widening topK', async () => { + it('uses search limit as retrieval depth without changing recall topK', async () => { const { presenter, repo } = makePresenter(enabledConfig) - repo.insert(seed('redis caching notes', 'm1')) - repo.insert(seed('redis cluster setup', 'm2')) - repo.insert(seed('redis persistence tuning', 'm3')) + seedRedisRange(repo, 20) - const limited = await presenter.searchMemories('deepchat', 'redis', { limit: 2 }) - expect(limited).toHaveLength(2) + const limited = await presenter.searchMemories('deepchat', 'redis', { limit: 8 }) + expect(limited).toHaveLength(8) + + const recall = await presenter.recall('deepchat', 'redis') + expect(recall).toHaveLength(6) + }) + + it('defaults management search depth to 50 and clamps direct callers at 100', async () => { + const { presenter, repo } = makePresenter(enabledConfig) + seedRedisRange(repo, 120) + + await expect(presenter.searchMemories('deepchat', 'redis')).resolves.toHaveLength(50) + await expect( + presenter.searchMemories('deepchat', 'redis', { limit: 500 }) + ).resolves.toHaveLength(100) + }) + + it('does not let persona rows occupy management search result slots', async () => { + const { presenter, repo } = makePresenter(enabledConfig) + for (let i = 0; i < 20; i += 1) { + repo.insert({ + id: `persona-${i}`, + agentId: 'deepchat', + kind: 'persona', + content: `redis persona ${i}`, + status: 'fts_only' + }) + } + seedRedisRange(repo, 3) + + const hits = await presenter.searchMemories('deepchat', 'redis', { limit: 3 }) + + expect(hits.map((hit) => hit.row.id)).toEqual(['m0', 'm1', 'm2']) + }) + + it('does not prune vectors from the read-only management search path', async () => { + const { presenter, repo, store } = makePresenter(enabledConfig) + const [id] = presenter.writeMemoriesSync([{ kind: 'semantic', content: 'redis cache' }], { + agentId: 'deepchat' + }) + await presenter.processPendingEmbeddings('deepchat') + repo.archive(id, Date.now()) + const filterPrunable = vi.spyOn(repo, 'filterPrunableVectorRefs') + const deleteByMemoryIds = vi.spyOn(store, 'deleteByMemoryIds') + + await presenter.searchMemories('deepchat', 'redis') - const all = await presenter.searchMemories('deepchat', 'redis') - expect(all.length).toBeGreaterThanOrEqual(3) + expect(filterPrunable).not.toHaveBeenCalled() + expect(deleteByMemoryIds).not.toHaveBeenCalled() + expect(store.vectors.has(id)).toBe(true) }) it('returns nothing for an empty query', async () => { diff --git a/test/main/presenter/memoryVectorStore.test.ts b/test/main/presenter/memoryVectorStore.test.ts index 06c19ba48..6b333d582 100644 --- a/test/main/presenter/memoryVectorStore.test.ts +++ b/test/main/presenter/memoryVectorStore.test.ts @@ -25,6 +25,16 @@ interface TestStore { upsert(records: MemoryVectorRecord[]): Promise } +interface QueryableStore { + connection: { runAndReadAll: ReturnType } + vectorTable: string + query: ReturnType + queryByMemoryId( + memoryId: string, + options: { topK: number } + ): Promise> +} + function makeStore(onRun: (sql: string) => void = () => {}) { const calls: string[] = [] const connection = { @@ -66,6 +76,57 @@ describe('MemoryVectorStore.upsert transaction (C4, AC-4.2)', () => { }) }) +describe('MemoryVectorStore.queryByMemoryId', () => { + it('reads the stored source vector, reuses parameterized query, and excludes itself', async () => { + const connection = { + runAndReadAll: vi.fn(async () => ({ + getRowObjectsJson: () => [{ embedding: [0.1, 0.2] }] + })) + } + const store = Object.create(MemoryVectorStore.prototype) as QueryableStore + store.connection = connection + store.vectorTable = 'memory_vector' + store.query = vi.fn(async () => [ + { memoryId: 'm1', distance: 0 }, + { memoryId: 'm2', distance: 0.12 }, + { memoryId: 'm3', distance: 0.2 } + ]) + + const matches = await store.queryByMemoryId('m1', { topK: 2 }) + + expect(matches).toEqual([ + { memoryId: 'm2', distance: 0.12 }, + { memoryId: 'm3', distance: 0.2 } + ]) + expect(connection.runAndReadAll).toHaveBeenCalledWith( + expect.stringContaining('SELECT embedding'), + ['m1'] + ) + expect(store.query).toHaveBeenCalledWith([0.1, 0.2], { topK: 3 }) + }) + + it('returns no neighbors when the source vector is missing or malformed', async () => { + const connection = { + runAndReadAll: vi.fn(async () => ({ + getRowObjectsJson: () => [] + })) + } + const store = Object.create(MemoryVectorStore.prototype) as QueryableStore + store.connection = connection + store.vectorTable = 'memory_vector' + store.query = vi.fn(async () => []) + + await expect(store.queryByMemoryId('missing', { topK: 2 })).resolves.toEqual([]) + expect(store.query).not.toHaveBeenCalled() + + connection.runAndReadAll.mockResolvedValueOnce({ + getRowObjectsJson: () => [{ embedding: ['bad'] }] + }) + await expect(store.queryByMemoryId('bad', { topK: 2 })).resolves.toEqual([]) + expect(store.query).not.toHaveBeenCalled() + }) +}) + interface EmbeddingMeta { provider: string model: string diff --git a/test/main/routes/memoryDto.test.ts b/test/main/routes/memoryDto.test.ts index ac5c73e81..9be698ad9 100644 --- a/test/main/routes/memoryDto.test.ts +++ b/test/main/routes/memoryDto.test.ts @@ -361,6 +361,12 @@ describe('memory.search route contract', () => { expect( memorySearchRoute.input.parse({ agentId: 'deepchat', query: 'redis', limit: 5 }).limit ).toBe(5) + expect( + memorySearchRoute.input.parse({ agentId: 'deepchat', query: 'redis', limit: 100 }).limit + ).toBe(100) + expect( + memorySearchRoute.input.safeParse({ agentId: 'deepchat', query: 'redis', limit: 101 }).success + ).toBe(false) expect(memorySearchRoute.input.safeParse({ agentId: 'has space', query: 'x' }).success).toBe( false )