From e4567fbb89b8be89183e5c293861dd34771f690a Mon Sep 17 00:00:00 2001 From: yyhhyyyyyy Date: Tue, 7 Jul 2026 15:38:33 +0800 Subject: [PATCH] fix(memory): improve agent memory reliability --- .../presenter/agentRuntimePresenter/index.ts | 82 +++ src/main/presenter/memoryPresenter/context.ts | 6 +- .../memoryPresenter/core/injectionPort.ts | 4 + src/main/presenter/memoryPresenter/index.ts | 29 +- .../infra/embeddingPipeline.ts | 4 +- .../memoryPresenter/runtimeConstants.ts | 1 + .../services/conflictService.ts | 18 +- .../services/maintenanceService.ts | 469 +++++++++++------- .../services/personaService.ts | 62 ++- .../services/reflectionService.ts | 48 +- .../services/retrievalService.ts | 63 ++- .../services/writeCoordinator.ts | 29 +- src/main/presenter/memoryPresenter/types.ts | 19 +- .../sqlitePresenter/schemaCatalog.ts | 8 +- .../sqlitePresenter/tables/agentMemory.ts | 5 +- .../tables/agentMemoryAudit.ts | 112 ++++- test/main/presenter/agentMemoryTable.test.ts | 154 +++++- .../agentRuntimePresenter.test.ts | 106 ++++ test/main/presenter/fakes/memoryFakes.ts | 11 +- test/main/presenter/memoryAdd.test.ts | 32 ++ test/main/presenter/memoryExtraction.test.ts | 47 +- test/main/presenter/memoryPresenter.test.ts | 126 ++++- .../memorySessionExtractionLock.test.ts | 14 + 23 files changed, 1177 insertions(+), 272 deletions(-) diff --git a/src/main/presenter/agentRuntimePresenter/index.ts b/src/main/presenter/agentRuntimePresenter/index.ts index 10e11fe5c..43a33af9b 100644 --- a/src/main/presenter/agentRuntimePresenter/index.ts +++ b/src/main/presenter/agentRuntimePresenter/index.ts @@ -170,6 +170,11 @@ type PendingInteractionEntry = { blockIndex: number } +type MemoryInjectionAccessTurnEntry = { + ids: Set + touchedAt: number +} + type ProcessPendingInputSource = PendingInputEnqueueSource | 'steer' type PendingTapeViewContext = { @@ -218,6 +223,8 @@ const PROVIDER_OVERFLOW_RETRY_EXTRA_RESERVE_CAP = 8_192 const AUTO_APPROVE_REVIEW_MAX_RECENT_MESSAGES = 8 const AUTO_APPROVE_REVIEW_MAX_CONTENT_CHARS = 2_000 const AUTO_APPROVE_REVIEW_TIMEOUT_MS = 30_000 +const MEMORY_INJECTION_ACCESS_TURN_TTL_MS = 30 * 60 * 1000 +const MEMORY_INJECTION_ACCESS_MAX_TURNS_PER_SESSION = 128 function normalizePermissionMode(mode: PermissionMode | null | undefined): PermissionMode { return mode === 'default' || mode === 'auto_approve' ? mode : 'full_access' @@ -644,6 +651,7 @@ export class AgentRuntimePresenter implements IAgentImplementation { private readonly memoryPort?: MemoryRuntimePort private readonly memoryExtractionChains = new Map>() private readonly memoryExtractionEpochs = new Map() + private readonly memoryInjectionAccessByTurn = new Map() private readonly cacheImage?: (data: string) => Promise private readonly skillPresenter?: Pick< ISkillPresenter, @@ -928,6 +936,7 @@ export class AgentRuntimePresenter implements IAgentImplementation { this.runtimeActivatedSkillsBySession.delete(sessionId) this.sessionCompactionStates.delete(sessionId) this.drainingPendingQueues.delete(sessionId) + this.clearMemoryInjectionAccessForSession(sessionId) this.toolPresenter?.clearConversationToolMapping?.(sessionId) } @@ -2435,6 +2444,7 @@ export class AgentRuntimePresenter implements IAgentImplementation { const injection = await this.memoryPort.buildInjection(agentId, query) const assembled = appendMemorySectionWithManifest(systemPrompt, injection) if (assembled.manifest) { + this.recordMemoryInjectionAccess(agentId, sessionId, assembled.manifest.selected, messageId) try { this.sqlitePresenter.deepchatTapeEntriesTable.appendAnchor({ sessionId, @@ -2453,6 +2463,78 @@ export class AgentRuntimePresenter implements IAgentImplementation { } } + private recordMemoryInjectionAccess( + agentId: string, + sessionId: string, + selected: Array<{ id: string }>, + messageId?: string | null + ): void { + if (!this.memoryPort || selected.length === 0) return + const selectedIds = [...new Set(selected.map((item) => item.id).filter(Boolean))] + if (!selectedIds.length) return + + let idsToRecord = selectedIds + let seen: Set | undefined + if (messageId) { + const now = Date.now() + this.pruneMemoryInjectionAccessForSession(sessionId, now) + const key = this.memoryInjectionAccessKey(sessionId, messageId) + let entry = this.memoryInjectionAccessByTurn.get(key) + if (!entry) { + entry = { ids: new Set(), touchedAt: now } + this.memoryInjectionAccessByTurn.set(key, entry) + this.pruneMemoryInjectionAccessForSession(sessionId, now) + } else { + entry.touchedAt = now + } + seen = entry.ids + const trackedIds = seen + idsToRecord = selectedIds.filter((id) => !trackedIds.has(id)) + if (!idsToRecord.length) return + } + + try { + this.memoryPort.recordInjectionAccess(agentId, idsToRecord) + if (seen) { + for (const id of idsToRecord) seen.add(id) + } + } catch (error) { + logger.warn(`[DeepChatAgent] memory access accounting skipped: ${String(error)}`) + } + } + + private memoryInjectionAccessKey(sessionId: string, messageId: string): string { + return `${sessionId}\u0000${messageId}` + } + + private clearMemoryInjectionAccessForSession(sessionId: string): void { + const prefix = `${sessionId}\u0000` + for (const key of this.memoryInjectionAccessByTurn.keys()) { + if (key.startsWith(prefix)) this.memoryInjectionAccessByTurn.delete(key) + } + } + + private pruneMemoryInjectionAccessForSession(sessionId: string, now: number = Date.now()): void { + const prefix = `${sessionId}\u0000` + const entries: Array<{ key: string; touchedAt: number }> = [] + for (const [key, entry] of this.memoryInjectionAccessByTurn) { + if (!key.startsWith(prefix)) continue + if (now - entry.touchedAt > MEMORY_INJECTION_ACCESS_TURN_TTL_MS) { + this.memoryInjectionAccessByTurn.delete(key) + continue + } + entries.push({ key, touchedAt: entry.touchedAt }) + } + if (entries.length <= MEMORY_INJECTION_ACCESS_MAX_TURNS_PER_SESSION) return + entries.sort( + (left, right) => left.touchedAt - right.touchedAt || left.key.localeCompare(right.key) + ) + const deleteCount = entries.length - MEMORY_INJECTION_ACCESS_MAX_TURNS_PER_SESSION + for (const entry of entries.slice(0, deleteCount)) { + this.memoryInjectionAccessByTurn.delete(entry.key) + } + } + private triggerMemoryExtractionFromCompaction(sessionId: string, intent: CompactionIntent): void { if (!this.memoryPort) return const agentId = this.getSessionAgentId(sessionId) ?? 'deepchat' diff --git a/src/main/presenter/memoryPresenter/context.ts b/src/main/presenter/memoryPresenter/context.ts index d46fa5d7e..b4b5b8b5e 100644 --- a/src/main/presenter/memoryPresenter/context.ts +++ b/src/main/presenter/memoryPresenter/context.ts @@ -19,7 +19,10 @@ export function isUniqueConstraintError(error: unknown): boolean { export class MemoryRuntimeContext { private disposed = false - constructor(readonly deps: MemoryPresenterDeps) {} + constructor( + readonly deps: MemoryPresenterDeps, + private readonly onAgentMemoryMutated?: (agentId: string) => void + ) {} get isDisposed(): boolean { return this.disposed @@ -70,6 +73,7 @@ export class MemoryRuntimeContext { } emitChanged(agentId: string, reason: MemoryUpdateReason, context?: MemoryUpdateContext): void { + this.onAgentMemoryMutated?.(agentId) if (context) this.deps.onMemoryChanged?.(agentId, reason, context) else this.deps.onMemoryChanged?.(agentId, reason) } diff --git a/src/main/presenter/memoryPresenter/core/injectionPort.ts b/src/main/presenter/memoryPresenter/core/injectionPort.ts index 8f2b77351..8fa268ae2 100644 --- a/src/main/presenter/memoryPresenter/core/injectionPort.ts +++ b/src/main/presenter/memoryPresenter/core/injectionPort.ts @@ -96,6 +96,10 @@ export interface MemoryInjectionPort { // Adds extraction entry points on top of injection. Extraction is an independent cheap // LLM call that never touches summarization. export interface MemoryRuntimePort extends MemoryInjectionPort { + // Records memory rows that actually entered the assembled runtime prompt. Runtime owns the + // final manifest visibility; memory presenter still owns the storage mutation. + recordInjectionAccess(agentId: string, memoryIds: string[], accessedAt?: number): void + // Extracts memories from a span and writes them (status=pending_embedding). // Resolves { ok:true, createdIds } (createdIds may be empty) or { ok:false } on failure. // Never throws or blocks the caller; on ok:false the caller must keep its cursor for retry. diff --git a/src/main/presenter/memoryPresenter/index.ts b/src/main/presenter/memoryPresenter/index.ts index 06b8c89f7..5f90c1ca6 100644 --- a/src/main/presenter/memoryPresenter/index.ts +++ b/src/main/presenter/memoryPresenter/index.ts @@ -67,7 +67,10 @@ export class MemoryPresenter implements MemoryRuntimePort { private readonly management: ManagementService constructor(deps: MemoryPresenterDeps) { - this.runtime = new MemoryRuntimeContext(deps) + let retrievalService: RetrievalService | null = null + this.runtime = new MemoryRuntimeContext(deps, (agentId) => { + retrievalService?.invalidateKeywordStats(agentId) + }) this.rows = new MemoryRowMutations(this.runtime) this.vectorStore = new VectorStoreManager(this.runtime) this.embedding = new EmbeddingPipeline(this.runtime, this.vectorStore, this.rows, { @@ -92,6 +95,7 @@ export class MemoryPresenter implements MemoryRuntimePort { memoryIds ) }) + retrievalService = this.retrieval this.reflection = new ReflectionService(this.runtime, { syncWorkingMemoryAfterMutation: (agentId) => @@ -128,8 +132,10 @@ export class MemoryPresenter implements MemoryRuntimePort { warmVectorStore: (agentId, embedding) => this.embedding.warmVectorStore(agentId, embedding), warmEmbeddingConnection: (agentId, embedding) => this.embedding.warmEmbeddingConnection(agentId, embedding), - maybeReflect: (agentId, model) => this.reflection.maybeReflect(agentId, model), - maybeEvolvePersona: (agentId, model) => this.persona.maybeEvolvePersona(agentId, model), + maybeReflect: (agentId, model) => + this.reflection.runMaintenanceReflectionPass(agentId, model), + maybeEvolvePersona: (agentId, model) => + this.persona.runMaintenancePersonaPass(agentId, model), runChallengeResolutionPass: (agentId, model) => this.conflict.runChallengeResolutionPass(agentId, model), runConsolidationPass: (agentId) => this.runConsolidationPass(agentId) @@ -282,7 +288,9 @@ export class MemoryPresenter implements MemoryRuntimePort { } writeMemoriesSync(candidates: MemoryCandidate[], options: WriteMemoriesOptions): string[] { - return this.writeCoordinator.writeMemoriesSync(candidates, options) + const ids = this.writeCoordinator.writeMemoriesSync(candidates, options) + if (ids.length > 0) this.retrieval.invalidateKeywordStats(options.agentId) + return ids } processPendingEmbeddings(agentId: string, limit = 50): Promise { @@ -384,6 +392,19 @@ export class MemoryPresenter implements MemoryRuntimePort { return this.retrieval.buildInjection(agentId, query) } + recordInjectionAccess( + agentId: string, + memoryIds: string[], + accessedAt: number = Date.now() + ): void { + if (!this.runtime.canReadAgentMemory(agentId)) return + const uniqueIds = [...new Set(memoryIds.map((id) => id.trim()).filter(Boolean))] + if (!uniqueIds.length) return + const ownedIds = this.runtime.deps.repository.listByIds(agentId, uniqueIds).map((row) => row.id) + if (!ownedIds.length) return + this.runtime.deps.repository.recordAccessBatch(ownedIds, accessedAt) + } + refreshWorkingMemory(agentId: string): void { this.workingMemory.refreshWorkingMemory(agentId) } diff --git a/src/main/presenter/memoryPresenter/infra/embeddingPipeline.ts b/src/main/presenter/memoryPresenter/infra/embeddingPipeline.ts index 4931b6559..09ad77fc1 100644 --- a/src/main/presenter/memoryPresenter/infra/embeddingPipeline.ts +++ b/src/main/presenter/memoryPresenter/infra/embeddingPipeline.ts @@ -302,7 +302,9 @@ export class EmbeddingPipeline { private async runBackfill(agentId: string): Promise { await Promise.resolve() if (!this.ctx.canContinueAgentMemoryTask(agentId)) return - this.ctx.deps.repository.requeueForEmbedding(agentId, ['fts_only']) + if (this.ctx.deps.repository.listEmbeddingStatusIds(agentId, ['fts_only'], 1).length) { + this.ctx.deps.repository.requeueForEmbedding(agentId, ['fts_only']) + } await this.drainUntilExhausted(agentId) } diff --git a/src/main/presenter/memoryPresenter/runtimeConstants.ts b/src/main/presenter/memoryPresenter/runtimeConstants.ts index f1881bb9a..c81996ce0 100644 --- a/src/main/presenter/memoryPresenter/runtimeConstants.ts +++ b/src/main/presenter/memoryPresenter/runtimeConstants.ts @@ -25,6 +25,7 @@ export const DECISION_NEIGHBOR_TOP_S = 10 export const CONSOLIDATION_IDLE_MS = 5 * 60 * 1000 export const CONSOLIDATION_COOLDOWN_MS = 6 * 60 * 60 * 1000 +export const CONSOLIDATION_FAILURE_COOLDOWN_MS = 30 * 60 * 1000 export const CONSOLIDATION_MAX_LLM_CALLS = 8 export const CONSOLIDATION_MAX_INPUT_TOKENS = 24000 export const CONSOLIDATION_MERGE_SIMILARITY = 0.85 diff --git a/src/main/presenter/memoryPresenter/services/conflictService.ts b/src/main/presenter/memoryPresenter/services/conflictService.ts index f5c9b2d08..123513576 100644 --- a/src/main/presenter/memoryPresenter/services/conflictService.ts +++ b/src/main/presenter/memoryPresenter/services/conflictService.ts @@ -8,7 +8,12 @@ import { } from '../core/decision' import { normalizeMemoryCandidate } from '../core/candidates' import { buildMemoryProvenanceKey } from '../core/scoring' -import type { AgentMemoryRow, MemoryConflictPair, MemoryConflictResolution } from '../types' +import type { + AgentMemoryRow, + MemoryConflictPair, + MemoryConflictResolution, + MemoryMaintenanceStepResult +} from '../types' import { type MemoryModelRef, type MemoryRuntimeContext } from '../context' interface ConflictResolutionOptions { @@ -152,8 +157,13 @@ export class ConflictService { .filter((row) => row.id !== excludeChallengerId && row.conflict_with === targetId) } - async runChallengeResolutionPass(agentId: string, model: MemoryModelRef): Promise { + async runChallengeResolutionPass( + agentId: string, + model: MemoryModelRef + ): Promise { let touched = false + let calls = 0 + let failures = 0 for (const pair of this.listConflicts(agentId)) { const promptCandidate = normalizeMemoryCandidate({ kind: pair.challenger.kind === 'episodic' ? 'episodic' : 'semantic', @@ -165,9 +175,11 @@ export class ConflictService { const prompt = buildDecisionPrompt(promptCandidate, [{ content: pair.target.content }]) let decision: MemoryDecision = ADD_DECISION try { + calls += 1 const raw = await this.ctx.deps.generateText(model.providerId, model.modelId, prompt) decision = parseDecision(raw, 1) } catch (error) { + failures += 1 logger.warn(`[Memory] challenge decision failed: ${String(error)}`) continue } @@ -190,6 +202,6 @@ export class ConflictService { touched = true } } - return touched + return { touched, calls, failures } } } diff --git a/src/main/presenter/memoryPresenter/services/maintenanceService.ts b/src/main/presenter/memoryPresenter/services/maintenanceService.ts index 9e93f072c..714d16cab 100644 --- a/src/main/presenter/memoryPresenter/services/maintenanceService.ts +++ b/src/main/presenter/memoryPresenter/services/maintenanceService.ts @@ -13,6 +13,7 @@ import { normalizeMemoryCandidate } from '../core/candidates' import { estimateTokens } from '../core/injectionPort' import { CONSOLIDATION_COOLDOWN_MS, + CONSOLIDATION_FAILURE_COOLDOWN_MS, CONSOLIDATION_IDLE_MS, CONSOLIDATION_MAX_INPUT_TOKENS, CONSOLIDATION_MAX_LLM_CALLS, @@ -30,8 +31,9 @@ import { type AgentMemoryRow, type ConsolidationScanCursor, FORGET_HALF_LIFE_MS, - type MemoryPersonaDraftResult, - type MemoryReflectionResult + type MemoryMaintenancePersonaResult, + type MemoryMaintenanceReflectionResult, + type MemoryMaintenanceStepResult } from '../types' import type { MemoryRowMutations } from './rowMutations' import { embeddingFingerprint, type MemoryModelRef, type MemoryRuntimeContext } from '../context' @@ -40,6 +42,7 @@ export class MaintenanceService { private readonly consolidationTimers = new Map() private readonly consolidationTimerDueAt = new Map() private readonly lastConsolidationAt = new Map() + private readonly lastConsolidationFailureAt = new Map() private readonly consolidationScanCursor = new Map() private readonly consolidationRuns = new Set>() private maintenanceStartTimer: NodeJS.Timeout | null = null @@ -72,12 +75,15 @@ export class MaintenanceService { maybeReflect: ( agentId: string, model: MemoryModelRef - ) => Promise + ) => Promise maybeEvolvePersona: ( agentId: string, model: MemoryModelRef - ) => Promise - runChallengeResolutionPass: (agentId: string, model: MemoryModelRef) => Promise + ) => Promise + runChallengeResolutionPass: ( + agentId: string, + model: MemoryModelRef + ) => Promise runConsolidationPass: (agentId: string) => Promise } ) {} @@ -118,6 +124,7 @@ export class MaintenanceService { this.consolidationTimers.clear() this.consolidationTimerDueAt.clear() this.lastConsolidationAt.clear() + this.lastConsolidationFailureAt.clear() this.consolidationScanCursor.clear() } @@ -250,6 +257,11 @@ export class MaintenanceService { await this.runCheapMaintenance(agentId, now, true) return } + const latestFailureAt = this.getLatestMaintenanceFailureAt(agentId) + if (latestFailureAt !== null && now - latestFailureAt < CONSOLIDATION_FAILURE_COOLDOWN_MS) { + await this.runCheapMaintenance(agentId, now, true) + return + } await this.runCheapMaintenance(agentId, now, false) const model = this.ctx.resolveConsolidationModel(agentId) if (!model) { @@ -264,76 +276,109 @@ export class MaintenanceService { }) return } + const previousLast = last this.lastConsolidationAt.set(agentId, now) let touched = false + const llmStats: MemoryMaintenanceStepResult = { touched: false, calls: 0, failures: 0 } + let completedHeavyPass = false try { - touched = await this.mergeNearDuplicates(agentId, now, model) - } catch (error) { - logger.warn(`[Memory] consolidation merge failed for ${agentId}: ${String(error)}`) - } - if (!this.ctx.canWriteAgentMemory(agentId)) return - try { - if (await this.ports.runChallengeResolutionPass(agentId, model)) touched = true - } catch (error) { - logger.warn(`[Memory] challenge resolution failed for ${agentId}: ${String(error)}`) - } - if (!this.ctx.canWriteAgentMemory(agentId)) return - try { - const reflection = await this.ports.maybeReflect(agentId, model) - if (reflection) { - this.ctx.writeAudit(agentId, { - eventType: 'memory/reflect', - actorType: 'scheduler', - status: 'completed', - inputRefs: { memoryIds: reflection.sourceMemoryIds }, - outputRefs: { memoryIds: reflection.reflectionIds }, - model - }) - touched = true + try { + const merge = await this.mergeNearDuplicates(agentId, now, model) + this.addLlmStats(llmStats, merge) + if (merge.touched) touched = true + } catch (error) { + logger.warn(`[Memory] consolidation merge failed for ${agentId}: ${String(error)}`) } - } catch (error) { - logger.warn(`[Memory] background reflection failed for ${agentId}: ${String(error)}`) - } - if (!this.ctx.canWriteAgentMemory(agentId)) return - try { - const personaDraft = await this.ports.maybeEvolvePersona(agentId, model) - if (personaDraft) { + if (!this.ctx.canWriteAgentMemory(agentId)) return + try { + const challenge = await this.ports.runChallengeResolutionPass(agentId, model) + this.addLlmStats(llmStats, challenge) + if (challenge.touched) touched = true + } catch (error) { + logger.warn(`[Memory] challenge resolution failed for ${agentId}: ${String(error)}`) + } + if (!this.ctx.canWriteAgentMemory(agentId)) return + try { + const reflectionPass = await this.ports.maybeReflect(agentId, model) + this.addLlmStats(llmStats, reflectionPass) + const reflection = reflectionPass.result + if (reflection) { + this.ctx.writeAudit(agentId, { + eventType: 'memory/reflect', + actorType: 'scheduler', + status: 'completed', + inputRefs: { memoryIds: reflection.sourceMemoryIds }, + outputRefs: { memoryIds: reflection.reflectionIds }, + model + }) + touched = true + } + } catch (error) { + logger.warn(`[Memory] background reflection failed for ${agentId}: ${String(error)}`) + } + if (!this.ctx.canWriteAgentMemory(agentId)) return + try { + const personaPass = await this.ports.maybeEvolvePersona(agentId, model) + this.addLlmStats(llmStats, personaPass) + const personaDraft = personaPass.result + if (personaDraft) { + this.ctx.writeAudit(agentId, { + eventType: 'persona/evolve', + actorType: 'scheduler', + status: 'completed', + outputRefs: { + draftId: personaDraft.draftId, + needsReview: personaDraft.needsReview, + changeRatio: personaDraft.changeRatio + }, + model + }) + } + } catch (error) { + logger.warn(`[Memory] background persona evolution failed for ${agentId}: ${String(error)}`) + } + if (!this.ctx.canWriteAgentMemory(agentId)) return + if (this.didAllAttemptedLlmCallsFail(llmStats)) { + this.lastConsolidationAt.set(agentId, previousLast) + this.lastConsolidationFailureAt.set(agentId, now) this.ctx.writeAudit(agentId, { - eventType: 'persona/evolve', + eventType: 'memory/maintenance_llm', actorType: 'scheduler', - status: 'completed', - outputRefs: { - draftId: personaDraft.draftId, - needsReview: personaDraft.needsReview, - changeRatio: personaDraft.changeRatio - }, - model + status: 'failed', + reason: 'all-llm-steps-failed', + outputRefs: { calls: llmStats.calls, failures: llmStats.failures }, + model, + createdAt: now }) + return } - } catch (error) { - logger.warn(`[Memory] background persona evolution failed for ${agentId}: ${String(error)}`) - } - 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, { - eventType: 'memory/maintenance_llm', - actorType: 'scheduler', - status: 'completed', - outputRefs: { touched }, - model, - createdAt: now - }) - - if (touched) { - void this.ports.triggerEmbedding(agentId).catch((error) => { - logger.warn(`[Memory] background embedding failed: ${String(error)}`) + this.refreshDecayScores(agentId, now) + this.archiveStale(agentId, now) + await this.pruneDeadVectors(agentId) + this.ports.syncWorkingMemoryAfterMutation(agentId) + this.stampConsolidation(agentId, now) + this.ctx.writeAudit(agentId, { + eventType: 'memory/maintenance_llm', + actorType: 'scheduler', + status: 'completed', + outputRefs: { touched }, + model, + createdAt: now }) - this.ctx.emitChanged(agentId, 'extract') + completedHeavyPass = true + this.lastConsolidationFailureAt.delete(agentId) + + if (touched) { + void this.ports.triggerEmbedding(agentId).catch((error) => { + logger.warn(`[Memory] background embedding failed: ${String(error)}`) + }) + this.ctx.emitChanged(agentId, 'extract') + } + } finally { + if (!completedHeavyPass && this.lastConsolidationAt.get(agentId) === now) { + this.lastConsolidationAt.set(agentId, previousLast) + } } } @@ -341,147 +386,187 @@ export class MaintenanceService { agentId: string, now: number, model: MemoryModelRef - ): Promise { - const embedding = this.ctx.deps.resolveAgentConfig(agentId)?.memoryEmbedding - 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, { + ): Promise { + const result: MemoryMaintenanceStepResult = { touched: false, calls: 0, failures: 0 } + try { + const embedding = this.ctx.deps.resolveAgentConfig(agentId)?.memoryEmbedding + if (!embedding?.providerId || !embedding?.modelId) return result + 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 result + await this.ports.warmVectorStore(agentId, currentEmbedding) + if (!this.ctx.canWriteAgentMemory(agentId)) return result + + 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 }) - } - 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 - } + if (cursor && scanRows.length === 0) { + scanRows = this.ctx.deps.repository.listConsolidationScanRows(agentId, { + embeddingDim: dimensions, + embeddingModel: fingerprint, + limit: CONSOLIDATION_MAX_NEIGHBOR_SCANS + 1 + }) + } + const hasMore = scanRows.length > CONSOLIDATION_MAX_NEIGHBOR_SCANS + scanRows = scanRows.slice(0, CONSOLIDATION_MAX_NEIGHBOR_SCANS) + if (!scanRows.length) { + this.consolidationScanCursor.delete(agentId) + return result + } - let calls = 0 - let inputTokens = 0 - const merged = new Set() - let lastScanned: AgentMemoryRow | null = null - let touched = false + let calls = 0 + let inputTokens = 0 + const merged = new Set() + let lastScanned: AgentMemoryRow | null = null + let touched = false - 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 + 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 matches: Array<{ memoryId: string; distance: number }> = [] - try { - matches = await this.ports.queryNeighborsByMemoryId( - agentId, - currentEmbedding, - dimensions, - source.id, - DECISION_NEIGHBOR_TOP_S - ) - } catch { - continue - } - if (!this.ctx.canWriteAgentMemory(agentId)) break - 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)) + let matches: Array<{ memoryId: string; distance: number }> = [] + try { + matches = await this.ports.queryNeighborsByMemoryId( + agentId, + currentEmbedding, + dimensions, + source.id, + DECISION_NEIGHBOR_TOP_S + ) + } catch { continue - neighbor = neighborRow - break - } - if (!neighbor) continue + } + if (!this.ctx.canWriteAgentMemory(agentId)) break + 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: source.kind === 'episodic' ? 'episodic' : 'semantic', - category: source.category, - content: source.content, - importance: source.importance - }) - if (!promptCandidate) continue - const prompt = buildDecisionPrompt(promptCandidate, [{ content: neighbor.content }]) - calls += 1 - inputTokens += estimateTokens(prompt) - let decision: MemoryDecision = ADD_DECISION - try { - const raw = await this.ctx.deps.generateText(model.providerId, model.modelId, prompt) - decision = parseDecision(raw, 1) - } catch (error) { - logger.warn(`[Memory] consolidation decision failed: ${String(error)}`) - continue - } - if (!this.ctx.canWriteAgentMemory(agentId)) break - - if (decision.decision === 'UPDATE' || decision.decision === 'SUPERSEDE') { - const [primary, secondary] = - source.created_at >= neighbor.created_at ? [source, neighbor] : [neighbor, source] - const mergedContent = decision.mergedContent ?? primary.content - const secondaryCategory = isAgentMemoryCategory(secondary.category) - ? secondary.category - : null - const update = this.rows.applyContentUpdate( - agentId, - primary, - mergedContent, - now, - secondaryCategory - ) - if (update.action === 'suppressed') { - this.ctx.deps.repository.setLastConsolidatedAt(source.id, now) - this.ctx.deps.repository.setLastConsolidatedAt(neighbor.id, now) - merged.add(source.id) - merged.add(neighbor.id) + const promptCandidate = normalizeMemoryCandidate({ + 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 }]) + calls += 1 + result.calls += 1 + inputTokens += estimateTokens(prompt) + let decision: MemoryDecision = ADD_DECISION + try { + const raw = await this.ctx.deps.generateText(model.providerId, model.modelId, prompt) + decision = parseDecision(raw, 1) + } catch (error) { + result.failures += 1 + logger.warn(`[Memory] consolidation decision failed: ${String(error)}`) continue } - const survivorId = update.id - this.rows.bumpConfidence(survivorId) - this.ctx.deps.repository.setImportance(survivorId, secondary.importance) - this.ctx.deps.repository.updateStatus(survivorId, 'pending_embedding') - if (secondary.id !== survivorId) { - this.ctx.deps.repository.markSuperseded(secondary.id, survivorId) + if (!this.ctx.canWriteAgentMemory(agentId)) break + + if (decision.decision === 'UPDATE' || decision.decision === 'SUPERSEDE') { + const [primary, secondary] = + source.created_at >= neighbor.created_at ? [source, neighbor] : [neighbor, source] + const mergedContent = decision.mergedContent ?? primary.content + const secondaryCategory = isAgentMemoryCategory(secondary.category) + ? secondary.category + : null + const update = this.rows.applyContentUpdate( + agentId, + primary, + mergedContent, + now, + secondaryCategory + ) + if (update.action === 'suppressed') { + this.ctx.deps.repository.setLastConsolidatedAt(source.id, now) + this.ctx.deps.repository.setLastConsolidatedAt(neighbor.id, now) + merged.add(source.id) + merged.add(neighbor.id) + continue + } + const survivorId = update.id + this.rows.bumpConfidence(survivorId) + this.ctx.deps.repository.setImportance(survivorId, secondary.importance) + this.ctx.deps.repository.updateStatus(survivorId, 'pending_embedding') + if (secondary.id !== survivorId) { + this.ctx.deps.repository.markSuperseded(secondary.id, survivorId) + } + merged.add(primary.id) + merged.add(secondary.id) + merged.add(survivorId) + touched = true + result.touched = true } - merged.add(primary.id) - merged.add(secondary.id) - merged.add(survivorId) - touched = true + this.ctx.deps.repository.setLastConsolidatedAt(source.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) + } + result.touched = result.touched || touched + return result + } catch (error) { + logger.warn(`[Memory] consolidation merge scan aborted for ${agentId}: ${String(error)}`) + return result } - 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) + } + + private addLlmStats( + total: MemoryMaintenanceStepResult, + next: { touched?: boolean; calls: number; failures: number } + ): void { + total.calls += next.calls + total.failures += next.failures + total.touched = total.touched || next.touched === true + } + + private didAllAttemptedLlmCallsFail(stats: { calls: number; failures: number }): boolean { + return stats.calls > 0 && stats.failures >= stats.calls + } + + private getLatestMaintenanceFailureAt(agentId: string): number | null { + const localFailureAt = this.lastConsolidationFailureAt.get(agentId) + if (localFailureAt !== undefined) return localFailureAt + const persisted = this.ctx.deps.auditRepository?.listByAgent(agentId, { + eventType: 'memory/maintenance_llm', + status: 'failed', + limit: 1 + })[0]?.created_at + if (persisted !== undefined) { + this.lastConsolidationFailureAt.set(agentId, persisted) + return persisted } - return touched + return null } private isLiveConsolidationNeighbor( @@ -591,6 +676,7 @@ export class MaintenanceService { clearCooldown(agentId: string): void { this.lastConsolidationAt.delete(agentId) + this.lastConsolidationFailureAt.delete(agentId) } cleanupAgent(agentId: string): void { @@ -600,6 +686,7 @@ export class MaintenanceService { this.consolidationTimers.delete(agentId) this.consolidationTimerDueAt.delete(agentId) this.lastConsolidationAt.delete(agentId) + this.lastConsolidationFailureAt.delete(agentId) this.consolidationScanCursor.delete(agentId) } diff --git a/src/main/presenter/memoryPresenter/services/personaService.ts b/src/main/presenter/memoryPresenter/services/personaService.ts index 9a98ea756..bff69f5e4 100644 --- a/src/main/presenter/memoryPresenter/services/personaService.ts +++ b/src/main/presenter/memoryPresenter/services/personaService.ts @@ -12,7 +12,11 @@ import { sanitizeSelfModel, PERSONA_MAX_CHANGE_RATIO } from '../core/extraction' -import type { AgentMemoryRow, MemoryPersonaDraftResult } from '../types' +import type { + AgentMemoryRow, + MemoryMaintenancePersonaResult, + MemoryPersonaDraftResult +} from '../types' import { type MemoryModelRef, type MemoryRuntimeContext } from '../context' export class PersonaService { @@ -58,8 +62,23 @@ export class PersonaService { model: MemoryModelRef, sourceSession?: string | null ): Promise { + return (await this.runMaintenancePersonaPass(agentId, model, sourceSession)).result + } + + async runMaintenancePersonaPass( + agentId: string, + model: MemoryModelRef, + sourceSession?: string | null + ): Promise { + let calls = 0 + let failures = 0 + const finish = (result: MemoryPersonaDraftResult | null): MemoryMaintenancePersonaResult => ({ + result, + calls, + failures + }) if (!this.ctx.canWriteAgentMemory(agentId) || !this.ctx.isPersonaEvolutionEnabled(agentId)) { - return null + return finish(null) } try { return await this.withPersonaLock(agentId, async () => { @@ -67,14 +86,14 @@ export class PersonaService { !this.ctx.canWriteAgentMemory(agentId) || !this.ctx.isPersonaEvolutionEnabled(agentId) ) { - return null + return finish(null) } - if (this.ctx.deps.repository.getDraftPersona(agentId)) return null + if (this.ctx.deps.repository.getDraftPersona(agentId)) return finish(null) const units = this.ctx.deps.repository.listByAgent(agentId, { kinds: ['semantic', 'reflection', 'episodic'] }) - if (units.length < MIN_MEMORIES_FOR_PERSONA) return null + if (units.length < MIN_MEMORIES_FOR_PERSONA) return finish(null) const previous = this.ctx.deps.repository.getActivePersona(agentId) const watermark = Math.max( @@ -84,7 +103,7 @@ export class PersonaService { const recentImportance = units .filter((unit) => unit.created_at > watermark) .reduce((sum, unit) => sum + Math.min(1, Math.max(0, unit.importance)), 0) - if (recentImportance < PERSONA_EVOLUTION_IMPORTANCE_THRESHOLD) return null + if (recentImportance < PERSONA_EVOLUTION_IMPORTANCE_THRESHOLD) return finish(null) const maxUnitCreatedAt = units.reduce((max, unit) => Math.max(max, unit.created_at), 0) const top = units @@ -92,35 +111,42 @@ export class PersonaService { .sort((a, b) => b.importance - a.importance || b.created_at - a.created_at) .slice(0, PERSONA_MEMORY_LIMIT) const personaModel = this.ctx.resolveExtractionModel(agentId, model) - const raw = await this.ctx.deps.generateText( - personaModel.providerId, - personaModel.modelId, - buildReflectionPrompt( - previous?.content ?? null, - top.map((row) => row.content) + let raw = '' + try { + calls += 1 + raw = await this.ctx.deps.generateText( + personaModel.providerId, + personaModel.modelId, + buildReflectionPrompt( + previous?.content ?? null, + top.map((row) => row.content) + ) ) - ) + } catch (error) { + failures += 1 + throw error + } if ( !this.ctx.canWriteAgentMemory(agentId) || !this.ctx.isPersonaEvolutionEnabled(agentId) ) { - return null + return finish(null) } const content = sanitizeSelfModel(raw) if (!content || content === (previous?.content?.trim() ?? '')) { this.personaAttemptWatermark.set(agentId, maxUnitCreatedAt) - return null + return finish(null) } const changeRatio = personaChangeRatio(previous?.content ?? null, content) const needsReview = previous ? changeRatio > PERSONA_MAX_CHANGE_RATIO : false const draftId = this.evolvePersona(agentId, content, sourceSession ?? null) this.personaAttemptWatermark.set(agentId, maxUnitCreatedAt) - if (!draftId) return null - return { draftId, needsReview, changeRatio } + if (!draftId) return finish(null) + return finish({ draftId, needsReview, changeRatio }) }) } catch (error) { logger.warn(`[Memory] persona evolution skipped: ${String(error)}`) - return null + return finish(null) } } diff --git a/src/main/presenter/memoryPresenter/services/reflectionService.ts b/src/main/presenter/memoryPresenter/services/reflectionService.ts index 49e450768..e7737a116 100644 --- a/src/main/presenter/memoryPresenter/services/reflectionService.ts +++ b/src/main/presenter/memoryPresenter/services/reflectionService.ts @@ -9,7 +9,7 @@ import { } from '../runtimeConstants' import { buildMemoryProvenanceKey } from '../core/scoring' import { buildReflectionInsightsPrompt, parseReflectionInsights } from '../core/extraction' -import type { MemoryReflectionResult } from '../types' +import type { MemoryMaintenanceReflectionResult, MemoryReflectionResult } from '../types' import { isUniqueConstraintError, type MemoryModelRef, type MemoryRuntimeContext } from '../context' export class ReflectionService { @@ -28,12 +28,27 @@ export class ReflectionService { model: MemoryModelRef, sourceSession?: string | null ): Promise { - if (!this.ctx.canWriteAgentMemory(agentId)) return null + return (await this.runMaintenanceReflectionPass(agentId, model, sourceSession)).result + } + + async runMaintenanceReflectionPass( + agentId: string, + model: MemoryModelRef, + sourceSession?: string | null + ): Promise { + let calls = 0 + let failures = 0 + const finish = (result: MemoryReflectionResult | null): MemoryMaintenanceReflectionResult => ({ + result, + calls, + failures + }) + if (!this.ctx.canWriteAgentMemory(agentId)) return finish(null) try { const units = this.ctx.deps.repository.listByAgent(agentId, { kinds: ['episodic', 'semantic'] }) - if (units.length < MIN_MEMORIES_FOR_REFLECTION) return null + if (units.length < MIN_MEMORIES_FOR_REFLECTION) return finish(null) const lastReflection = this.ctx.deps.repository.listByAgent(agentId, { kinds: ['reflection'], @@ -46,7 +61,7 @@ export class ReflectionService { const recentImportance = units .filter((unit) => unit.created_at > watermark) .reduce((sum, unit) => sum + Math.min(1, Math.max(0, unit.importance)), 0) - if (recentImportance < REFLECTION_IMPORTANCE_THRESHOLD) return null + if (recentImportance < REFLECTION_IMPORTANCE_THRESHOLD) return finish(null) const maxUnitCreatedAt = units.reduce((max, unit) => Math.max(max, unit.created_at), 0) const top = units @@ -54,12 +69,19 @@ export class ReflectionService { .sort((a, b) => b.importance - a.importance || b.created_at - a.created_at) .slice(0, REFLECTION_MEMORY_LIMIT) const reflectionModel = this.ctx.resolveExtractionModel(agentId, model) - const raw = await this.ctx.deps.generateText( - reflectionModel.providerId, - reflectionModel.modelId, - buildReflectionInsightsPrompt(top.map((row) => row.content)) - ) - if (!this.ctx.canWriteAgentMemory(agentId)) return null + let raw = '' + try { + calls += 1 + raw = await this.ctx.deps.generateText( + reflectionModel.providerId, + reflectionModel.modelId, + buildReflectionInsightsPrompt(top.map((row) => row.content)) + ) + } catch (error) { + failures += 1 + throw error + } + if (!this.ctx.canWriteAgentMemory(agentId)) return finish(null) const insights = parseReflectionInsights(raw) const reflectionIds: string[] = [] for (const insight of insights) { @@ -68,7 +90,7 @@ export class ReflectionService { } if (!reflectionIds.length) { this.reflectionAttemptWatermark.set(agentId, maxUnitCreatedAt) - return null + return finish(null) } this.reflectionAttemptWatermark.delete(agentId) @@ -77,10 +99,10 @@ export class ReflectionService { void this.ports.triggerEmbedding(agentId).catch((error) => { logger.warn(`[Memory] background embedding failed: ${String(error)}`) }) - return { reflectionIds, sourceMemoryIds: top.map((row) => row.id) } + return finish({ reflectionIds, sourceMemoryIds: top.map((row) => row.id) }) } catch (error) { logger.warn(`[Memory] reflection skipped: ${String(error)}`) - return null + return finish(null) } } diff --git a/src/main/presenter/memoryPresenter/services/retrievalService.ts b/src/main/presenter/memoryPresenter/services/retrievalService.ts index 6e44cadc7..7422d6228 100644 --- a/src/main/presenter/memoryPresenter/services/retrievalService.ts +++ b/src/main/presenter/memoryPresenter/services/retrievalService.ts @@ -39,6 +39,17 @@ type QueryEmbeddingInFlight = { promise: Promise } +type CachedRecallKeywordTermStat = { + stat: { + term: string + hitCount: number + totalRows: number + } + expiresAt: number +} + +const RECALL_KEYWORD_STATS_TTL_MS = 30_000 + function isLiveRecallVectorRow( agentId: string, row: AgentMemoryRow | undefined @@ -78,6 +89,7 @@ async function withSoftTimeout( export class RetrievalService { private readonly queryEmbeddingInFlight = new Map>() + private readonly keywordStatsCache = new Map>() constructor( private readonly ctx: MemoryRuntimeContext, @@ -125,13 +137,54 @@ export class RetrievalService { private buildAgentFacingRecallKeywordQuery(agentId: string, query: string): string { const candidates = extractRecallKeywordCandidates(query) if (!candidates.length) return '' - const stats = this.ctx.deps.repository.getRecallKeywordTermStats( + const stats = this.getCachedRecallKeywordTermStats( agentId, candidates.map((candidate) => candidate.term) ) return buildRecallKeywordQuery(selectRecallKeywordTerms(candidates, stats)) } + private getCachedRecallKeywordTermStats(agentId: string, terms: string[]) { + const normalizedTerms = [...new Set(terms.map((term) => term.trim().toLowerCase()))].filter( + Boolean + ) + if (!normalizedTerms.length) return [] + + const now = Date.now() + let agentCache = this.keywordStatsCache.get(agentId) + if (!agentCache) { + agentCache = new Map() + this.keywordStatsCache.set(agentId, agentCache) + } + + const missing: string[] = [] + const cached = new Map() + for (const term of normalizedTerms) { + const entry = agentCache.get(term) + if (entry && entry.expiresAt > now) { + cached.set(term, entry.stat) + } else { + agentCache.delete(term) + missing.push(term) + } + } + + if (missing.length > 0) { + const fresh = this.ctx.deps.repository.getRecallKeywordTermStats(agentId, missing) + for (const stat of fresh) { + const normalizedTerm = stat.term.trim().toLowerCase() + const entry = { ...stat, term: normalizedTerm } + agentCache.set(normalizedTerm, { + stat: entry, + expiresAt: now + RECALL_KEYWORD_STATS_TTL_MS + }) + cached.set(normalizedTerm, entry) + } + } + + return normalizedTerms.map((term) => cached.get(term) ?? { term, hitCount: 0, totalRows: 0 }) + } + private startQueryEmbedding( agentId: string, embedding: MemoryModelRef, @@ -360,7 +413,7 @@ export class RetrievalService { const working = this.workingMemory.readWorkingMemory(agentId) if (!working) this.workingMemory.scheduleWorkingRefresh(agentId) const recalled = query.trim() - ? await this.retrieve(agentId, query, Date.now(), true, { + ? await this.retrieve(agentId, query, Date.now(), false, { keywordQuery: this.buildAgentFacingRecallKeywordQuery(agentId, query), keywordMatchMode: 'any' }) @@ -396,9 +449,15 @@ export class RetrievalService { for (const key of this.queryEmbeddingInFlight.keys()) { if (key.startsWith(`${agentId}::`)) this.queryEmbeddingInFlight.delete(key) } + this.invalidateKeywordStats(agentId) } clearAll(): void { this.queryEmbeddingInFlight.clear() + this.keywordStatsCache.clear() + } + + invalidateKeywordStats(agentId: string): void { + this.keywordStatsCache.delete(agentId) } } diff --git a/src/main/presenter/memoryPresenter/services/writeCoordinator.ts b/src/main/presenter/memoryPresenter/services/writeCoordinator.ts index 0cb0e1614..77a61152d 100644 --- a/src/main/presenter/memoryPresenter/services/writeCoordinator.ts +++ b/src/main/presenter/memoryPresenter/services/writeCoordinator.ts @@ -165,17 +165,11 @@ export class WriteCoordinator { const provenanceKey = buildMemoryProvenanceKey(options.agentId, normalized.kind, content) const duplicate = this.ctx.deps.repository.getByProvenanceKey(options.agentId, provenanceKey) if (duplicate) { - const hit = this.rows.handleProvenanceHit(options.agentId, duplicate, { - allowDecisionForSuperseded: true - }) + const hit = this.rows.handleProvenanceHit(options.agentId, duplicate) if (hit.action === 'absorbed') { this.absorbArchivedProvenanceOwner(duplicate) created.push(duplicate.id) } - if (hit.action === 'continue') { - this.reviveProvenanceOwner(options.agentId, duplicate, Date.now(), normalized.category) - created.push(duplicate.id) - } continue } const id = this.rows.insertMemory( @@ -191,10 +185,10 @@ export class WriteCoordinator { } async extractAndStore(input: MemoryExtractionInput): Promise { - if (!this.ctx.canWriteAgentMemory(input.agentId)) return { ok: true, createdIds: [] } - if (this.ctx.isDisposed) return { ok: true, createdIds: [] } const span = input.spanText.trim() if (!span) return { ok: true, createdIds: [] } + if (!this.ctx.canWriteAgentMemory(input.agentId)) return { ok: false } + if (this.ctx.isDisposed) return { ok: false } const model = this.ctx.resolveExtractionModel(input.agentId, input.model) try { let shouldExtract = true @@ -208,7 +202,7 @@ export class WriteCoordinator { } catch (error) { logger.warn(`[Memory] triage skipped, extracting anyway: ${String(error)}`) } - if (!this.ctx.canWriteAgentMemory(input.agentId)) return { ok: true, createdIds: [] } + if (!this.ctx.canWriteAgentMemory(input.agentId)) return { ok: false } if (!shouldExtract) return { ok: true, createdIds: [] } const response = await this.ctx.deps.generateText( @@ -216,7 +210,7 @@ export class WriteCoordinator { model.modelId, buildExtractionPrompt(span) ) - if (!this.ctx.canWriteAgentMemory(input.agentId)) return { ok: true, createdIds: [] } + if (!this.ctx.canWriteAgentMemory(input.agentId)) return { ok: false } const parsed = parseMemoryCandidates(response) if (!parsed.ok) { logger.warn(`[Memory] extraction parse failed: ${parsed.reason}`) @@ -234,6 +228,9 @@ export class WriteCoordinator { let touched = false for (const candidate of candidates) { const outcome = await this.coordinateWrite(input.agentId, candidate, model, options, now) + // If a non-empty extraction is disabled mid-batch, keep the cursor for retry; any rows + // already written are picked up by the next embedding/backfill drain. + if (!this.ctx.canWriteAgentMemory(input.agentId)) return { ok: false } outcomes.push(outcome) createdIds.push(...createdIdsFromOutcome(outcome)) if (outcomeTouched(outcome)) touched = true @@ -470,17 +467,13 @@ export class WriteCoordinator { const provenanceKey = buildMemoryProvenanceKey(agentId, normalized.kind, content) const duplicate = this.ctx.deps.repository.getByProvenanceKey(agentId, provenanceKey) if (duplicate) { - const hit = this.rows.handleProvenanceHit(agentId, duplicate, { - allowDecisionForSuperseded: true - }) + const hit = this.rows.handleProvenanceHit(agentId, duplicate) if (hit.action === 'absorbed') { this.absorbArchivedProvenanceOwner(duplicate) return { action: 'updated', id: duplicate.id } } - if (hit.action === 'continue') { - return this.reviveProvenanceOwner(agentId, duplicate, Date.now(), normalized.category) - } - return { action: 'noop', reason: hit.reason, id: duplicate.id } + const reason = hit.action === 'noop' ? hit.reason : 'duplicate' + return { action: 'noop', reason, id: duplicate.id } } const id = this.rows.insertMemory(agentId, normalized, content, provenanceKey, options) return id ? { action: 'created', id } : { action: 'noop', reason: 'insert-skipped' } diff --git a/src/main/presenter/memoryPresenter/types.ts b/src/main/presenter/memoryPresenter/types.ts index 21af17702..714807a54 100644 --- a/src/main/presenter/memoryPresenter/types.ts +++ b/src/main/presenter/memoryPresenter/types.ts @@ -211,7 +211,6 @@ export interface MemoryVectorMatch { export interface MemoryVectorQueryOptions { topK: number - threshold?: number } export interface MemoryVectorRef { @@ -418,6 +417,24 @@ export interface MemoryReflectionResult { sourceMemoryIds: string[] } +export interface MemoryMaintenanceStepResult { + touched: boolean + calls: number + failures: number +} + +export interface MemoryMaintenanceReflectionResult { + result: MemoryReflectionResult | null + calls: number + failures: number +} + +export interface MemoryMaintenancePersonaResult { + result: MemoryPersonaDraftResult | null + calls: number + failures: number +} + // Outcome of a guarded persona-evolution pass: the new draft (never auto-active) plus the measured // drift from the current self-model. needsReview gates the draft out of any auto-approval path. export interface MemoryPersonaDraftResult { diff --git a/src/main/presenter/sqlitePresenter/schemaCatalog.ts b/src/main/presenter/sqlitePresenter/schemaCatalog.ts index 07bf89bc6..868444538 100644 --- a/src/main/presenter/sqlitePresenter/schemaCatalog.ts +++ b/src/main/presenter/sqlitePresenter/schemaCatalog.ts @@ -249,7 +249,13 @@ const CATALOG_DEFINITIONS: CatalogDefinition[] = [ }, { name: 'agent_memory_audit', - createTable: (db) => new AgentMemoryAuditTable(db) + createTable: (db) => new AgentMemoryAuditTable(db), + repairableColumns: { + memory_ref_id: 'ALTER TABLE agent_memory_audit ADD COLUMN memory_ref_id TEXT;' + }, + afterRepair: (db) => { + new AgentMemoryAuditTable(db).backfillMemoryRefIds() + } }, { name: 'new_session_active_skills', diff --git a/src/main/presenter/sqlitePresenter/tables/agentMemory.ts b/src/main/presenter/sqlitePresenter/tables/agentMemory.ts index eba614d7e..e0500cd05 100644 --- a/src/main/presenter/sqlitePresenter/tables/agentMemory.ts +++ b/src/main/presenter/sqlitePresenter/tables/agentMemory.ts @@ -1217,8 +1217,8 @@ export class AgentMemoryTable extends BaseTable { this.db.prepare("UPDATE agent_memory SET status = 'archived' WHERE id = ?").run(id) } - // SQL-expressible subset of the archive conditions: active, aged out, decayed, and exempt rows - // (anchors / persona) excluded. The zero-interaction check runs in the caller on this result. + // SQL-expressible subset of the archive conditions: active, never accessed, aged out, decayed, + // and exempt rows (anchors / persona) excluded. The caller may still defensively recheck. listArchiveCandidates(agentId: string, before: number, decayBelow: number): AgentMemoryRow[] { return this.db .prepare( @@ -1229,6 +1229,7 @@ export class AgentMemoryTable extends BaseTable { AND status != 'conflicted' AND is_anchor = 0 AND kind NOT IN ('persona', 'working') + AND access_count = 0 AND created_at < ? AND decay_score IS NOT NULL AND decay_score < ?` diff --git a/src/main/presenter/sqlitePresenter/tables/agentMemoryAudit.ts b/src/main/presenter/sqlitePresenter/tables/agentMemoryAudit.ts index d9da24de2..639be27c6 100644 --- a/src/main/presenter/sqlitePresenter/tables/agentMemoryAudit.ts +++ b/src/main/presenter/sqlitePresenter/tables/agentMemoryAudit.ts @@ -10,6 +10,7 @@ export interface AgentMemoryAuditRow { event_type: string actor_type: AgentMemoryAuditActorType session_id: string | null + memory_ref_id: string | null input_refs_json: string output_refs_json: string model_provider_id: string | null @@ -58,19 +59,71 @@ export interface AgentMemoryHealthAuditStats { recentFailures: AgentMemoryHealthRecentFailureRow[] } -const AGENT_MEMORY_AUDIT_SCHEMA_VERSION = 36 +const AGENT_MEMORY_AUDIT_SCHEMA_VERSION = 38 const AGENT_MEMORY_AUDIT_INDEX_SQL = ` CREATE INDEX IF NOT EXISTS idx_agent_memory_audit_agent_created ON agent_memory_audit(agent_id, created_at); CREATE INDEX IF NOT EXISTS idx_agent_memory_audit_agent_event ON agent_memory_audit(agent_id, event_type, created_at); + CREATE INDEX IF NOT EXISTS idx_agent_memory_audit_agent_memory_ref + ON agent_memory_audit(agent_id, memory_ref_id, created_at); +` + +const AGENT_MEMORY_AUDIT_OUTPUT_MEMORY_REF_SQL = ` + CASE + WHEN json_valid(output_refs_json) THEN + CASE + WHEN json_type(output_refs_json, '$.memoryId') = 'text' THEN + NULLIF(TRIM(CAST(json_extract(output_refs_json, '$.memoryId') AS TEXT)), '') + END + END +` + +const AGENT_MEMORY_AUDIT_INPUT_MEMORY_REF_SQL = ` + CASE + WHEN json_valid(input_refs_json) THEN + CASE + WHEN json_type(input_refs_json, '$.memoryId') = 'text' THEN + NULLIF(TRIM(CAST(json_extract(input_refs_json, '$.memoryId') AS TEXT)), '') + END + END +` + +const AGENT_MEMORY_AUDIT_BACKFILL_MEMORY_REF_SQL = ` + UPDATE agent_memory_audit + SET memory_ref_id = COALESCE( + ${AGENT_MEMORY_AUDIT_OUTPUT_MEMORY_REF_SQL}, + ${AGENT_MEMORY_AUDIT_INPUT_MEMORY_REF_SQL} + ) + WHERE memory_ref_id IS NULL + AND status = 'completed' + AND ( + (event_type = 'memory/forget' AND actor_type = 'runtime') + OR (event_type = 'memory/archive' AND actor_type = 'user') + OR event_type = 'memory/restore' + ) + AND COALESCE( + ${AGENT_MEMORY_AUDIT_OUTPUT_MEMORY_REF_SQL}, + ${AGENT_MEMORY_AUDIT_INPUT_MEMORY_REF_SQL} + ) IS NOT NULL; ` function stringifyMetadata(value: Record | undefined): string { return JSON.stringify(value ?? {}) } +function extractSingleMemoryRef(metadata: Record | undefined): string | null { + const memoryId = metadata?.memoryId + if (typeof memoryId !== 'string') return null + const trimmed = memoryId.trim() + return trimmed || null +} + +function deriveMemoryRefId(input: AgentMemoryAuditInsertInput): string | null { + return extractSingleMemoryRef(input.outputRefs) ?? extractSingleMemoryRef(input.inputRefs) +} + function metadataReferencesMemoryId(metadataJson: string, memoryId: string): boolean { try { const metadata = JSON.parse(metadataJson) as Record @@ -93,6 +146,7 @@ export class AgentMemoryAuditTable extends BaseTable { event_type TEXT NOT NULL, actor_type TEXT NOT NULL, session_id TEXT, + memory_ref_id TEXT, input_refs_json TEXT NOT NULL DEFAULT '{}', output_refs_json TEXT NOT NULL DEFAULT '{}', model_provider_id TEXT, @@ -114,6 +168,13 @@ export class AgentMemoryAuditTable extends BaseTable { if (version === 36) { return this.getCreateTableSQL() } + if (version === 38) { + return ` + ALTER TABLE agent_memory_audit ADD COLUMN memory_ref_id TEXT; + ${AGENT_MEMORY_AUDIT_BACKFILL_MEMORY_REF_SQL} + ${AGENT_MEMORY_AUDIT_INDEX_SQL} + ` + } return null } @@ -121,6 +182,10 @@ export class AgentMemoryAuditTable extends BaseTable { return AGENT_MEMORY_AUDIT_SCHEMA_VERSION } + backfillMemoryRefIds(): void { + this.db.exec(AGENT_MEMORY_AUDIT_BACKFILL_MEMORY_REF_SQL) + } + insert(input: AgentMemoryAuditInsertInput): AgentMemoryAuditRow { const row: AgentMemoryAuditRow = { id: input.id, @@ -128,6 +193,7 @@ export class AgentMemoryAuditTable extends BaseTable { event_type: input.eventType, actor_type: input.actorType, session_id: input.sessionId ?? null, + memory_ref_id: deriveMemoryRefId(input), input_refs_json: stringifyMetadata(input.inputRefs), output_refs_json: stringifyMetadata(input.outputRefs), model_provider_id: input.modelProviderId ?? null, @@ -145,6 +211,7 @@ export class AgentMemoryAuditTable extends BaseTable { event_type, actor_type, session_id, + memory_ref_id, input_refs_json, output_refs_json, model_provider_id, @@ -153,7 +220,7 @@ export class AgentMemoryAuditTable extends BaseTable { reason, created_at ) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)` ) .run( row.id, @@ -161,6 +228,7 @@ export class AgentMemoryAuditTable extends BaseTable { row.event_type, row.actor_type, row.session_id, + row.memory_ref_id, row.input_refs_json, row.output_refs_json, row.model_provider_id, @@ -234,14 +302,29 @@ export class AgentMemoryAuditTable extends BaseTable { } hasForgetEvent(agentId: string, memoryId: string): boolean { + return this.hasForgetEventFromRows(this.listForgetEventRows(agentId, memoryId), memoryId) + } + + private listForgetEventRows( + agentId: string, + memoryId: string + ): Array<{ + event_type: string + actor_type: AgentMemoryAuditActorType + memory_ref_id: string | null + input_refs_json: string + output_refs_json: string + }> { const rows = this.db .prepare( `SELECT event_type, actor_type, + memory_ref_id, input_refs_json, output_refs_json FROM agent_memory_audit WHERE agent_id = ? + AND (memory_ref_id = ? OR memory_ref_id IS NULL) AND status = 'completed' AND ( (event_type = 'memory/forget' AND actor_type = 'runtime') @@ -250,17 +333,32 @@ export class AgentMemoryAuditTable extends BaseTable { ) ORDER BY created_at DESC, id DESC` ) - .all(agentId) as Array<{ + .all(agentId, memoryId) as Array<{ event_type: string actor_type: AgentMemoryAuditActorType + memory_ref_id: string | null input_refs_json: string output_refs_json: string }> + return rows + } + + private hasForgetEventFromRows( + rows: Array<{ + event_type: string + actor_type: AgentMemoryAuditActorType + memory_ref_id: string | null + input_refs_json: string + output_refs_json: string + }>, + memoryId: string + ): boolean { for (const row of rows) { - if ( - !metadataReferencesMemoryId(row.input_refs_json, memoryId) && - !metadataReferencesMemoryId(row.output_refs_json, memoryId) - ) { + const referencesMemory = + row.memory_ref_id === memoryId || + metadataReferencesMemoryId(row.input_refs_json, memoryId) || + metadataReferencesMemoryId(row.output_refs_json, memoryId) + if (!referencesMemory) { continue } if (row.event_type === 'memory/restore') return false diff --git a/test/main/presenter/agentMemoryTable.test.ts b/test/main/presenter/agentMemoryTable.test.ts index b35c74c07..78098109c 100644 --- a/test/main/presenter/agentMemoryTable.test.ts +++ b/test/main/presenter/agentMemoryTable.test.ts @@ -1979,6 +1979,158 @@ describeIfSqlite('AgentMemoryTable FTS5 + migration', () => { } }) + it('agent memory audit derives memory_ref_id and falls back for legacy rows', () => { + const db = new DatabaseCtor(':memory:') + try { + const table = new AgentMemoryAuditTableCtor(db) + table.createTable() + table.insert({ + id: 'forget-indexed', + agentId: 'a', + eventType: 'memory/forget', + actorType: 'runtime', + status: 'completed', + inputRefs: { memoryId: 'm-indexed' }, + outputRefs: { memoryId: 'm-indexed' }, + createdAt: 100 + }) + expect( + db + .prepare('SELECT memory_ref_id FROM agent_memory_audit WHERE id = ?') + .get('forget-indexed') + ).toEqual({ memory_ref_id: 'm-indexed' }) + + table.insert({ + id: 'forget-legacy', + agentId: 'a', + eventType: 'memory/forget', + actorType: 'runtime', + status: 'completed', + inputRefs: { memoryId: 'm-legacy' }, + outputRefs: { memoryId: 'm-legacy' }, + createdAt: 200 + }) + db.prepare( + "UPDATE agent_memory_audit SET memory_ref_id = NULL WHERE id = 'forget-legacy'" + ).run() + + expect(table.hasForgetEvent('a', 'm-legacy')).toBe(true) + } finally { + db.close() + } + }) + + it('agent memory audit orders indexed and legacy memory refs together', () => { + const db = new DatabaseCtor(':memory:') + try { + const table = new AgentMemoryAuditTableCtor(db) + table.createTable() + table.insert({ + id: 'indexed-restore', + agentId: 'a', + eventType: 'memory/restore', + actorType: 'user', + status: 'completed', + inputRefs: { memoryId: 'm-mixed' }, + outputRefs: { memoryId: 'm-mixed' }, + createdAt: 100 + }) + table.insert({ + id: 'legacy-forget', + agentId: 'a', + eventType: 'memory/forget', + actorType: 'runtime', + status: 'completed', + inputRefs: { memoryId: 'm-mixed' }, + outputRefs: { memoryId: 'm-mixed' }, + createdAt: 200 + }) + db.prepare( + "UPDATE agent_memory_audit SET memory_ref_id = NULL WHERE id = 'legacy-forget'" + ).run() + expect(table.hasForgetEvent('a', 'm-mixed')).toBe(true) + + table.insert({ + id: 'legacy-restore', + agentId: 'a', + eventType: 'memory/restore', + actorType: 'user', + status: 'completed', + inputRefs: { memoryId: 'm-mixed' }, + outputRefs: { memoryId: 'm-mixed' }, + createdAt: 300 + }) + db.prepare( + "UPDATE agent_memory_audit SET memory_ref_id = NULL WHERE id = 'legacy-restore'" + ).run() + expect(table.hasForgetEvent('a', 'm-mixed')).toBe(false) + + table.insert({ + id: 'indexed-archive', + agentId: 'a', + eventType: 'memory/archive', + actorType: 'user', + status: 'completed', + inputRefs: { memoryId: 'm-mixed' }, + outputRefs: { memoryId: 'm-mixed' }, + createdAt: 400 + }) + expect(table.hasForgetEvent('a', 'm-mixed')).toBe(true) + } finally { + db.close() + } + }) + + it('agent memory audit migration tolerates malformed refs while backfilling memory_ref_id', () => { + const db = new DatabaseCtor(':memory:') + try { + const table = new AgentMemoryAuditTableCtor(db) + db.exec(` + CREATE TABLE agent_memory_audit ( + id TEXT PRIMARY KEY, + agent_id TEXT NOT NULL, + event_type TEXT NOT NULL, + actor_type TEXT NOT NULL, + session_id TEXT, + input_refs_json TEXT NOT NULL DEFAULT '{}', + output_refs_json TEXT NOT NULL DEFAULT '{}', + model_provider_id TEXT, + model_id TEXT, + status TEXT NOT NULL, + reason TEXT, + created_at INTEGER NOT NULL + ); + `) + const insert = db.prepare( + `INSERT INTO agent_memory_audit ( + id, + agent_id, + event_type, + actor_type, + input_refs_json, + output_refs_json, + status, + created_at + ) + VALUES (?, 'a', 'memory/forget', 'runtime', ?, ?, 'completed', ?)` + ) + insert.run('valid-input', '{"memoryId":"m-input"}', 'not-json', 100) + insert.run('malformed-only', 'not-json', 'also-not-json', 200) + + expect(() => db.exec(table.getMigrationSQL(38) ?? '')).not.toThrow() + expect( + db.prepare('SELECT memory_ref_id FROM agent_memory_audit WHERE id = ?').get('valid-input') + ).toEqual({ memory_ref_id: 'm-input' }) + expect( + db + .prepare('SELECT memory_ref_id FROM agent_memory_audit WHERE id = ?') + .get('malformed-only') + ).toEqual({ memory_ref_id: null }) + } finally { + db.close() + } + }) + it('agent memory audit hasForgetEvent does not miss older memory refs behind newer events', () => { const db = new DatabaseCtor(':memory:') try { @@ -2133,7 +2285,7 @@ describeIfSqlite('AgentMemoryTable FTS5 + migration', () => { table.updateDecayScore('anchored', 0.01) const candidates = table.listArchiveCandidates('a', 5000, 0.05) - expect(candidates.map((r) => r.id).sort()).toEqual(['accessed', 'stale']) + expect(candidates.map((r) => r.id).sort()).toEqual(['stale']) expect(table.countArchiveCandidates('a', 5000, 0.05)).toBe(1) } finally { db.close() diff --git a/test/main/presenter/agentRuntimePresenter/agentRuntimePresenter.test.ts b/test/main/presenter/agentRuntimePresenter/agentRuntimePresenter.test.ts index 20481ac57..c94ddbb5c 100644 --- a/test/main/presenter/agentRuntimePresenter/agentRuntimePresenter.test.ts +++ b/test/main/presenter/agentRuntimePresenter/agentRuntimePresenter.test.ts @@ -689,6 +689,7 @@ describe('AgentRuntimePresenter', () => { sqlitePresenter.newSessionsTable.get.mockReturnValue({ agent_id: 'a' }) ;(agent as any).memoryPort = { isEnabled: vi.fn(() => true), + recordInjectionAccess: vi.fn(), buildInjection: vi.fn(async () => ({ selfModel: null, working: null, @@ -721,6 +722,92 @@ describe('AgentRuntimePresenter', () => { expect.objectContaining({ sessionId: 's1', name: 'memory/view_assembled' }) ) }) + + it('records access only for selected injected memories and dedupes by message', async () => { + sqlitePresenter.newSessionsTable.get.mockReturnValue({ agent_id: 'a' }) + const recordInjectionAccess = vi.fn() + ;(agent as any).memoryPort = { + isEnabled: vi.fn(() => true), + recordInjectionAccess, + buildInjection: vi.fn(async () => ({ + selfModel: null, + working: null, + memories: [ + { id: 'selected', kind: 'semantic', content: 'redis fact' }, + { id: 'dropped', kind: 'semantic', content: 'x'.repeat(10_000) } + ], + payload: { + selfModel: null, + working: null, + memories: [ + { id: 'selected', kind: 'semantic', content: 'redis fact' }, + { id: 'dropped', kind: 'semantic', content: 'x'.repeat(10_000) } + ], + tokenBudget: 80 + }, + manifest: { + policyVersion: 1, + selected: [], + dropped: [], + tokenBudget: 80, + estimatedTokens: 0, + queryHash: 'query-hash' + } + })) + } + + await (agent as any).appendMemoryInjection('s1', 'base prompt', 'redis', 'user-message-1') + await (agent as any).appendMemoryInjection('s1', 'base prompt', 'redis', 'user-message-1') + await (agent as any).appendMemoryInjection('s1', 'base prompt', 'redis', null) + await (agent as any).appendMemoryInjection('s1', 'base prompt', 'redis', null) + + expect(recordInjectionAccess.mock.calls).toEqual([ + ['a', ['selected']], + ['a', ['selected']], + ['a', ['selected']] + ]) + }) + + it('bounds injection access dedupe state per session and clears it on destroy', async () => { + const recordInjectionAccess = vi.fn() + ;(agent as any).memoryPort = { + recordInjectionAccess + } + + for (let index = 0; index < 130; index += 1) { + ;(agent as any).recordMemoryInjectionAccess( + 'a', + 's1', + [{ id: `selected-${index}` }], + `message-${index}` + ) + } + + const accessByTurn = (agent as any).memoryInjectionAccessByTurn as Map + const sessionKeys = [...accessByTurn.keys()].filter((key) => key.startsWith('s1\u0000')) + expect(sessionKeys).toHaveLength(128) + expect(recordInjectionAccess).toHaveBeenCalledTimes(130) + + await agent.destroySession('s1') + expect([...accessByTurn.keys()].filter((key) => key.startsWith('s1\u0000'))).toHaveLength(0) + }) + + it('expires old injection access dedupe turns for active sessions', () => { + const nowSpy = vi.spyOn(Date, 'now').mockReturnValue(1_000) + ;(agent as any).memoryPort = { + recordInjectionAccess: vi.fn() + } + + ;(agent as any).recordMemoryInjectionAccess('a', 's1', [{ id: 'old' }], 'old-message') + nowSpy.mockReturnValue(31 * 60 * 1000) + ;(agent as any).recordMemoryInjectionAccess('a', 's1', [{ id: 'new' }], 'new-message') + + const accessByTurn = (agent as any).memoryInjectionAccessByTurn as Map + expect([...accessByTurn.keys()].filter((key) => key.startsWith('s1\u0000'))).toEqual([ + 's1\u0000new-message' + ]) + nowSpy.mockRestore() + }) }) describe('memory extraction lifecycle', () => { @@ -923,6 +1010,25 @@ describe('AgentRuntimePresenter', () => { ) }) + it('keeps the cursor unchanged when extraction returns ok:false', async () => { + await agent.initSession('s1', { providerId: 'openai', modelId: 'gpt-4' }) + const extractAndStore = vi.fn().mockResolvedValue({ ok: false }) + ;(agent as any).memoryPort = { + isEnabled: vi.fn(() => true), + extractAndStore + } + sqlitePresenter.deepchatSessionsTable.updateMemoryCursorOrderSeq.mockClear() + sqlitePresenter.deepchatTapeEntriesTable.appendAnchor.mockClear() + + await startExtraction(10) + + expect(extractAndStore).toHaveBeenCalledTimes(1) + expect( + sqlitePresenter.deepchatSessionsTable.updateMemoryCursorOrderSeq + ).not.toHaveBeenCalled() + expect(sqlitePresenter.deepchatTapeEntriesTable.appendAnchor).not.toHaveBeenCalled() + }) + it('computes tool admission signals from one tape read', async () => { installRuntimeRecords([ userRecord('u1', 1, 'Read package metadata.'), diff --git a/test/main/presenter/fakes/memoryFakes.ts b/test/main/presenter/fakes/memoryFakes.ts index 70de655e7..0508997bf 100644 --- a/test/main/presenter/fakes/memoryFakes.ts +++ b/test/main/presenter/fakes/memoryFakes.ts @@ -521,6 +521,7 @@ export class FakeRepository implements MemoryRepositoryPort { row.is_anchor === 0 && row.kind !== 'persona' && row.kind !== 'working' && + row.access_count === 0 && row.created_at < before && row.decay_score !== null && row.decay_score < decayBelow @@ -553,9 +554,7 @@ export class FakeRepository implements MemoryRepositoryPort { } countArchiveCandidates(agentId: string, before: number, decayBelow: number) { - return this.listArchiveCandidates(agentId, before, decayBelow).filter( - (row) => row.access_count === 0 - ).length + return this.listArchiveCandidates(agentId, before, decayBelow).length } listTopAccessed(agentId: string, limit: number) { @@ -784,6 +783,12 @@ export class FakeAuditRepository implements MemoryAuditRepositoryPort { event_type: input.eventType, actor_type: input.actorType, session_id: input.sessionId ?? null, + memory_ref_id: + typeof input.outputRefs?.memoryId === 'string' + ? input.outputRefs.memoryId + : typeof input.inputRefs?.memoryId === 'string' + ? input.inputRefs.memoryId + : null, input_refs_json: JSON.stringify(input.inputRefs ?? {}), output_refs_json: JSON.stringify(input.outputRefs ?? {}), model_provider_id: input.modelProviderId ?? null, diff --git a/test/main/presenter/memoryAdd.test.ts b/test/main/presenter/memoryAdd.test.ts index b5c179efe..7e7de47ca 100644 --- a/test/main/presenter/memoryAdd.test.ts +++ b/test/main/presenter/memoryAdd.test.ts @@ -307,4 +307,36 @@ describe('MemoryPresenter.addUserMemory (manual user write)', () => { expect(repo.listByAgent('deepchat')).toHaveLength(1) expect(repo.listByAgent('deepchat')[0]).toMatchObject({ kind: 'semantic', category: null }) }) + + it('keeps no-model superseded provenance hits as conservative noops', async () => { + const { presenter, repo } = makePresenter(enabledConfig) + const [ownerId] = presenter.writeMemoriesSync( + [{ kind: 'semantic', content: 'user likes redis' }], + { + agentId: 'deepchat' + } + ) + const [headId] = presenter.writeMemoriesSync( + [{ kind: 'semantic', content: 'user likes postgres' }], + { + agentId: 'deepchat' + } + ) + repo.markSuperseded(ownerId, headId) + + expect( + presenter.writeMemoriesSync([{ kind: 'semantic', content: 'user likes redis' }], { + agentId: 'deepchat' + }) + ).toEqual([]) + expect(repo.getById(ownerId)?.superseded_by).toBe(headId) + + await expect( + presenter.addUserMemory('deepchat', { + kind: 'semantic', + content: 'user likes redis' + }) + ).resolves.toEqual({ action: 'noop', reason: 'duplicate', id: ownerId }) + expect(repo.getById(ownerId)?.superseded_by).toBe(headId) + }) }) diff --git a/test/main/presenter/memoryExtraction.test.ts b/test/main/presenter/memoryExtraction.test.ts index 4ae8a3fc2..41c4a8074 100644 --- a/test/main/presenter/memoryExtraction.test.ts +++ b/test/main/presenter/memoryExtraction.test.ts @@ -140,7 +140,7 @@ describe('buildExtractionPrompt', () => { // extractAndStore end-to-end (fake LLM + fake repo): exercises the decoupled extraction chain. describe('MemoryPresenter.extractAndStore', () => { - it('extracts, dedupes, and writes pending memories; no-op when disabled', async () => { + it('extracts, dedupes, and writes pending memories; returns retryable failure when disabled', async () => { const { MemoryPresenter } = await import('@/presenter/memoryPresenter') const repo = makeFakeRepo() const generateText = vi.fn( @@ -165,13 +165,21 @@ describe('MemoryPresenter.extractAndStore', () => { }) }) - // disabled → no LLM call, no writes + // Disabled non-empty spans are retryable because they may have been queued while enabled. const none = await presenter.extractAndStore({ agentId: 'off', spanText: 'User: hi', model: { providerId: 'p', modelId: 'm' } }) - expect(none).toEqual({ ok: true, createdIds: [] }) + expect(none).toEqual({ ok: false }) + expect(generateText).not.toHaveBeenCalled() + + const empty = await presenter.extractAndStore({ + agentId: 'off', + spanText: ' ', + model: { providerId: 'p', modelId: 'm' } + }) + expect(empty).toEqual({ ok: true, createdIds: [] }) expect(generateText).not.toHaveBeenCalled() // enabled → extracts and writes @@ -331,6 +339,39 @@ describe('MemoryPresenter.extractAndStore triage gate, cheap model, lineage', () expect(repo.countByAgent('a')).toBe(0) }) + it('returns ok:false when memory is disabled while a non-empty span is awaiting triage', async () => { + let memoryEnabled = true + let resolveTriage: (value: string) => void = () => {} + const triageGate = new Promise((resolve) => { + resolveTriage = resolve + }) + const generateText = vi.fn(async (_p: string, _m: string, prompt: string) => { + if (prompt.includes('KEEP or SKIP')) return triageGate + return '[{"kind":"semantic","content":"user prefers redis"}]' + }) + const { presenter, repo } = await build( + { + get memoryEnabled() { + return memoryEnabled + } + }, + generateText + ) + + const extraction = presenter.extractAndStore({ + agentId: 'a', + spanText: 'User: I prefer redis', + model: { providerId: 'p', modelId: 'm' } + }) + await Promise.resolve() + memoryEnabled = false + resolveTriage('KEEP') + + await expect(extraction).resolves.toEqual({ ok: false }) + expect(generateText).toHaveBeenCalledTimes(1) + expect(repo.countByAgent('a')).toBe(0) + }) + it('falls through to extraction when triage itself fails', async () => { let call = 0 const generateText = vi.fn(async () => { diff --git a/test/main/presenter/memoryPresenter.test.ts b/test/main/presenter/memoryPresenter.test.ts index 3223afbdb..53e07653a 100644 --- a/test/main/presenter/memoryPresenter.test.ts +++ b/test/main/presenter/memoryPresenter.test.ts @@ -943,6 +943,39 @@ describe('MemoryPresenter recall + injection', () => { expect(payload?.memories.length).toBeGreaterThan(0) }) + it('records injection access only through the runtime-selected manifest seam', async () => { + const { presenter, repo } = makePresenter(enabledConfig) + const [id] = presenter.writeMemoriesSync([{ kind: 'semantic', content: 'redis fact' }], { + agentId: 'a' + }) + await presenter.processPendingEmbeddings('a') + + const payload = await presenter.buildInjection('a', 'redis') + expect(payload?.memories.map((memory) => memory.id)).toContain(id) + expect(repo.getById(id)?.access_count).toBe(0) + + presenter.recordInjectionAccess('a', [id, id], 1234) + expect(repo.getById(id)?.access_count).toBe(1) + expect(repo.getById(id)?.last_accessed).toBe(1234) + }) + + it('records injection access only for rows owned by the requested agent', () => { + const { presenter, repo } = makePresenter(enabledConfig) + const [aId] = presenter.writeMemoriesSync([{ kind: 'semantic', content: 'a redis fact' }], { + agentId: 'a' + }) + const [bId] = presenter.writeMemoriesSync([{ kind: 'semantic', content: 'b redis fact' }], { + agentId: 'b' + }) + + presenter.recordInjectionAccess('a', [aId, bId], 1234) + + expect(repo.getById(aId)?.access_count).toBe(1) + expect(repo.getById(aId)?.last_accessed).toBe(1234) + expect(repo.getById(bId)?.access_count).toBe(0) + expect(repo.getById(bId)?.last_accessed).toBeNull() + }) + it('buildInjection does not request heavy retrieval breakdown by default', async () => { const { presenter } = makePresenter(enabledConfig) presenter.writeMemoriesSync([{ kind: 'semantic', content: 'redis fact' }], { agentId: 'a' }) @@ -1811,6 +1844,28 @@ describe('MemoryPresenter management', () => { expect(recalled.map((item) => item.id)).toEqual(['m1']) }) + it('caches recall keyword stats per agent term and invalidates them on writes', async () => { + const { presenter, repo } = makePresenter({ memoryEnabled: true }) + repo.insert({ + id: 'm1', + agentId: 'a', + kind: 'semantic', + content: 'redis setup', + status: 'fts_only' + }) + const statsSpy = vi.spyOn(repo, 'getRecallKeywordTermStats') + + await presenter.recall('a', 'please redis setup') + await presenter.recall('a', 'please redis setup') + expect(statsSpy).toHaveBeenCalledTimes(1) + + presenter.writeMemoriesSync([{ kind: 'semantic', content: 'redis follow-up' }], { + agentId: 'a' + }) + await presenter.recall('a', 'please redis setup') + expect(statsSpy).toHaveBeenCalledTimes(2) + }) + it('agent-facing recall keeps a single domain term even when it is frequent', async () => { const { presenter, repo } = makePresenter({ memoryEnabled: true }) for (const id of ['m1', 'm2', 'm3', 'm4']) { @@ -3459,7 +3514,7 @@ describe('MemoryPresenter async write guards', () => { spanText: 'User: remember this later', model: { providerId: 'p', modelId: 'm' } }) - ).resolves.toEqual({ ok: true, createdIds: [] }) + ).resolves.toEqual({ ok: false }) expect(generateText).not.toHaveBeenCalled() expect(repo.countByAgent('a')).toBe(0) @@ -3503,7 +3558,7 @@ describe('MemoryPresenter async write guards', () => { managed = false releaseExtraction() - await expect(pending).resolves.toEqual({ ok: true, createdIds: [] }) + await expect(pending).resolves.toEqual({ ok: false }) expect(repo.countByAgent('a')).toBe(0) }) @@ -4033,6 +4088,15 @@ describe('MemoryPresenter embedding reindex (T5, AC-3.x)', () => { expect(repo.listByAgent('a')[0]?.embedding_model).toBe('p:m') }) + it('does not issue an empty fts_only requeue during backfill', async () => { + const { presenter, repo } = makePresenter(enabledConfig) + const requeueSpy = vi.spyOn(repo, 'requeueForEmbedding') + + await presenter.backfillEmbeddings('a') + + expect(requeueSpy).not.toHaveBeenCalled() + }) + it('re-drains rows a failed reindex left pending on the next backfill (P1-B)', async () => { const repo = new FakeRepository() const config: DeepChatAgentConfig = { @@ -5168,6 +5232,62 @@ describe('MemoryPresenter offline consolidation (T-B4..T-B6)', () => { expect(decisionCalls(generateText)).toBe(8) }) + it('records failed maintenance and throttles heavy passes when every LLM call fails', async () => { + const generateText = routedLLM({ throwDecision: true }) + const { presenter, repo, auditRepo } = makeLLMPresenter(generateText) + const now = 1_000 * DAY + const firstId = await seedEmbedded(presenter, 'user likes redis a') + const secondId = await seedEmbedded(presenter, 'user likes redis b') + repo.rows.get(firstId)!.created_at = now + repo.rows.get(secondId)!.created_at = now + 1 + + await presenter.runConsolidationPass('a', now) + const callsAfterFailure = decisionCalls(generateText) + expect(callsAfterFailure).toBeGreaterThan(0) + expect(auditRepo.getLatestCompletedEventAt('a', 'memory/maintenance_llm')).toBeNull() + expect(auditRepo.listByAgent('a')[0]).toMatchObject({ + event_type: 'memory/maintenance_llm', + status: 'failed', + reason: 'all-llm-steps-failed' + }) + expect(repo.getLastConsolidatedAt('a')).toBeNull() + + await presenter.runConsolidationPass('a', now + 5 * 60 * 1000) + expect(decisionCalls(generateText)).toBe(callsAfterFailure) + expect(auditRepo.getLatestCompletedEventAt('a', 'memory/maintenance_llm')).toBeNull() + + await presenter.runConsolidationPass('a', now + 31 * 60 * 1000) + expect(decisionCalls(generateText)).toBeGreaterThan(callsAfterFailure) + expect(auditRepo.getLatestCompletedEventAt('a', 'memory/maintenance_llm')).toBeNull() + }) + + it('does not keep completed cooldown when heavy maintenance aborts after memory is disabled', async () => { + const config: DeepChatAgentConfig = { ...embeddingConfig } + const generateText = vi.fn(async (_p: string, _m: string, prompt: string) => { + if (prompt.includes('Choose exactly ONE decision')) { + config.memoryEnabled = false + return '{"decision":"ADD","targetIndex":null,"mergedContent":null}' + } + return '' + }) + const { presenter, repo, auditRepo } = makeLLMPresenter(generateText, config) + const now = 1_000 * DAY + const firstId = await seedEmbedded(presenter, 'user likes redis a') + const secondId = await seedEmbedded(presenter, 'user likes redis b') + repo.rows.get(firstId)!.created_at = now + repo.rows.get(secondId)!.created_at = now + 1 + + await presenter.runConsolidationPass('a', now) + expect(decisionCalls(generateText)).toBe(1) + expect(auditRepo.getLatestCompletedEventAt('a', 'memory/maintenance_llm')).toBeNull() + + config.memoryEnabled = true + await presenter.runConsolidationPass('a', now + 60 * 1000) + + expect(decisionCalls(generateText)).toBeGreaterThan(1) + expect(auditRepo.getLatestCompletedEventAt('a', 'memory/maintenance_llm')).toBeNull() + }) + it('does not archive a just-merged old row in the same pass (T-B5)', async () => { const generateText = routedLLM({ decision: '{"decision":"SUPERSEDE","targetIndex":0,"mergedContent":"user prefers redis"}' @@ -6919,7 +7039,7 @@ describe('MemoryPresenter lifecycle revival (SDD-8)', () => { resolveTriage() const result = await extraction - expect(result).toEqual({ ok: true, createdIds: [] }) + expect(result).toEqual({ ok: false }) // Only the triage call ran; the extraction LLM is never fired after teardown begins. expect(generateText).toHaveBeenCalledTimes(1) expect(generateText.mock.calls[0][2]).toContain('KEEP or SKIP') diff --git a/test/main/presenter/memorySessionExtractionLock.test.ts b/test/main/presenter/memorySessionExtractionLock.test.ts index 3acc70548..6d528291d 100644 --- a/test/main/presenter/memorySessionExtractionLock.test.ts +++ b/test/main/presenter/memorySessionExtractionLock.test.ts @@ -118,4 +118,18 @@ describe('per-session extraction lock (C2, AC-2.3/2.4)', () => { expect(events).toEqual(['start1', 0, 'start2', 1]) }) + + it('advances the cursor only after ok:true extraction results', async () => { + let cursor = 0 + async function consumeSpan(toOrderSeq: number, result: { ok: boolean }) { + if (!result.ok) return + cursor = toOrderSeq + } + + await consumeSpan(10, { ok: false }) + expect(cursor).toBe(0) + + await consumeSpan(10, { ok: true }) + expect(cursor).toBe(10) + }) })