fix: keep crash repair away from live sessions

2026-07-23 18:31:37 +08:00
parent 022b42a450
commit 7f5ba286fe
12 changed files with 161 additions and 30 deletions
--- a/packages/session-persistence/session-persistence/README.md
+++ b/packages/session-persistence/session-persistence/README.md
@@ -11,7 +11,7 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l
 | `locate(meta): SessionLocation \| undefined` | Resolve an absolute per-session artifact target without I/O or materialization. Backends without an independent local artifact return `undefined`. |
 | `create(meta): Promise<void>` | Register a new session's metadata. MAY defer the physical write until the first `append` (lazy materialization). |
 | `append(id, events): Promise<void>` | Durably persist a batch. Append-only; first event `seq` == stored next-seq after any repair; rejects non-JSON-serializable data naming the offending type. |
-| `load(id): Promise<{ meta; events }>` | Reload meta + log. Preserves an interrupted (unclosed) final turn and closes it with synthetic closers — an error `tool/result` per unanswered `tool-call`, then `step/end?`+`turn/end {interrupted}` (a turn can be huge — never truncated); only a torn tail fragment is dropped. Events contiguous (`events[i].seq === i`); rejects a committed-region gap/parse error or unknown `version`. |
+| `load(id): Promise<{ meta; events }>` | Return a flushed balanced snapshot for a live session, rejecting while its turn is open; cold load preserves an interrupted final turn and closes it with synthetic `tool/result`/`step/end?`/`turn/end {interrupted}` events. Only a torn tail fragment is dropped; committed corruption and unknown `version` reject. |
 | `list(): Promise<SessionHeader[]>` | Lightweight listing from metadata, no full-log parse. A zero-event lazily-materialized session is absent from `list`. |

 ## Invariants every backend must honor
@@ -27,6 +27,8 @@ The persisted unit IS the existing `SessionEvent` (event-sourced model — the l

 Each `session/event` copies its event into the session controller and starts an eager drain without blocking the producer. Concurrent notifications share the current drain; events admitted during a write remain pending and trigger the next batch. `session/flush` is an observation barrier that waits until the controller has no current or pending batch. An eager failure is logged and retains the batch; the next explicit flush or backend teardown retries it and surfaces failure to its caller.

+Crash repair is cold-only. For a live id, `load(id)` snapshots the authoritative in-memory log, waits for that snapshot to become durable, and returns it only when balanced; an open live turn rejects instead of receiving synthetic interruption closers. A cold id follows storage repair normally. HMR adoption likewise uses the separate `loadLive` hook and never closes the active turn.
+
 When a live session emits `session/disposed`, the coordinator waits for its controller, serializes a final drain, then releases state owned by that exact `Session` object. Failed retirement leaves the controller in the live-session map, so backend teardown can retry it. Backend teardown stops event admission first, flushes every remaining controller, awaits per-id operations, and only then closes the storage handle.

 The side-effect-free `locate` query remains backend-owned because it describes storage topology rather than write orchestration.
--- a/packages/session-persistence/session-persistence/src/coordinator.ts
+++ b/packages/session-persistence/session-persistence/src/coordinator.ts
@@ -256,6 +256,8 @@ export class PersistenceCoordinator<TornMarker = unknown> {
   * @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))
  }

@@ -279,6 +281,18 @@ export class PersistenceCoordinator<TornMarker = unknown> {
    return { meta, events: balanced }
  }

+  /** Return a durable balanced live snapshot without applying cold crash repair. */
+  private async loadLiveSnapshot(session: Session): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
+    const meta = structuredClone(session.header)
+    const events = session.events.map(event => structuredClone(event))
+    await this.flush(session)
+    if (events.length === 0) throw new Error(`session "${session.id}" not found`)
+    if (interruptedTurnClosers(events).length > 0) {
+      throw new Error(`cannot load session "${session.id}" while its live turn is open; use the live Session or wait for the turn to close`)
+    }
+    return { meta, events }
+  }
+
  // Listing is a direct backend read and needs no coordinator state.

  // --- per-id serialization + adoption helpers ---
--- a/packages/session-persistence/session-persistence/src/index.ts
+++ b/packages/session-persistence/session-persistence/src/index.ts
@@ -74,9 +74,11 @@ export abstract class SessionPersistence extends Service {

  /**
   * Load a header and balanced contiguous log. A complete interrupted final
- * turn is preserved and durably closed with missing tool errors plus any open
- * step and turn boundaries; only a torn final record is discarded. Unknown
- * versions and corruption in the committed prefix reject.
+   * turn is preserved and durably closed with missing tool errors plus any open
+   * step and turn boundaries; only a torn final record is discarded. Unknown
+   * versions and corruption in the committed prefix reject. Implementations
+   * MUST NOT crash-repair an identity still bound to a live Session: a balanced
+   * live log may return as a durable snapshot, while an open live turn rejects.
   * @param id - the persisted session to reload.
   * @returns the header and a log ending on a balanced `turn/end`.
   */
--- a/packages/session-persistence/session-persistence/tests/coordinator-contract.ts
+++ b/packages/session-persistence/session-persistence/tests/coordinator-contract.ts
@@ -88,6 +88,51 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
      }
    })

+    it('rejects crash-repair load while a live session owns the persisted prefix', async () => {
+      const fix = await makeFixture()
+      const { ctx, fiber } = await freshCtx(fix)
+      let session!: Session
+      const sessionFiber = await ctx.plugin(Object.assign((inner: Context) => {
+        session = inner.sessions.create(SessionId('live-load'), { meta: { cwd: WORK } })
+      }, { inject: ['sessions'] }))
+      try {
+        session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+        await ctx.sessions.flush(session)
+
+        await expect(ctx.sessionPersistence.load(session.id))
+          .rejects.toThrow(`cannot load session "${session.id}" while its live turn is open`)
+
+        send(session, oneTurnLog().slice(1))
+        await ctx.sessions.flush(session)
+        await sessionFiber.dispose()
+
+        await vi.waitFor(async () => {
+          const loaded = await ctx.sessionPersistence.load(session.id)
+          expect(loaded.events.map(event => event.type)).toEqual(oneTurnLog().map(event => event.type))
+          expect(loaded.events.at(-1)).toMatchObject({
+            type: 'turn/end',
+            data: { reason: { kind: 'completed' } },
+          })
+        })
+      } finally {
+        await sessionFiber.dispose()
+        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)
+      try {
+        const session = ctx.sessions.create(SessionId('empty-live'), { meta: { cwd: WORK } })
+        await expect(ctx.sessionPersistence.load(session.id)).rejects.toThrow(/not found/)
+      } finally {
+        await fiber.dispose()
+        await fix.cleanup()
+      }
+    })
+
    it('round-trips the seed boundary (seedLength) through persistence', async () => {
      // A forked child records how many leading events were inherited via the seed; the
      // boundary must survive a reload (so a resume/replay can tell the inherited prefix from
--- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts
+++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts
@@ -245,7 +245,6 @@ describe('PersistenceCoordinator retirement', () => {
    const backendFiber = await ctx.plugin(Object.assign((inner: Context) => {
      coordinator = new PersistenceCoordinator(inner, backend)
    }, { inject: ['sessions'] }))
-    const loadGate = Promise.withResolvers<boolean>()

    try {
      const id = SessionId('retiring-lazy-owner')
@@ -254,52 +253,42 @@ describe('PersistenceCoordinator retirement', () => {
        first = inner.sessions.create(id)
      }, { inject: ['sessions'] }))
      await ctx.sessions.flush(first)
-
-      const baselineLoads = backend.loadAttempts
-      backend.beforeLoadStored = async () => { await loadGate.promise }
-      const blockingLoad = coordinator.load(id)
-      await vi.waitFor(() => { expect(backend.loadAttempts).toBe(baselineLoads + 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'] }))

-      loadGate.resolve(true)
-      await expect(blockingLoad).rejects.toThrow(/not found/)
      await expect(ctx.sessions.flush(reuse)).resolves.toBeUndefined()
    } finally {
-      loadGate.resolve(true)
      await backendFiber.dispose()
      await ctx.fiber.dispose()
    }
  })

-  it('a retiring owner with buffered events still rejects same-id reuse', async () => {
+  it('a replacement queued before retirement cleanup still collides with the live owner', async () => {
    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>()
+    const appendGate = Promise.withResolvers<boolean>()

    try {
-      const id = SessionId('retiring-buffered-owner')
+      const id = SessionId('retiring-live-owner')
      let first!: Session
      const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
        first = inner.sessions.create(id)
      }, { inject: ['sessions'] }))
      await ctx.sessions.flush(first)
+      backend.beforeAppend = async () => { await appendGate.promise }
      first.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
      first.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
-
-      const baselineLoads = backend.loadAttempts
-      backend.beforeLoadStored = async () => { await loadGate.promise }
-      const blockingLoad = coordinator.load(id)
-      await vi.waitFor(() => { expect(backend.loadAttempts).toBe(baselineLoads + 1) })
+      await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
      await firstFiber.dispose()

      let reuse!: Session
@@ -308,14 +297,56 @@ describe('PersistenceCoordinator retirement', () => {
      }, { inject: ['sessions'] }))
      const reuseFlush = ctx.sessions.flush(reuse)

-      loadGate.resolve(true)
-      await expect(blockingLoad).rejects.toThrow(/not found/)
+      appendGate.resolve(true)
+      await expect(reuseFlush).rejects.toThrow(/bound to a different live session/)
+      expect(backend.store.get(id)?.events.map(event => event.seq)).toEqual([0, 1])
+    } finally {
+      appendGate.resolve(true)
+      await backendFiber.dispose()
+      await ctx.fiber.dispose()
+    }
+  })
+
+  it('a racing cold load survives retirement cleanup and rejects same-id reuse', async () => {
+    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)
+    }, { inject: ['sessions'] }))
+    const appendGate = Promise.withResolvers<boolean>()
+
+    try {
+      const id = SessionId('retiring-buffered-owner')
+      let first!: Session
+      const firstFiber = await ctx.plugin(Object.assign((inner: Context) => {
+        first = inner.sessions.create(id)
+      }, { inject: ['sessions'] }))
+      await ctx.sessions.flush(first)
+      backend.beforeAppend = async () => { await appendGate.promise }
+      first.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
+      first.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+      await vi.waitFor(() => { expect(backend.appendAttempts).toBe(1) })
+      await firstFiber.dispose()
+      const coldLoad = coordinator.load(id)
+
+      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)
+      await expect(coldLoad).resolves.toMatchObject({
+        events: [{ seq: 0 }, { seq: 1 }],
+      })
      await expect(reuseFlush).rejects.toThrow(/id collision/)
      await vi.waitFor(() => {
        expect(backend.store.get(id)?.events.map(event => event.seq)).toEqual([0, 1])
      })
    } finally {
-      loadGate.resolve(true)
+      appendGate.resolve(true)
      await backendFiber.dispose()
      await ctx.fiber.dispose()
    }