From 8a5fd875f9989f6b4b93dff423ab93d6cf4ef157 Mon Sep 17 00:00:00 2001 From: yyhhyyyyyy Date: Sat, 19 Sep 2026 00:12:46 +0800 Subject: [PATCH 1/3] fix(memory): make persona switches atomic --- src/main/memory/services/personaService.ts | 26 +++++---- test/main/memory/memoryUpdateNative.test.ts | 58 +++++++++++++++++++++ 2 files changed, 74 insertions(+), 10 deletions(-) diff --git a/src/main/memory/services/personaService.ts b/src/main/memory/services/personaService.ts index 69cca68c0..5dd7182ec 100644 --- a/src/main/memory/services/personaService.ts +++ b/src/main/memory/services/personaService.ts @@ -28,7 +28,8 @@ import type { MemoryLifecycleRepositoryPort, MemoryMutationRepositoryPort, MemoryReadRepositoryPort, - MemoryTextGenerationPort + MemoryTextGenerationPort, + MemoryTransactionPort } from '../ports' export class PersonaService { @@ -41,7 +42,8 @@ export class PersonaService { ctx: MemoryRuntimeContext repository: MemoryReadRepositoryPort & MemoryMutationRepositoryPort & - MemoryLifecycleRepositoryPort + MemoryLifecycleRepositoryPort & + MemoryTransactionPort textGeneration: MemoryTextGenerationPort } ) { @@ -203,10 +205,12 @@ export class PersonaService { return memoryCommandRejected('invalid-state') } const current = this.ports.repository.getActivePersona(agentId) - if (current && current.id !== draft.id) { - this.ports.repository.setPersonaState(current.id, 'superseded', draft.id) - } - this.ports.repository.setPersonaState(draft.id, 'active', null) + this.ports.repository.runInTransaction(() => { + if (current && current.id !== draft.id) { + this.ports.repository.setPersonaState(current.id, 'superseded', draft.id) + } + this.ports.repository.setPersonaState(draft.id, 'active', null) + }) this.ctx.markDomainMutationCommitted(agentId) this.ctx.emitChanged(agentId, 'persona-approve') return memoryCommandApplied() @@ -289,10 +293,12 @@ export class PersonaService { (target.persona_state == null && target.superseded_by != null) if (!isHistorical) return memoryCommandRejected('invalid-state') if (current && current.is_anchor === 1) return memoryCommandRejected('anchored') - if (current) { - this.ports.repository.setPersonaState(current.id, 'superseded', versionId) - } - this.ports.repository.setPersonaState(versionId, 'active', null) + this.ports.repository.runInTransaction(() => { + if (current) { + this.ports.repository.setPersonaState(current.id, 'superseded', versionId) + } + this.ports.repository.setPersonaState(versionId, 'active', null) + }) this.ctx.markDomainMutationCommitted(agentId) this.ctx.emitChanged(agentId, 'persona-rollback') return memoryCommandApplied() diff --git a/test/main/memory/memoryUpdateNative.test.ts b/test/main/memory/memoryUpdateNative.test.ts index dac9a2aff..02ab65195 100644 --- a/test/main/memory/memoryUpdateNative.test.ts +++ b/test/main/memory/memoryUpdateNative.test.ts @@ -46,6 +46,64 @@ function memoryAuditTable(database: InstanceType) { } describeIfNative('Memory update SQLite integration', () => { + it.each(['approve', 'rollback'] as const)( + 'rolls back both persona states when %s activation fails', + async (operation) => { + const directory = actualFs.mkdtempSync(join(tmpdir(), 'deepchat-persona-atomic-')) + const sqlite = new MainDatabaseCtor(join(directory, 'agent.db')) + const repository = memoryTable(sqlite) + const onMemoryChanged = vi.fn() + const presenter = new MemoryService({ + repository, + resolveAgentConfig: () => ({ memoryEnabled: true }), + executeWithRateLimit: async () => undefined, + getEmbeddings: async () => [], + createVectorStore: async () => new FakeVectorStore(), + resetVectorStore: async () => undefined, + onMemoryChanged + }) + try { + const first = presenter.evolvePersona('a', 'first self-model')! + await presenter.approvePersonaDraft('a', first) + const second = presenter.evolvePersona('a', 'second self-model')! + if (operation === 'rollback') await presenter.approvePersonaDraft('a', second) + const target = operation === 'approve' ? second : first + const active = operation === 'approve' ? first : second + const before = repository.listPersonaVersions('a') + onMemoryChanged.mockClear() + const setState = repository.setPersonaState.bind(repository) + const failure = vi + .spyOn(repository, 'setPersonaState') + .mockImplementation((id, ...args) => { + if (id === target && args[0] === 'active') throw new Error('activation failed') + setState(id, ...args) + }) + + await expect( + operation === 'approve' + ? presenter.approvePersonaDraft('a', target) + : presenter.rollbackPersona('a', target) + ).rejects.toThrow('activation failed') + + expect(repository.listPersonaVersions('a')).toEqual(before) + expect(repository.getActivePersona('a')?.id).toBe(active) + expect(onMemoryChanged).not.toHaveBeenCalled() + failure.mockRestore() + await expect( + operation === 'approve' + ? presenter.approvePersonaDraft('a', target) + : presenter.rollbackPersona('a', target) + ).resolves.toEqual({ action: 'applied' }) + expect(repository.getActivePersona('a')?.id).toBe(target) + expect(repository.getById(active)?.persona_state).toBe('superseded') + } finally { + await presenter.dispose() + sqlite.close() + actualFs.rmSync(directory, { recursive: true, force: true }) + } + } + ) + it('rejects partial canonical insert state in fake and SQLite repositories', () => { const directory = actualFs.mkdtempSync(join(tmpdir(), 'deepchat-memory-insert-state-')) const sqlite = new MainDatabaseCtor(join(directory, 'agent.db')) From 574b70b3b67cb812b2590e7d1bce563079ada534 Mon Sep 17 00:00:00 2001 From: yyhhyyyyyy Date: Sat, 19 Sep 2026 00:13:15 +0800 Subject: [PATCH 2/3] fix(memory): verify coverage within lease bounds --- docs/issues/memory-coverage-capacity/spec.md | 45 +++++++ src/main/memory/infra/embeddingPipeline.ts | 134 +++++++++---------- test/main/memory/embeddingPipeline.test.ts | 124 ++++++++++++++++- 3 files changed, 230 insertions(+), 73 deletions(-) create mode 100644 docs/issues/memory-coverage-capacity/spec.md diff --git a/docs/issues/memory-coverage-capacity/spec.md b/docs/issues/memory-coverage-capacity/spec.md new file mode 100644 index 000000000..ce372f85b --- /dev/null +++ b/docs/issues/memory-coverage-capacity/spec.md @@ -0,0 +1,45 @@ +# Memory coverage verification at capacity + +## Problem and cause + +Coverage verification uses the embedding drain's 200-batch guard for both SQLite and vector ID +enumeration. A full final page is indistinguishable from an unfinished listing. At 102,400 current +IDs, warming never certifies the store and requests a full reindex even when no vector is missing. +Rebuilding does not reduce the ID count, so subsequent warms repeat the same recovery. + +## Design + +Keep keyset pages of 512 IDs, but bound verification by the existing vector lease deadline rather +than the drain batch count. Enumerate both sources inside that lease. Yield between SQLite pages +so the main process can run cancellation and deadline callbacks; revalidate the operation fence, +read epoch and lease generation around asynchronous work. An unfinished or invalidated listing +must neither certify readiness nor trigger a reset merely because verification did not finish. + +Use one ID set for the remaining missing vectors, subtracting each sidecar page as it arrives; +retain only orphan IDs separately. Requeue missing rows and delete orphans in batches of at most +512 IDs, yielding and checking cancellation between batches. The lease bounds total work, while +page limits bound each synchronous database mutation. This is not a new vector index, scheduler or +configurable limit. Embedding drain limits, database schemas, Tape semantics and the authoritative +SQLite/projection boundary remain unchanged. + +## Acceptance and implementation + +- [x] Healthy coverage below, at and above the former page cap can become ready without reset. +- [x] Missing-vector requeue and orphan deletion remain bounded to one page per call. +- [x] Cancellation, clear or an expired lease stops verification and prevents late readiness. +- [x] Review P0-P3 risks, remove unnecessary design, run behavior/native/performance and static gates. + +## Validation outcome + +- Boundary tests enumerate 102,399, 102,400 and 102,401 IDs; the last case has equal source/store + counts but a missing ID and an orphan, guarding against count-only certification. +- On the base implementation the two larger boundary cases fail, as do the deadline and 513-ID + repair/cancellation cases. They pass with the fix. These large enumerations use port fixtures; + they are not a latency benchmark for a real 100k-row SQLite/DuckDB database. +- Review found a synchronous bulk-requeue risk after removing the page cap. Batching the repair + resolves it; follow-up assumption, concurrency, cascade and scale reviews found no remaining + P0-P3 findings. Ablation removed the duplicated complete ID arrays/sets and the truncation flag. +- Combined branch gates passed: format, i18n, lint, typecheck, 940 memory behavior tests, + 334 native tests (two VSS-dependent legacy tests skipped), ten performance tests and seven evals. + +No GitHub issue sync or push is authorized. Rollback is a code revert without a data migration. diff --git a/src/main/memory/infra/embeddingPipeline.ts b/src/main/memory/infra/embeddingPipeline.ts index 67305b34a..3ba9d101b 100644 --- a/src/main/memory/infra/embeddingPipeline.ts +++ b/src/main/memory/infra/embeddingPipeline.ts @@ -1043,7 +1043,13 @@ export class EmbeddingPipeline { } } - const coverage = await this.verifyVectorCoverage(agentId, embedding, dimensions, fingerprint) + const coverage = await this.verifyVectorCoverage( + agentId, + embedding, + dimensions, + fingerprint, + operationFence + ) if ( !this.ctx.canContinueOperation(operationFence) || !this.ctx.canUseCurrentMemoryEmbedding(agentId, embedding) @@ -1052,13 +1058,6 @@ export class EmbeddingPipeline { } if (!coverage.verified) { this.ports.vectorStore.clearReady(agentId) - if (coverage.authoritativeListingTruncated && !this.reindexing.has(agentId)) { - void this.reindexEmbeddings(agentId, true).catch((error) => { - logger.warn( - `[Memory] incomplete vector store rebuild failed for ${agentId}: ${String(error)}` - ) - }) - } return { outcome: 'deferred', error: new MemoryReindexFailure( @@ -1096,106 +1095,97 @@ export class EmbeddingPipeline { ) } - private collectCurrentEmbeddedIds( - agentId: string, - dimensions: number, - fingerprint: string - ): { ids: string[]; complete: boolean } { - const ids: string[] = [] - let afterId: string | null = null - for (let guard = 0; guard < REINDEX_MAX_BATCHES; guard += 1) { - const page = this.ports.repository.listCurrentEmbeddedIds( - agentId, - dimensions, - fingerprint, - afterId, - ORPHAN_RECONCILE_BATCH - ) - ids.push(...page) - if (page.length < ORPHAN_RECONCILE_BATCH) return { ids, complete: true } - afterId = page[page.length - 1] - } - return { ids, complete: false } - } - private async verifyVectorCoverage( agentId: string, embedding: MemoryModelRef, dimensions: number, - fingerprint: string + fingerprint: string, + operationFence: MemoryOperationFence ): Promise<{ verified: boolean - authoritativeListingTruncated: boolean generation: number }> { return this.ports.vectorStore.withVectorMutation(agentId, async () => { const readEpoch = this.ctx.captureReadEpoch(agentId) - const authoritative = this.collectCurrentEmbeddedIds(agentId, dimensions, fingerprint) - if (!authoritative.complete) { - return { verified: false, authoritativeListingTruncated: true, generation: -1 } - } const outcome = await this.ports.vectorStore.withStoreLease( agentId, embedding, dimensions, async (store, generation) => { - if (!store.isUsable()) { - return { - verified: false, - authoritativeListingTruncated: false, - generation - } - } - const sidecarIds: string[] = [] + const isCurrent = () => + this.ctx.canContinueOperation(operationFence) && + this.ctx.isReadEpochCurrent(agentId, readEpoch) && + this.ctx.canUseCurrentMemoryEmbedding(agentId, embedding) && + this.ports.vectorStore.isGenerationCurrent(agentId, generation) + if (!store.isUsable() || !isCurrent()) return { verified: false, generation } + + // Coverage is bounded by the lease deadline, not the embedding drain's batch budget. + // Yield even on synchronous SQLite pages so the deadline and cancellation can run. + const missingIds = new Set() let afterId: string | null = null - let complete = false - for (let guard = 0; guard < REINDEX_MAX_BATCHES; guard += 1) { + while (isCurrent()) { + const page = this.ports.repository.listCurrentEmbeddedIds( + agentId, + dimensions, + fingerprint, + afterId, + ORPHAN_RECONCILE_BATCH + ) + for (const id of page) missingIds.add(id) + if (page.length < ORPHAN_RECONCILE_BATCH) break + afterId = page[page.length - 1] + await this.waitForBackgroundTick() + } + if (!isCurrent()) return { verified: false, generation } + + const extras: string[] = [] + afterId = null + while (isCurrent()) { const page = await store.listMemoryIds(afterId, ORPHAN_RECONCILE_BATCH) - if (!this.ports.vectorStore.isGenerationCurrent(agentId, generation)) { - return { verified: false, authoritativeListingTruncated: false, generation } - } - sidecarIds.push(...page) - if (page.length < ORPHAN_RECONCILE_BATCH) { - complete = true - break + if (!isCurrent()) return { verified: false, generation } + for (const id of page) { + if (!missingIds.delete(id)) extras.push(id) } + if (page.length < ORPHAN_RECONCILE_BATCH) break afterId = page[page.length - 1] + await this.waitForBackgroundTick() } - if (!complete) { - return { verified: false, authoritativeListingTruncated: false, generation } - } - const authoritativeSet = new Set(authoritative.ids) - const sidecarSet = new Set(sidecarIds) + if (!isCurrent()) return { verified: false, generation } // A ready row without a vector only needs its own embedding again. Requeueing exactly // those rows removes them from the ready set, so the certificate below stays truthful // and the ordinary backfill drain repairs them without resetting the store. - const missingIds = authoritative.ids.filter((id) => !sidecarSet.has(id)) - if (missingIds.length > 0) { - const requeued = this.ports.repository.requeueReadyEmbeddingsByIds(agentId, missingIds) + if (missingIds.size > 0) { + let requeued = 0 + let remaining = missingIds.size + let batch: string[] = [] + for (const id of missingIds) { + batch.push(id) + remaining -= 1 + if (batch.length < ORPHAN_RECONCILE_BATCH && remaining > 0) continue + requeued += this.ports.repository.requeueReadyEmbeddingsByIds(agentId, batch) + batch = [] + if (remaining > 0) await this.waitForBackgroundTick() + if (!isCurrent()) return { verified: false, generation } + } logger.warn( - `[Memory] requeued ${requeued} of ${missingIds.length} ready rows whose vectors were missing for ${agentId}` + `[Memory] requeued ${requeued} of ${missingIds.size} ready rows whose vectors were missing for ${agentId}` ) } - const extras = sidecarIds.filter((id) => !authoritativeSet.has(id)) for (let start = 0; start < extras.length; start += ORPHAN_RECONCILE_BATCH) { await store.deleteByMemoryIds(extras.slice(start, start + ORPHAN_RECONCILE_BATCH)) - if (!this.ports.vectorStore.isGenerationCurrent(agentId, generation)) { - return { verified: false, authoritativeListingTruncated: false, generation } - } + if (start + ORPHAN_RECONCILE_BATCH < extras.length) await this.waitForBackgroundTick() + if (!isCurrent()) return { verified: false, generation } } - return { verified: true, authoritativeListingTruncated: false, generation } + return { verified: true, generation } } ) if ( + !this.ctx.canContinueOperation(operationFence) || !this.ctx.isReadEpochCurrent(agentId, readEpoch) || !this.ctx.canUseCurrentMemoryEmbedding(agentId, embedding) || !this.ports.vectorStore.isGenerationCurrent(agentId, outcome.generation) ) { - return { - verified: false, - authoritativeListingTruncated: false, - generation: outcome.generation - } + return { verified: false, generation: outcome.generation } } return outcome }) diff --git a/test/main/memory/embeddingPipeline.test.ts b/test/main/memory/embeddingPipeline.test.ts index 1f08887ac..7dc98bd41 100644 --- a/test/main/memory/embeddingPipeline.test.ts +++ b/test/main/memory/embeddingPipeline.test.ts @@ -1,6 +1,9 @@ import { describe, expect, it, vi } from 'vitest' -import { ERROR_RETRY_COOLDOWN_MS } from '@/memory/runtimeConstants' +import { + ERROR_RETRY_COOLDOWN_MS, + VECTOR_STORE_OPERATION_TIMEOUT_MS +} from '@/memory/runtimeConstants' import { type IMemoryVectorStore } from '@/memory/types' import logger from '@shared/logger' import type { DeepChatAgentConfig } from '@shared/types/agent-interface' @@ -498,6 +501,125 @@ describe('MemoryService.processPendingEmbeddings (batch + fairness)', () => { }) describe('MemoryService embedding reindex (T5, AC-3.x)', () => { + it.each([102_399, 102_400, 102_401])( + 'verifies %i current IDs without a capacity-triggered rebuild', + async (count) => { + const { presenter, repo, store, resetVectorStore } = makePresenter(enabledConfig) + const ids = Array.from( + { length: count }, + (_, index) => `id-${String(index).padStart(6, '0')}` + ) + // Above the former cap, place both a missing vector and an orphan in the last page. + // Equal row counts alone must not establish coverage. + const needsRepair = count === 102_401 + const vectorIds = needsRepair ? [...ids.slice(0, -1), 'z-orphan'] : ids + vi.spyOn(repo, 'getCurrentEmbeddingDimension').mockReturnValue(4) + const listRows = vi + .spyOn(repo, 'listCurrentEmbeddedIds') + .mockImplementation((_agent, _dimensions, _fingerprint, afterId, limit) => { + const offset = afterId === null ? 0 : ids.indexOf(afterId) + 1 + return ids.slice(offset, offset + limit) + }) + const listVectors = vi + .spyOn(store, 'listMemoryIds') + .mockImplementation(async (afterId, limit) => { + const offset = afterId === null ? 0 : vectorIds.indexOf(afterId) + 1 + return vectorIds.slice(offset, offset + limit) + }) + const requeue = vi + .spyOn(repo, 'requeueReadyEmbeddingsByIds') + .mockReturnValue(needsRepair ? 1 : 0) + const remove = vi.spyOn(store, 'deleteByMemoryIds') + const runtime = memoryRuntimeForTests(presenter) + const reindex = vi.spyOn(runtime.embeddingService, 'reindexEmbeddings') + try { + await runtime.embeddingService.warmVectorStore('a', { providerId: 'p', modelId: 'm' }) + expect(runtime.isVectorReady('a')).toBe(true) + expect(resetVectorStore).not.toHaveBeenCalled() + expect(reindex).not.toHaveBeenCalled() + expect(listRows.mock.calls.every((args) => args[4] === 512)).toBe(true) + expect(listVectors.mock.calls.every((args) => args[1] === 512)).toBe(true) + if (needsRepair) { + expect(requeue).toHaveBeenCalledExactlyOnceWith('a', [ids[count - 1]]) + expect(remove).toHaveBeenCalledExactlyOnceWith(['z-orphan']) + } else { + expect(requeue).not.toHaveBeenCalled() + expect(remove).not.toHaveBeenCalled() + } + } finally { + await presenter.dispose() + } + } + ) + + it.each([false, true])( + 'bounds missing-vector repairs and observes cancellation=%s', + async (cancel) => { + const config = { ...enabledConfig } + const { presenter, repo, resetVectorStore } = makePresenter(config) + const ids = Array.from({ length: 513 }, (_, index) => `id-${String(index).padStart(4, '0')}`) + vi.spyOn(repo, 'getCurrentEmbeddingDimension').mockReturnValue(4) + vi.spyOn(repo, 'listCurrentEmbeddedIds').mockImplementation( + (_agent, _dimensions, _fingerprint, afterId, limit) => { + const offset = afterId === null ? 0 : ids.indexOf(afterId) + 1 + return ids.slice(offset, offset + limit) + } + ) + const requeue = vi + .spyOn(repo, 'requeueReadyEmbeddingsByIds') + .mockImplementation((_agent, batch) => { + if (cancel) config.memoryEnabled = false + return batch.length + }) + const runtime = memoryRuntimeForTests(presenter) + try { + await runtime.embeddingService.warmVectorStore('a', { providerId: 'p', modelId: 'm' }) + expect(requeue.mock.calls.map(([, batch]) => batch.length)).toEqual( + cancel ? [512] : [512, 1] + ) + expect(requeue.mock.calls.flatMap(([, batch]) => batch)).toEqual( + cancel ? ids.slice(0, 512) : ids + ) + expect(runtime.isVectorReady('a')).toBe(!cancel) + expect(resetVectorStore).not.toHaveBeenCalled() + } finally { + await presenter.dispose() + } + } + ) + + it('stops paginated coverage at the lease deadline without resetting the store', async () => { + vi.useFakeTimers() + const { presenter, repo, store, resetVectorStore } = makePresenter(enabledConfig) + const runtime = memoryRuntimeForTests(presenter) + vi.spyOn(repo, 'getCurrentEmbeddingDimension').mockReturnValue(4) + let page = 0 + const listRows = vi.spyOn(repo, 'listCurrentEmbeddedIds').mockImplementation(() => { + // Simulate a slow but progressing SQLite listing; each page uses part of the real lease budget. + vi.advanceTimersByTime(10_000) + return Array.from({ length: 512 }, (_, index) => `id-${page++}-${index}`) + }) + const listVectors = vi.spyOn(store, 'listMemoryIds') + const requeue = vi.spyOn(repo, 'requeueReadyEmbeddingsByIds') + const remove = vi.spyOn(store, 'deleteByMemoryIds') + try { + const warm = runtime.embeddingService.warmVectorStore('a', { providerId: 'p', modelId: 'm' }) + await vi.advanceTimersByTimeAsync(VECTOR_STORE_OPERATION_TIMEOUT_MS + 10) + await warm + expect(listRows).toHaveBeenCalledTimes(3) + expect(listVectors).not.toHaveBeenCalled() + expect(requeue).not.toHaveBeenCalled() + expect(remove).not.toHaveBeenCalled() + expect(runtime.isVectorReady('a')).toBe(false) + expect(resetVectorStore).not.toHaveBeenCalled() + await vi.advanceTimersByTimeAsync(100) + expect(listRows).toHaveBeenCalledTimes(3) + } finally { + await presenter.dispose() + vi.useRealTimers() + } + }) + it('serializes coverage verification with embedding persistence for the same agent', async () => { const { presenter, repo, store } = makePresenter(enabledConfig) repo.insert({ From 703ce95af987a37b93ebf13cc0efa26bc6f0af86 Mon Sep 17 00:00:00 2001 From: yyhhyyyyyy Date: Sat, 19 Sep 2026 00:13:46 +0800 Subject: [PATCH 3/3] fix(memory): track partial writes before awaits --- .../memory-partial-write-cancellation/spec.md | 44 ++++++++ src/main/memory/services/writeCoordinator.ts | 30 +++-- test/main/memory/writeCoordinator.test.ts | 103 +++++++++++++++++- 3 files changed, 169 insertions(+), 8 deletions(-) create mode 100644 docs/issues/memory-partial-write-cancellation/spec.md diff --git a/docs/issues/memory-partial-write-cancellation/spec.md b/docs/issues/memory-partial-write-cancellation/spec.md new file mode 100644 index 000000000..3e0e030d0 --- /dev/null +++ b/docs/issues/memory-partial-write-cancellation/spec.md @@ -0,0 +1,44 @@ +# Partial memory writes and cancellation + +## Cause and reproduction + +Extraction can restore an existing archived claim during candidate preparation, then await the +decision provider for another candidate. Mutation epochs and working projection invalidation were +deferred until the whole batch returned. Disabling memory while the provider was pending skipped +that bookkeeping; enabling memory again reused the old working projection despite the restored row. + +A deterministic deferred-provider test is needed before the fix because the failure depends on +partial commit, cancellation and re-enable ordering. It reproduces the stale projection; clear and +dispose variants characterize the existing no-late-write contract. + +## Design and boundaries + +Record mutation epochs and dirty working state synchronously when a batch candidate commits, before +another provider await. The shared kernel owns this for extraction and model-assisted remember; +direct remembers keep their synchronous bookkeeping. Do not duplicate it after batch completion. + +Cancellation still returns failure for extraction, so the ingestion cursor does not advance. Keep +the fence around final events, embedding and consolidation: the cancelled batch must not dispatch +late work, recreate cleared claims, or revive disposed runtime state. Existing recovery drains own +pending embeddings. This does not alter Tape/Journal dispatch, database schema or shared events. + +## Acceptance and implementation + +- [x] After partial commit then disable/re-enable, injection reflects the committed claim. +- [x] Clear/dispose while a decision is pending prevents late writes, events and provider work. +- [x] Successful and failed batches retain their outcome, audit and retry contracts. +- [x] Review P0-P3 risks, ablate unnecessary design, and run the memory and static gates. + +## Validation outcome + +The disable/re-enable regression fails against the base implementation and passes with the fix. +The clear/dispose variants pass on both versions, preserving the existing cancellation boundary. +Immediate preparation, initial decision application and retry application all record committed +outcomes through one helper. The remaining direct map assignments record no-op outcomes only. +Post-batch duplicate bookkeeping is removed; no new callback, scheduler or public contract is added. + +Combined branch verification passed: format, i18n, lint, typecheck, 940 memory behavior tests, +334 native tests (two VSS-dependent legacy tests skipped), ten performance tests and seven evals. +Final P0-P3 reviews found no remaining findings in the change scope. + +Rollback is code-only. No push or GitHub issue sync is authorized. diff --git a/src/main/memory/services/writeCoordinator.ts b/src/main/memory/services/writeCoordinator.ts index 5ab118629..ebc255532 100644 --- a/src/main/memory/services/writeCoordinator.ts +++ b/src/main/memory/services/writeCoordinator.ts @@ -420,8 +420,6 @@ export class WriteCoordinator { outcomes.push(outcome) createdIds.push(...createdIdsFromOutcome(outcome)) if (outcomeTouched(outcome)) { - this.ctx.markDomainMutationCommitted(input.agentId) - this.ports.markWorkingMemoryDirty(input.agentId) touched = true } } @@ -785,7 +783,8 @@ export class WriteCoordinator { continue } const { settled } = preparation - run.outcomesByIndex.set( + this.recordBatchOutcome( + run, indexed.candidateIndex, settled.state === 'forgotten' ? { action: 'noop', reason: 'forgotten' } @@ -804,6 +803,20 @@ export class WriteCoordinator { return prepared } + private recordBatchOutcome( + run: BatchRun, + candidateIndex: number, + outcome: MemoryWriteOutcome + ): void { + run.outcomesByIndex.set(candidateIndex, outcome) + // Commit bookkeeping cannot wait for the batch: another provider await may be cancelled + // after this claim is durable. Events and follow-up jobs still obey the caller's fence. + if (outcomeTouched(outcome)) { + this.ctx.markDomainMutationCommitted(run.ctx.agentId) + this.ports.markWorkingMemoryDirty(run.ctx.agentId) + } + } + // The single decision kernel for extraction batches and one-off remembers alike: settle // provenance-decided candidates synchronously, recall neighbors and ask the decision model once // for the rest, apply, then give CAS losers one bounded retry that reuses their query vectors. @@ -849,7 +862,7 @@ export class WriteCoordinator { false ) if (applied.action === 'retry') retryCandidates.push(item) - else outcomesByIndex.set(item.candidateIndex, applied) + else this.recordBatchOutcome(run, item.candidateIndex, applied) } catch (error) { failBatch(run, error, 'apply') break @@ -916,7 +929,8 @@ export class WriteCoordinator { retryBatch.decisions.get(item.candidateIndex), true ) - outcomesByIndex.set( + this.recordBatchOutcome( + run, item.candidateIndex, applied.action === 'retry' ? { action: 'noop', reason: 'concurrent-update' } : applied ) @@ -1372,8 +1386,10 @@ export class WriteCoordinator { return { action: 'noop', reason: 'disposed' } } if (outcomeTouched(outcome)) { - this.ctx.markDomainMutationCommitted(ctx.agentId) - this.ports.markWorkingMemoryDirty(ctx.agentId) + if (!resolvedModel) { + this.ctx.markDomainMutationCommitted(ctx.agentId) + this.ports.markWorkingMemoryDirty(ctx.agentId) + } this.ctx.emitChanged(ctx.agentId, 'extract') if (outcome.action !== 'challenged') { void this.ports.triggerEmbedding(ctx.agentId).catch((error) => { diff --git a/test/main/memory/writeCoordinator.test.ts b/test/main/memory/writeCoordinator.test.ts index 2f9fa4b74..2fe79e8ab 100644 --- a/test/main/memory/writeCoordinator.test.ts +++ b/test/main/memory/writeCoordinator.test.ts @@ -14,7 +14,13 @@ import { makePresenter, textToVector } from './support/memoryFakes' -import { decisionCalls, makeLLMPresenter, routedLLM, seedEmbedded } from './serviceTestSupport' +import { + decisionCalls, + deferred, + makeLLMPresenter, + routedLLM, + seedEmbedded +} from './serviceTestSupport' import { MemoryService, embeddingDimensions, waitForMemoryCondition } from './serviceTestSupport' @@ -738,6 +744,101 @@ describe('MemoryService change events (onMemoryChanged)', () => { }) describe('extraction batch recovery', () => { + it.each(['disable', 'clear', 'dispose'] as const)( + 'keeps partial commits coherent without resuming a batch after %s', + async (cancellation) => { + vi.useFakeTimers() + const repo = createFakeRepository() + let config: DeepChatAgentConfig = { memoryEnabled: true } + const decisionStarted = deferred() + const decision = deferred() + const onMemoryChanged = vi.fn() + const getEmbeddings = vi.fn(async () => []) + const presenter = new MemoryService({ + repository: repo, + resolveAgentConfig: () => config, + getEmbeddings, + generateText: async (_p, _m, prompt) => { + if (prompt.includes('KEEP or SKIP')) return 'KEEP' + if (prompt.includes('JSON array')) { + return JSON.stringify([ + { kind: 'semantic', content: 'restored redis preference', importance: 0.9 }, + { kind: 'semantic', content: 'new postgres preference', importance: 0.8 } + ]) + } + decisionStarted.resolve() + return decision.promise + }, + createVectorStore: async () => new FakeVectorStore(), + resetVectorStore: async () => undefined, + onMemoryChanged + }) + try { + repo.insert({ + id: 'restored', + agentId: 'a', + kind: 'semantic', + content: 'restored redis preference', + importance: 0.9, + status: 'archived', + provenanceKey: buildMemoryProvenanceKey('a', 'semantic', 'restored redis preference') + }) + repo.insert({ + id: 'neighbor', + agentId: 'a', + kind: 'semantic', + content: 'old postgres preference', + importance: 0.8, + status: 'fts_only' + }) + presenter.captureExecutionToken('a') + presenter.refreshWorkingMemory('a') + const before = await presenter.buildInjection('a', '') + expect(before?.payload.working).toContain('old postgres preference') + expect(before?.payload.working).not.toContain('restored redis preference') + const pending = presenter.extractAndStore({ + agentId: 'a', + spanText: 'User: update my preferences', + model: { providerId: 'p', modelId: 'm' } + }) + await decisionStarted.promise + expect(repo.getById('restored')?.lifecycle_state).toBe('active') + + if (cancellation === 'disable') { + config = { memoryEnabled: false } + presenter.onAgentMemoryMaintenanceConfigChanged('a') + } else if (cancellation === 'clear') { + await presenter.clearMemories('a') + } else { + await presenter.dispose() + } + onMemoryChanged.mockClear() + decision.resolve( + JSON.stringify([{ candidateIndex: 1, decision: 'ADD', targetIndex: null }]) + ) + await expect(pending).resolves.toEqual({ ok: false }) + expect(repo.listByAgent('a').some((row) => row.content === 'new postgres preference')).toBe( + false + ) + expect(onMemoryChanged).not.toHaveBeenCalled() + expect(getEmbeddings).not.toHaveBeenCalled() + if (cancellation === 'disable') { + config = { memoryEnabled: true } + presenter.onAgentMemoryMaintenanceConfigChanged('a') + expect((await presenter.buildInjection('a', ''))?.payload.working).toContain( + 'restored redis preference' + ) + } else if (cancellation === 'clear') { + expect(repo.countByAgent('a')).toBe(0) + } + } finally { + decision.resolve('[]') + await presenter.dispose() + vi.useRealTimers() + } + } + ) + it('finalizes committed candidates when a retried candidate fails while settling', async () => { const repo = createFakeRepository() const auditRepo = new FakeAuditRepository()