diff --git a/docs/architecture/memory-system.md b/docs/architecture/memory-system.md index 978aa14cc..bd59bb2c3 100644 --- a/docs/architecture/memory-system.md +++ b/docs/architecture/memory-system.md @@ -258,7 +258,7 @@ Maintenance 只处理有界 seed batch 和有界 same-scope vector neighbors； ## Maintenance 和可观测性 -`MaintenanceService` 拥有 timer、cooldown、并发预算和 stop/drain；`MergeService` 只负责有界 +`MaintenanceService` 拥有 timer、cooldown 和并发预算；`MemoryService` 统一暂停与排空；`MergeService` 只负责有界 near-duplicate merge，沿用 runner 传入的 operation fence、业务时间和共享预算。用户 conflict resolution 的后续调度由 facade 负责；自动 challenge pass 每次成功应用后通知 Maintenance 调度， 即使后续 pair 失败也不丢失已经产生的调度。`ConflictService` 不持有 Maintenance 的构造依赖。 @@ -268,9 +268,12 @@ fence Memory、drain accepted work、关闭 store/SQLite、执行操作、reopen `stopBackgroundMaintenance` 同步清空全部 prewarm/startup/consolidation timer、拒绝新的 arm 与 pass， 并对每个持有 in-flight pass 的 Agent 推进 execution fence、中止其 provider 请求，让 pass 及其委托的 challenge/merge/reflection/persona 子步骤在下一个 checkpoint 停止，而不是等完一个 provider deadline； -`drainBackgroundMaintenance` 在有界超时内等待这些 pass 落定，超时即让 database maintenance 失败而不是 -带着未落定的 pass 关闭 SQLite。`startBackgroundMaintenance` 在 stop 之后可以重新 arm，startup pass -不会因此丢失。 +`drainBackgroundMaintenance` 在一个有界超时内等待 consolidation、embedding/prewarm 和 clear work +落定，超时即让 database maintenance 失败而不是带着未落定的任务关闭 SQLite。暂停期间 dirty working +refresh 不访问数据库、不删除 projection，恢复后重新调度。Clear 在批次或 await 边界暂停，保留 durable +job 并拒绝未完成的请求；如果 drain 超时使数据库维护取消，恢复原数据库后仍在途的 clear 可以继续完成。 +`startBackgroundMaintenance` 在 stop 之后恢复 admission、dirty refresh 和 pending clear，并重新 arm +startup pass。 启动恢复按 Agent 顺序处理 pending clear job，避免多个遗留 namespace 在同一个 event-loop tick 同时执行首批同步事务。Shutdown 只等待当前有界 batch；未完成 job 保持可恢复。 diff --git a/docs/issues/memory-maintenance-drain/spec.md b/docs/issues/memory-maintenance-drain/spec.md new file mode 100644 index 000000000..2016db692 --- /dev/null +++ b/docs/issues/memory-maintenance-drain/spec.md @@ -0,0 +1,43 @@ +# Memory database-maintenance drain + +## Cause and scope + +The application closes SQLite after Memory's stop/drain handshake. That handshake only tracks +consolidation passes: started startup prewarms and durable clears can still access the repository +after drain reports success. Controlled async probes reproduced both continuations. + +## Design + +Keep the existing MemoryService boundary. Pause Memory admission in the runtime context before +stopping timers, invalidate existing execution fences and abort provider work. Drain accepted +consolidation, embedding and clear work under the existing shared timeout. Report pending Agent +IDs on timeout so the application refuses to close SQLite. New memory routes must be blocked +during the application's database-maintenance window, just like session routes. + +Clear stops at a batch/await boundary with its durable job intact and rejects the interrupted +request rather than claiming completion. Resume admission and pending clears only after database +reopen. Preserve shutdown behavior, schema and public route contracts. Do not split agentMemory.ts +or introduce another scheduler, setting or dependency. + +## Acceptance + +- [x] Started prewarm cannot access SQLite after a successful drain. +- [x] Pending clear pauses durably and resumes after reopening. +- [x] Unsettled work is reported at the deadline; resume does not revive old execution fences. +- [x] Review P1/P2/P3, ablate unnecessary design, and run relevant and combined verification. + +## Validation + +Six maintenance regressions cover prewarm, clear batch boundaries, replacement databases, +blocked vector reset, early resume after a failed drain, and dirty working projections. All six +fail against the unchanged production baseline. Combined validation with the scoped retrieval +fix passes format, i18n, lint, application and Memory test typechecks, 951 behavior tests, +334 native tests (two skipped), 10 performance tests and seven evaluation tests. + +Review found and fixed dirty projection access during pause. Simplification removes the separate +maintenance pause flag and drain implementation; accepted task maps and the shared deadline helper +remain the source of truth. Clear is not tied to model/config execution generations: a failed drain +prevents database replacement, so resuming the unchanged database may finish the accepted clear. +Recovery reloads pending clear fences from the reopened database rather than retaining stale jobs. + +No push or GitHub issue sync is authorized. diff --git a/src/main/app/composition.ts b/src/main/app/composition.ts index db6cb5534..4a5d1e79b 100644 --- a/src/main/app/composition.ts +++ b/src/main/app/composition.ts @@ -3361,6 +3361,7 @@ export async function createMainProcessControl(dependencies: { if ( routeName.startsWith('chat.') || routeName.startsWith('sessions.') || + routeName.startsWith('memory.') || routeName.startsWith('orchestration.') || routeName.startsWith('remoteControl.') || routeName.startsWith('cronJobs.') diff --git a/src/main/memory/context.ts b/src/main/memory/context.ts index 9c5be0cdb..0ccd68e92 100644 --- a/src/main/memory/context.ts +++ b/src/main/memory/context.ts @@ -58,6 +58,7 @@ export function isUniqueConstraintError(error: unknown): boolean { export class MemoryRuntimeContext { private disposed = false + private paused = false private readonly readEpochByAgent = new Map() private readonly executionStateByAgent = new Map() private readonly pendingMemoryClearAgentIds: Set @@ -72,6 +73,21 @@ export class MemoryRuntimeContext { return this.disposed } + get isPaused(): boolean { + return this.paused + } + + pause(): void { + this.paused = true + for (const agentId of this.executionStateByAgent.keys()) { + this.invalidateAgentOperations(agentId) + } + } + + resume(): void { + this.paused = false + } + now(): number { const now = this.clock.now() if (!Number.isFinite(now)) { @@ -102,6 +118,7 @@ export class MemoryRuntimeContext { isOperationFenceCurrent(fence: MemoryOperationFence): boolean { return ( !this.disposed && + !this.paused && (this.executionStateByAgent.get(fence.agentId)?.generation ?? 0) === fence.generation ) } @@ -200,6 +217,11 @@ export class MemoryRuntimeContext { return this.pendingMemoryClearAgentIds.has(agentId) } + syncPendingMemoryClears(agentIds: readonly string[]): void { + this.pendingMemoryClearAgentIds.clear() + for (const agentId of agentIds) this.pendingMemoryClearAgentIds.add(agentId) + } + markMemoryClearPending(agentId: string): void { this.pendingMemoryClearAgentIds.add(agentId) } @@ -211,6 +233,7 @@ export class MemoryRuntimeContext { canWriteAgentMemory(agentId: string): boolean { return ( !this.disposed && + !this.paused && !this.isMemoryClearPending(agentId) && this.isManagedAgent(agentId) && this.isEnabled(agentId) @@ -218,7 +241,7 @@ export class MemoryRuntimeContext { } canManageAgentMemory(agentId: string): boolean { - return !this.disposed && this.isManagedAgent(agentId) + return !this.disposed && !this.paused && this.isManagedAgent(agentId) } canManageClaimMemory(agentId: string): boolean { @@ -226,7 +249,7 @@ export class MemoryRuntimeContext { } canReadDirectivePlane(agentId: string): boolean { - return !this.disposed && this.isManagedAgent(agentId) && this.isEnabled(agentId) + return this.canManageAgentMemory(agentId) && this.isEnabled(agentId) } canReadAgentMemory(agentId: string): boolean { @@ -238,16 +261,13 @@ export class MemoryRuntimeContext { } canUseCurrentMemoryEmbedding(agentId: string, embedding: MemoryModelRef): boolean { + if (!this.canReadAgentMemory(agentId)) return false const current = this.options.policy.resolveAgentConfig(agentId)?.memoryEmbedding - return ( - current?.providerId === embedding.providerId && - current?.modelId === embedding.modelId && - this.canReadAgentMemory(agentId) - ) + return current?.providerId === embedding.providerId && current?.modelId === embedding.modelId } emitChanged(agentId: string, reason: MemoryUpdateReason, context?: MemoryUpdateContext): void { - if (this.disposed) return + if (this.disposed || this.paused) return this.options.onAgentMemoryMutated?.(agentId) if (context) this.options.changeSink?.onMemoryChanged?.(agentId, reason, context) else this.options.changeSink?.onMemoryChanged?.(agentId, reason) @@ -267,7 +287,7 @@ export class MemoryRuntimeContext { createdAt?: number } ): void { - if (this.disposed) return + if (this.disposed || this.paused) return this.options.auditWriter?.insert({ id: `audit-${nanoid(12)}`, agentId, diff --git a/src/main/memory/index.ts b/src/main/memory/index.ts index b9cec2d73..86e81f163 100644 --- a/src/main/memory/index.ts +++ b/src/main/memory/index.ts @@ -47,7 +47,8 @@ import type { MemoryPersonaDraftResult, MemoryReflectionResult } from './types' -import { REINDEX_MAX_BATCHES } from './runtimeConstants' +import { MAINTENANCE_DRAIN_TIMEOUT_MS, REINDEX_MAX_BATCHES } from './runtimeConstants' +import { withSoftDeadline } from './core/asyncDeadline' import { MemoryRuntimeContext } from './context' import { MemoryRowMutations } from './services/rowMutations' import { VectorStoreManager } from './infra/vectorStoreManager' @@ -375,6 +376,9 @@ export class MemoryService implements MemoryRuntimePort { } startBackgroundMaintenance(): void { + if (this.runtime.isDisposed) return + this.runtime.resume() + this.workingMemory.resumeDirtyRefreshes() void this.management.resumePendingMemoryClears().catch((error) => { logger.error(`[Memory] pending clear recovery failed: ${String(error)}`) }) @@ -382,11 +386,35 @@ export class MemoryService implements MemoryRuntimePort { } stopBackgroundMaintenance(): void { + this.runtime.pause() this.maintenance.stopBackgroundMaintenance() } - drainBackgroundMaintenance(timeoutMs?: number): Promise { - return this.maintenance.drainBackgroundMaintenance(timeoutMs) + async drainBackgroundMaintenance( + timeoutMs: number = MAINTENANCE_DRAIN_TIMEOUT_MS + ): Promise { + const deadline = performance.now() + timeoutMs + while (true) { + const pending = [ + ...this.maintenance.getInFlight(), + ...this.embedding.getInFlight(), + ...this.management.getInFlightMemoryClears() + ] + if (!pending.length) return [] + const result = await withSoftDeadline( + Promise.allSettled(pending), + Math.max(0, deadline - performance.now()) + ) + if (result.timedOut) { + return [ + ...new Set([ + ...this.maintenance.getInFlightAgentIds(), + ...this.embedding.getInFlightAgentIds(), + ...this.management.getInFlightClearAgentIds() + ]) + ].sort() + } + } } warmActiveAgents(): void { diff --git a/src/main/memory/infra/embeddingPipeline.ts b/src/main/memory/infra/embeddingPipeline.ts index 3ba9d101b..d47c0e943 100644 --- a/src/main/memory/infra/embeddingPipeline.ts +++ b/src/main/memory/infra/embeddingPipeline.ts @@ -1279,6 +1279,18 @@ export class EmbeddingPipeline { this.embeddingWarmups.set(key, tracked) } + getInFlightAgentIds(): string[] { + return [ + ...new Set([ + ...this.reindexing.keys(), + ...this.backfilling.keys(), + ...this.embeddingDrains.keys(), + ...[...this.vectorStoreWarmups.keys()].map((key) => key.split('::')[0]), + ...[...this.embeddingWarmupAgents.values()].flatMap((agents) => [...agents]) + ]) + ] + } + getInFlight(): Promise[] { return [ ...this.reindexing.values(), diff --git a/src/main/memory/services/maintenanceService.ts b/src/main/memory/services/maintenanceService.ts index 0ca3a7a1d..eb1d59636 100644 --- a/src/main/memory/services/maintenanceService.ts +++ b/src/main/memory/services/maintenanceService.ts @@ -8,7 +8,6 @@ import { CONSOLIDATION_DIRTY_SEED_LIMIT, CONSOLIDATION_FAILURE_COOLDOWN_MS, CONSOLIDATION_IDLE_MS, - MAINTENANCE_DRAIN_TIMEOUT_MS, MAINTENANCE_HEAVY_MAX_CONCURRENCY, MAINTENANCE_START_DELAY_MS, STARTUP_ARM_STAGGER_MS, @@ -67,7 +66,6 @@ export class MaintenanceService { private prewarmStartTimer: NodeJS.Timeout | null = null private readonly prewarmTimers = new Map() private maintenanceStarted = false - private maintenancePaused = false // Heavy passes run in this order under one shared budget; each is fenced independently so a // stop request lands at the next boundary instead of after the whole sequence. @@ -205,7 +203,6 @@ export class MaintenanceService { startBackgroundMaintenance(): void { if (this.ctx.isDisposed || this.maintenanceStarted) return this.maintenanceStarted = true - this.maintenancePaused = false this.prewarmStartTimer = setTimeout(() => { this.prewarmStartTimer = null if (this.ctx.isDisposed) return @@ -226,12 +223,11 @@ export class MaintenanceService { * fence invalidated and its provider requests aborted, so the pass and the * sub-services it delegates to stop at their next checkpoint instead of * waiting out a provider deadline. `startBackgroundMaintenance` re-arms after - * the caller's maintenance window; `drainBackgroundMaintenance` waits for the - * fenced passes to settle. + * the caller's maintenance window. MemoryService drains these passes together + * with embedding and clear work before the database can close. */ stopBackgroundMaintenance(): void { this.maintenanceStarted = false - this.maintenancePaused = true if (this.prewarmStartTimer) { clearTimeout(this.prewarmStartTimer) this.prewarmStartTimer = null @@ -250,22 +246,6 @@ export class MaintenanceService { } } - /** Waits for in-flight passes and returns the agents whose pass is still running. */ - async drainBackgroundMaintenance( - timeoutMs: number = MAINTENANCE_DRAIN_TIMEOUT_MS - ): Promise { - let timer: ReturnType | undefined - await Promise.race([ - Promise.allSettled(this.consolidationPasses.values()), - new Promise((resolve) => { - timer = setTimeout(resolve, timeoutMs) - if (typeof timer.unref === 'function') timer.unref() - }) - ]) - if (timer) clearTimeout(timer) - return [...this.consolidationPasses.keys()].sort() - } - prepareDispose(): void { this.stopBackgroundMaintenance() this.lastConsolidationAt.clear() @@ -359,7 +339,7 @@ export class MaintenanceService { delayMs: number = CONSOLIDATION_IDLE_MS, options: { preserveEarlier?: boolean } = {} ): void { - if (this.ctx.isDisposed || this.maintenancePaused) return + if (this.ctx.isDisposed || this.ctx.isPaused) return const dueAt = Date.now() + delayMs const existing = this.consolidationTimers.get(agentId) const existingDueAt = this.consolidationTimerDueAt.get(agentId) @@ -389,7 +369,7 @@ export class MaintenanceService { const effectiveNow = now ?? this.ctx.now() const existing = this.consolidationPasses.get(agentId) if (existing) return existing - if (this.maintenancePaused) return + if (this.ctx.isPaused) return const tracked = this.runConsolidationPassInternal(agentId, effectiveNow).finally(() => { if (this.consolidationPasses.get(agentId) === tracked) { this.consolidationPasses.delete(agentId) @@ -643,6 +623,10 @@ export class MaintenanceService { return [...this.consolidationPasses.values()] } + getInFlightAgentIds(): string[] { + return [...this.consolidationPasses.keys()] + } + clearInFlight(): void { this.consolidationPasses.clear() } diff --git a/src/main/memory/services/managementService.ts b/src/main/memory/services/managementService.ts index 2a8eef2fb..0cdf71314 100644 --- a/src/main/memory/services/managementService.ts +++ b/src/main/memory/services/managementService.ts @@ -167,6 +167,9 @@ export class ManagementService { } private enqueueMemoryClear(agentId: string): Promise { + if (this.ctx.isPaused) { + return Promise.reject(new Error('[Memory] clear paused for database maintenance')) + } const existing = this.clearOperations.get(agentId) if (existing) return existing const operation = this.runMemoryClear(agentId) @@ -184,9 +187,16 @@ export class ManagementService { return [...this.clearOperations.values()] } + getInFlightClearAgentIds(): string[] { + return [...this.clearOperations.keys()] + } + async resumePendingMemoryClears(): Promise { const jobs = this.ports.repository.listPendingMemoryClearJobs() + // Database maintenance may replace SQLite rather than reopen the same file. + this.ctx.syncPendingMemoryClears(jobs.map((job) => job.agentId)) for (const job of jobs) { + if (this.ctx.isPaused || this.ctx.isDisposed) return if (!isSafeAgentId(job.agentId)) { logger.error(`[Memory] refusing to resume clear job with invalid agent id: ${job.agentId}`) continue @@ -729,6 +739,7 @@ export class ManagementService { this.ports.syncWorkingMemoryAfterMutation(agentId) while (job.phase === 'claims') { + if (this.ctx.isPaused) throw new Error('[Memory] clear paused for database maintenance') const batch = this.ports.repository.processMemoryClearBatch(agentId) if (!batch) { const remaining = this.ports.repository.countByAgent(agentId) @@ -776,6 +787,7 @@ export class ManagementService { return { removed: job.removed, cleanupPendingRestart } } + if (this.ctx.isPaused) throw new Error('[Memory] clear paused for database maintenance') if (!this.ports.repository.completeMemoryClear(agentId)) { throw new Error(`[Memory] clear job returned to claim cleanup for ${agentId}`) } diff --git a/src/main/memory/services/retrievalService.ts b/src/main/memory/services/retrievalService.ts index 02cb2c2d3..58d5041eb 100644 --- a/src/main/memory/services/retrievalService.ts +++ b/src/main/memory/services/retrievalService.ts @@ -398,7 +398,7 @@ export class RetrievalService { currentEmbedding, dimensions, queryIndexes.map((index) => vectors[index] as number[]), - vectorCandidateLimit + MEMORY_RETRIEVAL_MAX_CANDIDATES ) latencyMs.vector = performance.now() - vectorStartedAt if ( @@ -456,7 +456,16 @@ export class RetrievalService { .filter((row): row is AgentMemoryRow => isLiveDecisionRow(agentId, row)) const currentVectorMatches: Array<{ row: AgentMemoryRow; similarity: number }> = [] if (vectorContext && vectorFingerprint) { - for (const match of vectorMatches[index]) { + // As in normal recall, scan once and widen locally when scope or row revalidation + // removes the nearest hits. Keep the initial page when it already supplies neighbors. + let limit = vectorCandidateLimit + const matches = vectorMatches[index] + for (let matchIndex = 0; matchIndex < matches.length; matchIndex += 1) { + if (matchIndex >= limit) { + if (currentVectorMatches.length >= DECISION_NEIGHBOR_TOP_S) break + limit = nextMemoryRetrievalCandidateLimit(limit) + } + const match = matches[matchIndex] const row = rowsById.get(match.memoryId) if ( isCurrentRecallVectorRow(agentId, row, vectorContext.dimensions, vectorFingerprint) && @@ -465,6 +474,12 @@ export class RetrievalService { currentVectorMatches.push({ row, similarity: match.similarity }) } } + if ( + currentVectorMatches.length < DECISION_NEIGHBOR_TOP_S && + matches.length >= MEMORY_RETRIEVAL_MAX_CANDIDATES + ) { + degradations.add('candidateBudgetExhausted') + } } const neighbors = fuse(ftsRows, currentVectorMatches, { topK: DECISION_NEIGHBOR_TOP_S, diff --git a/src/main/memory/services/workingMemoryService.ts b/src/main/memory/services/workingMemoryService.ts index d4fcea6f9..e801ae6a9 100644 --- a/src/main/memory/services/workingMemoryService.ts +++ b/src/main/memory/services/workingMemoryService.ts @@ -47,6 +47,7 @@ export class WorkingMemoryService implements WorkingMemoryReadPort { } readWorkingMemory(agentId: string): string | null { + if (this.ctx.isPaused) return null this.flushWorkingMemoryIfDirty(agentId) const row = this.resolveWorkingRow(agentId) const content = row?.content?.trim() @@ -116,6 +117,7 @@ export class WorkingMemoryService implements WorkingMemoryReadPort { const timer = this.workingRefreshTimers.get(agentId) if (timer) clearTimeout(timer) this.workingRefreshTimers.delete(agentId) + if (this.ctx.isPaused) return if (this.workingMemoryDirty.delete(agentId)) { try { if (this.ctx.canReadAgentMemory(agentId)) this.refreshWorkingMemory(agentId) @@ -135,6 +137,10 @@ export class WorkingMemoryService implements WorkingMemoryReadPort { this.scheduleDirtyRefresh(agentId) } + resumeDirtyRefreshes(): void { + for (const agentId of this.workingMemoryDirty) this.scheduleDirtyRefresh(agentId) + } + private scheduleDirtyRefresh(agentId: string): void { const existing = this.workingRefreshTimers.get(agentId) if (existing) clearTimeout(existing) diff --git a/test/main/memory/maintenanceService.test.ts b/test/main/memory/maintenanceService.test.ts index 80393d7fc..ae29d3342 100644 --- a/test/main/memory/maintenanceService.test.ts +++ b/test/main/memory/maintenanceService.test.ts @@ -1,6 +1,7 @@ import { describe, expect, it, vi } from 'vitest' import type { MaintenanceBudget } from '@/memory/core/maintenanceBudget' +import { WORKING_REFRESH_DEBOUNCE_MS } from '@/memory/runtimeConstants' import type { AgentMemoryRow, MemoryVectorMatch } from '@/memory/domain/types' import type { DeepChatAgentConfig } from '@shared/types/agent-interface' import { createControlledPromise } from './serviceHarness' @@ -1315,6 +1316,142 @@ describe('MemoryService offline consolidation (T-B4..T-B6)', () => { expect(decisionCalls(generateText)).toBe(0) }) + it('fences started prewarm and reports its unsettled store open during drain', async () => { + const repo = createFakeRepository() + repo.rows.set('m1', makeRow('m1')) + const open = createControlledPromise() + const createVectorStore = vi.fn(() => open.promise) + const presenter = new MemoryService({ + repository: repo, + resolveAgentConfig: () => enabledConfig, + getEmbeddings: async (_p, _m, texts) => texts.map(textToVector), + getDimensions: embeddingDimensions, + generateText: async () => '', + createVectorStore, + resetVectorStore: async () => undefined + }) + const token = presenter.captureExecutionToken('a') + presenter.warmActiveAgents() + await vi.waitFor(() => expect(createVectorStore).toHaveBeenCalledTimes(1)) + presenter.stopBackgroundMaintenance() + await expect(presenter.drainBackgroundMaintenance(20)).resolves.toEqual(['a']) + const staleRead = vi.spyOn(repo, 'hasStaleEmbeddings') + const store = new FakeVectorStore() + store.vectors.set('m1', textToVector('m1')) + open.resolve(store) + await expect(presenter.drainBackgroundMaintenance()).resolves.toEqual([]) + expect(staleRead).not.toHaveBeenCalled() + presenter.warmActiveAgents() + expect(presenter.isEnabled('a')).toBe(false) + presenter.startBackgroundMaintenance() + expect(presenter.isEnabled('a')).toBe(true) + expect(presenter.canContinueExecution(token)).toBe(false) + await presenter.dispose() + }) + + it.each([false, true])( + 'pauses clear between batches (replace database: %s)', + async (replaceDatabase) => { + const { presenter, repo } = makeLLMPresenter(routedLLM({})) + for (let index = 0; index < 257; index++) { + repo.rows.set(`m${index}`, makeRow(`m${index}`)) + } + const clear = presenter.clearMemories('a') + const interrupted = expect(clear).rejects.toThrow('paused for database maintenance') + presenter.stopBackgroundMaintenance() + await expect(presenter.drainBackgroundMaintenance()).resolves.toEqual([]) + await interrupted + expect(repo.countByAgent('a')).toBe(1) + expect(repo.listPendingMemoryClearJobs()).toMatchObject([{ agentId: 'a', removed: 256 }]) + await expect(presenter.clearMemories('a')).rejects.toThrow('paused for database maintenance') + + if (replaceDatabase) { + repo.retireAgentMemoryNamespace('a') + repo.rows.set('restored', makeRow('restored')) + } + presenter.startBackgroundMaintenance() + await vi.waitFor(() => expect(repo.listPendingMemoryClearJobs()).toEqual([])) + expect(repo.countByAgent('a')).toBe(replaceDatabase ? 1 : 0) + expect(presenter.isEnabled('a')).toBe(true) + await presenter.dispose() + } + ) + + it.each([false, true])( + 'waits for vector clear across pause (resume early: %s)', + async (resumeEarly) => { + const repo = createFakeRepository() + repo.rows.set('m1', makeRow('m1')) + const reset = createControlledPromise() + const resetVectorStore = vi.fn(() => reset.promise) + const presenter = new MemoryService({ + repository: repo, + resolveAgentConfig: () => enabledConfig, + getEmbeddings: async (_p, _m, texts) => texts.map(textToVector), + getDimensions: embeddingDimensions, + generateText: async () => '', + createVectorStore: async () => new FakeVectorStore(), + resetVectorStore + }) + const clear = presenter.clearMemories('a') + const outcome = clear.then( + (removed) => ({ removed }), + (error: Error) => ({ error: error.message }) + ) + await vi.waitFor(() => expect(resetVectorStore).toHaveBeenCalledTimes(1)) + presenter.stopBackgroundMaintenance() + await expect(presenter.drainBackgroundMaintenance(20)).resolves.toEqual(['a']) + const complete = vi.spyOn(repo, 'completeMemoryClear') + // A failed drain prevents database replacement. Resuming that unchanged database can + // finish the accepted clear; it must not be tied to a model/config execution fence. + if (resumeEarly) presenter.startBackgroundMaintenance() + reset.resolve() + await expect(presenter.drainBackgroundMaintenance()).resolves.toEqual([]) + if (resumeEarly) { + await expect(outcome).resolves.toEqual({ removed: 1 }) + } else { + await expect(outcome).resolves.toEqual({ + error: '[Memory] clear paused for database maintenance' + }) + expect(complete).not.toHaveBeenCalled() + expect(repo.listPendingMemoryClearJobs()).toMatchObject([ + { agentId: 'a', phase: 'vectors' } + ]) + presenter.startBackgroundMaintenance() + } + await vi.waitFor(() => expect(repo.listPendingMemoryClearJobs()).toEqual([])) + expect(complete).toHaveBeenCalledWith('a') + await presenter.dispose() + } + ) + + it('retains a dirty working projection without database access while paused', async () => { + vi.useFakeTimers() + const { presenter, repo } = makeLLMPresenter(routedLLM({}), { memoryEnabled: true }) + try { + await presenter.rememberMemory({ kind: 'semantic', content: 'first fact' }, { agentId: 'a' }) + await vi.advanceTimersByTimeAsync(WORKING_REFRESH_DEBOUNCE_MS) + const working = [...repo.rows.values()].find((row) => row.kind === 'working')! + expect(working.content).toContain('first fact') + await presenter.rememberMemory({ kind: 'semantic', content: 'second fact' }, { agentId: 'a' }) + presenter.stopBackgroundMaintenance() + await expect(presenter.drainBackgroundMaintenance()).resolves.toEqual([]) + const read = vi.spyOn(repo, 'getByProvenanceKey') + const remove = vi.spyOn(repo, 'deleteInternalMemory') + await vi.advanceTimersByTimeAsync(WORKING_REFRESH_DEBOUNCE_MS) + expect(read).not.toHaveBeenCalled() + expect(remove).not.toHaveBeenCalled() + expect(repo.rows.get(working.id)?.content).not.toContain('second fact') + + presenter.startBackgroundMaintenance() + await vi.advanceTimersByTimeAsync(WORKING_REFRESH_DEBOUNCE_MS) + expect(repo.rows.get(working.id)?.content).toContain('second fact') + } finally { + await presenter.dispose() + vi.useRealTimers() + } + }) + it('prewarms enabled active agents before the delayed maintenance arm', async () => { vi.useFakeTimers() try { diff --git a/test/main/memory/retrievalService.test.ts b/test/main/memory/retrievalService.test.ts index 9a26421a2..6f5a03d6b 100644 --- a/test/main/memory/retrievalService.test.ts +++ b/test/main/memory/retrievalService.test.ts @@ -11,12 +11,23 @@ import { resolveRetrieval, retrievalScore } from '@/memory/core/scoring' -import { MEMORY_TEMPORAL_UNCERTAIN_STATE_FACTOR } from '@/memory/core/temporal' +import { + ATEMPORAL_MEMORY_METADATA, + MEMORY_TEMPORAL_UNCERTAIN_STATE_FACTOR +} from '@/memory/core/temporal' +import { MemoryRuntimeContext, type MemoryModelRef } from '@/memory/context' +import { RetrievalService } from '@/memory/services/retrievalService' import { createMemoryProviderCapacityError } from '@/memory/core/providerCancellation' import { MEMORY_RETRIEVAL_MAX_CANDIDATES } from '@/memory/core/retrievalBudget' import { FTS_SIMILARITY_BASELINE } from '@/memory/types' import type { DeepChatAgentConfig } from '@shared/types/agent-interface' -import { enabledConfig, makePresenter, textToVector } from './support/memoryFakes' +import { + createFakeRepository, + FakeVectorStore, + enabledConfig, + makePresenter, + textToVector +} from './support/memoryFakes' import { DAY, deferred, @@ -875,6 +886,217 @@ describe('MemoryService recall + injection', () => { }) }) +describe('scope-aware decision retrieval', () => { + const scope = [{ type: 'session' as const, id: 'target-session' }] + const candidates = ['redis preference', 'vue preference'].map((content) => ({ + kind: 'semantic' as const, + category: null, + content, + importance: 0.5, + temporal: ATEMPORAL_MEMORY_METADATA + })) + + function setup(foreignCount: number) { + const repository = createFakeRepository() + const store = new FakeVectorStore() + const config = { ...enabledConfig } + const policy = { resolveAgentConfig: () => config } + const ctx = new MemoryRuntimeContext({ + policy, + providerControl: { abortAgent: vi.fn(), abortAll: vi.fn() } + }) + for (let index = 0; index < foreignCount; index += 1) { + const id = `foreign-${index}` + repository.rows.set(id, makeRow(id, { scope_type: 'session', scope_id: 'foreign-session' })) + store.vectors.set(id, [1, 0, 0, 0]) + } + for (const [id, vector] of [ + ['target-redis', [0.9, 0.1, 0, 0]], + ['target-vue', [0.1, 0.9, 0, 0]] + ] as const) { + repository.rows.set(id, makeRow(id, { scope_type: 'session', scope_id: 'target-session' })) + store.vectors.set(id, [...vector]) + } + vi.spyOn(repository, 'searchWithStrategy').mockReturnValue({ rows: [], strategy: 'fts-only' }) + const getEmbeddings = vi.fn( + async (_agent: string, _provider: string, _model: string, texts: string[]) => + texts.map(textToVector) + ) + const queryBatch = vi.fn( + async ( + _agent: string, + _embedding: MemoryModelRef, + _dimensions: number, + vectors: number[][], + topK: number + ) => Promise.all(vectors.map((vector) => store.query(vector, { topK }))) + ) + const recordRecall = vi.fn() + const service = new RetrievalService({ + ctx, + repository, + policy, + embeddingGateway: { + getEmbeddings, + getDimensions: async () => ({ data: { dimensions: 4, normalized: false } }) + }, + vectorStore: { + getRecallHealth: () => 'available', + hasReadyCertificate: () => true, + query: async () => [], + queryBatch, + markReady: () => undefined, + clearReady: vi.fn() + }, + workingMemory: { + readWorkingMemory: () => null, + flushWorkingMemoryIfDirty: () => undefined, + scheduleWorkingRefresh: () => undefined + }, + warmVectorStore: async () => undefined, + warmEmbeddingConnection: () => undefined, + reindexEmbeddings: async () => undefined, + backfillEmbeddings: async () => undefined, + isReindexing: () => false, + deletePrunableVectorsForMemoryIds: async () => [], + getActiveSuppressionTopics: () => [], + diagnostics: { recordRecall } + }) + return { service, repository, store, config, getEmbeddings, queryBatch, recordRecall } + } + + it('finds the applicable thirteenth vector while batching distinct queries once', async () => { + const { service, store, getEmbeddings, queryBatch } = setup(12) + const nearest = await store.query(textToVector('redis'), { topK: 13 }) + expect(nearest.slice(0, 12).every((match) => match.memoryId.startsWith('foreign-'))).toBe(true) + expect(nearest[11].distance).toBeLessThan(nearest[12].distance) + expect(nearest[12].memoryId).toBe('target-redis') + + const results = await service.retrieveForDecisions( + 'a', + candidates, + 3000, + undefined, + undefined, + scope + ) + + expect(results.map((result) => result.neighbors.map((neighbor) => neighbor.id))).toEqual([ + ['target-redis'], + ['target-vue'] + ]) + expect(getEmbeddings).toHaveBeenCalledTimes(1) + expect(getEmbeddings.mock.calls[0][3]).toEqual(candidates.map((candidate) => candidate.content)) + expect(queryBatch).toHaveBeenCalledTimes(1) + expect(queryBatch.mock.calls[0][3]).toEqual( + candidates.map((candidate) => textToVector(candidate.content)) + ) + expect(queryBatch.mock.calls[0][4]).toBe(800) + }) + + it('keeps the original page when it already contains enough applicable neighbors', async () => { + const { service, repository, recordRecall } = setup(12) + for (let index = 0; index < 12; index += 1) { + const row = repository.rows.get(`foreign-${index}`)! + row.scope_id = 'target-session' + row.importance = 0 + } + repository.rows.get('target-redis')!.importance = 1 + + const [result] = await service.retrieveForDecisions( + 'a', + [candidates[0]], + 3000, + undefined, + undefined, + scope + ) + + expect(result.neighbors.map((neighbor) => neighbor.id)).toEqual([ + 'foreign-0', + 'foreign-1', + 'foreign-2' + ]) + expect(recordRecall).toHaveBeenCalledWith( + 'a', + expect.objectContaining({ + vectorCandidates: 12, + degradations: [] + }) + ) + }) + + it('bounds an exhausted pool at 800 and keeps retry snapshots and pinned order', async () => { + const { service, repository, getEmbeddings, queryBatch, recordRecall } = setup(800) + for (const id of ['head-a', 'head-b']) { + repository.rows.set(id, makeRow(id, { scope_type: 'session', scope_id: 'target-session' })) + } + const snapshot = { vector: textToVector('redis'), providerId: 'p', modelId: 'm', dimensions: 4 } + const results = await service.retrieveForDecisions( + 'a', + candidates, + 3000, + [snapshot, undefined], + [['head-b', 'head-a']], + scope + ) + + expect(results[0].neighbors.map((neighbor) => neighbor.id)).toEqual(['head-b', 'head-a']) + expect(results[0].queryVector).toEqual(snapshot) + expect(results[1]).toEqual({ neighbors: [], queryVector: undefined }) + expect(getEmbeddings).not.toHaveBeenCalled() + expect(queryBatch).toHaveBeenCalledTimes(1) + expect(queryBatch.mock.calls[0][3]).toEqual([snapshot.vector]) + expect(queryBatch.mock.calls[0][4]).toBe(800) + expect(recordRecall).toHaveBeenCalledWith( + 'a', + expect.objectContaining({ + degradations: ['candidateBudgetExhausted'] + }) + ) + }) + + it('returns the applicable vector at the budget boundary', async () => { + const { service, recordRecall } = setup(799) + const [result] = await service.retrieveForDecisions( + 'a', + [candidates[0]], + 3000, + undefined, + undefined, + scope + ) + expect(result.neighbors.map((neighbor) => neighbor.id)).toEqual(['target-redis']) + expect(recordRecall).toHaveBeenCalledWith( + 'a', + expect.objectContaining({ + degradations: ['candidateBudgetExhausted'] + }) + ) + }) + + it('discards the widened pool when memory is disabled during the scan', async () => { + const { service, repository, config, queryBatch } = setup(12) + const scan = deferred>>() + queryBatch.mockImplementationOnce(() => scan.promise) + const revalidate = vi.spyOn(repository, 'listApplicableByIds') + const retrieval = service.retrieveForDecisions( + 'a', + [candidates[0]], + 3000, + undefined, + undefined, + scope + ) + await vi.waitFor(() => expect(queryBatch).toHaveBeenCalledTimes(1)) + config.memoryEnabled = false + scan.resolve([[{ memoryId: 'target-redis', distance: 0.1 }]]) + + await expect(retrieval).resolves.toEqual([{ neighbors: [] }]) + expect(revalidate).not.toHaveBeenCalled() + }) +}) + describe('MemoryService forgetting score (T-B1..T-B2)', () => { it('decay only reranks: an old active memory still appears, just lower (T-B1)', () => { const now = 1_000 * DAY