fix: serialize persistence ownership selection

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

@@ -255,10 +255,13 @@ export class PersistenceCoordinator<TornMarker = unknown> {
    * @param id - the persisted session to reload.
    * @returns the header plus the event log, ending on a balanced `turn/end`.
    */
-  load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
-    const live = this.ctx.sessions.get(id)
-    if (live !== undefined) return this.loadLiveSnapshot(live)
-    return this.serialize(id, () => this.loadCore(id))
+  async load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
+    const selected = await this.serialize(id, async () => {
+      const live = this.ctx.sessions.get(id)
+      if (live !== undefined) return { live }
+      return { loaded: await this.loadCore(id) }
+    })
+    return 'loaded' in selected ? selected.loaded : this.loadLiveSnapshot(selected.live)
   }
 
   private async loadCore(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {

packages/session-persistence/session-persistence/tests/coordinator-contract.ts

@@ -121,6 +121,39 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
       }
     })
 
+    it('rechecks live ownership after a cold load enters the per-id chain', async () => {
+      const fix = await makeFixture()
+      const { ctx, fiber } = await freshCtx(fix)
+      try {
+        const id = SessionId('queued-load-live-race')
+        const header = meta(id, WORK)
+        const start: SessionEvent = {
+          type: 'turn/start',
+          seq: 0,
+          time: 1,
+          data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } },
+        }
+        await ctx.sessionPersistence.create(header)
+        await ctx.sessionPersistence.append(id, [start])
+
+        const loading = ctx.sessionPersistence.load(id)
+        const live = ctx.sessions.create(id, { seed: [start], meta: header })
+        await expect(loading).rejects.toThrow(/live turn is open/)
+
+        live.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+        await ctx.sessions.flush(live)
+        const loaded = await ctx.sessionPersistence.load(id)
+        expect(loaded.events.map(event => event.type)).toEqual(['turn/start', 'turn/end'])
+        expect(loaded.events.at(-1)).toMatchObject({
+          type: 'turn/end',
+          data: { reason: { kind: 'completed' } },
+        })
+      } finally {
+        await fiber.dispose()
+        await fix.cleanup()
+      }
+    })
+
     it('does not load an unmaterialized empty live session', async () => {
       const fix = await makeFixture()
       const { ctx, fiber } = await freshCtx(fix)

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

@@ -354,6 +354,7 @@ describe('PersistenceCoordinator retirement', () => {
       coordinator = new PersistenceCoordinator(inner, backend)
     }, { inject: ['sessions'] }))
     const appendGate = Promise.withResolvers<boolean>()
+    const loadGate = Promise.withResolvers<boolean>()
 
     try {
       const id = SessionId('retiring-buffered-owner')
@@ -367,15 +368,20 @@ describe('PersistenceCoordinator retirement', () => {
       first.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
       await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
       await firstFiber.dispose()
+      const baselineLoads = backend.loadAttempts
+      backend.beforeLoadStored = async () => { await loadGate.promise }
       const coldLoad = coordinator.load(id)
 
+      appendGate.resolve(true)
+      await vi.waitFor(() => { expect(backend.loadAttempts).toBe(baselineLoads + 1) })
+
       let reuse!: Session
       await ctx.plugin(Object.assign((inner: Context) => {
         reuse = inner.sessions.create(id)
       }, { inject: ['sessions'] }))
       const reuseFlush = ctx.sessions.flush(reuse)
 
-      appendGate.resolve(true)
+      loadGate.resolve(true)
       await expect(coldLoad).resolves.toMatchObject({
         events: [{ seq: 0 }, { seq: 1 }],
       })
@@ -385,6 +391,7 @@ describe('PersistenceCoordinator retirement', () => {
       })
     } finally {
       appendGate.resolve(true)
+      loadGate.resolve(true)
       await backendFiber.dispose()
       await ctx.fiber.dispose()
     }