fix(memory): fix atomicity and recovery

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.

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.

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<string>()
           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
     })

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()