fix: close eager persistence races

packages/session-persistence/session-persistence/src/coordinator.ts

@@ -346,7 +346,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     ctx.effect(() => async () => {
       let disposeError: unknown
       try {
-        const errors = await settledErrors([...this.live.keys()].map(session => this.flushForDispose(session)))
+        const errors = await settledErrors([...this.live.keys()].map(session => this.flush(session)))
         while (this.chains.size > 0) await Promise.allSettled([...this.chains.values()])
         if (errors.length > 0) {
           throw new AggregateError(errors, `${this.backend.name} dispose failed`)
@@ -474,7 +474,12 @@ export class PersistenceCoordinator<TornMarker = unknown> {
         if (suffix.length > 0) await this.appendCore(id, suffix)
         return
       }
-      throw new Error(`session "${id}" is already bound to a different live session in this backend (id collision)`)
+      const owner = this.live.get(tracked.owner)
+      if (!tracked.materialized && !owner?.pending.length) {
+        this.states.delete(id)
+      } else {
+        throw new Error(`session "${id}" is already bound to a different live session in this backend (id collision)`)
+      }
     }
 
     // case 2/3: an artifact at THIS cwd is adopted as a live prefix (or rejected
@@ -530,18 +535,14 @@ export class PersistenceCoordinator<TornMarker = unknown> {
   private async flush(session: Session): Promise<void> {
     const live = this.initFor(session)
     await live.init
+    const overlapping = live.flush
+    if (overlapping !== undefined) await Promise.allSettled([overlapping])
     while (live.flush !== undefined || live.pending.length > 0) {
-      await this.ensureFlush(session, live)
+      if (live.flush !== undefined) await live.flush
+      else await this.ensureFlush(session, live)
     }
   }
 
-  /** Let an eager attempt settle, then make one teardown-owned retry observable. */
-  private async flushForDispose(session: Session): Promise<void> {
-    const current = this.live.get(session)?.flush
-    if (current !== undefined) await Promise.allSettled([current])
-    await this.flush(session)
-  }
-
   /** Start an eager drain without exposing its failure to the synchronous append. */
   private scheduleDrain(session: Session, live: LiveSessionState): void {
     void this.ensureFlush(session, live).catch((error: unknown) => {
@@ -549,9 +550,8 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     })
   }
 
-  /** Return the current drain, or start one for the complete pending batch. */
+  /** Start one drain for the complete pending batch. */
   private ensureFlush(session: Session, live: LiveSessionState): Promise<void> {
-    if (live.flush !== undefined) return live.flush
     const flush = live.init
       .then(() => this.serialize(session.header.id, () => this.drain(session.header.id, live)))
       .finally(() => { live.flush = undefined })

packages/session-persistence/session-persistence/tests/persistence.spec.ts

@@ -234,6 +234,41 @@ describe('PersistenceCoordinator eager writes', () => {
       await ctx.fiber.dispose()
     }
   })
+
+  it('retries a failed overlapping eager write at the explicit flush barrier', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const backend = new ControlledBackend()
+    const appendGate = Promise.withResolvers<boolean>()
+    backend.beforeAppend = async (attempt) => {
+      if (attempt === 1) {
+        await appendGate.promise
+        throw new Error('transient eager failure')
+      }
+    }
+    const fiber = await ctx.plugin(Object.assign((inner: Context) => {
+      new PersistenceCoordinator(inner, backend)
+    }, { inject: ['sessions'] }))
+
+    try {
+      const session = ctx.sessions.create(SessionId('eager-flush-retry'))
+      await ctx.sessions.flush(session)
+      session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+      session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+      await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
+
+      const barriers = [ctx.sessions.flush(session), ctx.sessions.flush(session)]
+      appendGate.resolve(true)
+
+      await expect(Promise.all(barriers)).resolves.toEqual([undefined, undefined])
+      expect(backend.appendAttempts).toBe(2)
+      expect(backend.store.get(session.id)?.events.map(event => event.seq)).toEqual([0, 1])
+    } finally {
+      appendGate.resolve(true)
+      await fiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
 })
 
 describe('PersistenceCoordinator retirement', () => {
@@ -241,29 +276,32 @@ describe('PersistenceCoordinator retirement', () => {
     const ctx = new Context()
     await ctx.plugin(SessionStore)
     const backend = new ControlledBackend()
-    let coordinator!: PersistenceCoordinator<never>
     const backendFiber = await ctx.plugin(Object.assign((inner: Context) => {
-      coordinator = new PersistenceCoordinator(inner, backend)
+      new PersistenceCoordinator(inner, backend)
     }, { inject: ['sessions'] }))
+    const loadGate = Promise.withResolvers<boolean>()
+    backend.beforeLoadStored = async (attempt) => {
+      if (attempt === 1) await loadGate.promise
+    }
 
     try {
       const id = SessionId('retiring-lazy-owner')
-      let first!: Session
       const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
-        first = inner.sessions.create(id)
+        inner.sessions.create(id)
       }, { inject: ['sessions'] }))
-      await ctx.sessions.flush(first)
+      await vi.waitFor(() => { expect(backend.loadAttempts).toBe(1) })
       await firstFiber.dispose()
-      const internals = coordinator as unknown as CoordinatorInternals
-      await vi.waitFor(() => { expect(internals.states.has(id)).toBe(false) })
 
       let reuse!: Session
       await ctx.plugin(Object.assign((inner: Context) => {
         reuse = inner.sessions.create(id)
       }, { inject: ['sessions'] }))
+      const reuseFlush = ctx.sessions.flush(reuse)
 
-      await expect(ctx.sessions.flush(reuse)).resolves.toBeUndefined()
+      loadGate.resolve(true)
+      await expect(reuseFlush).resolves.toBeUndefined()
     } finally {
+      loadGate.resolve(true)
       await backendFiber.dispose()
       await ctx.fiber.dispose()
     }