feat(session-query): checkpoint build round 1

packages/session-persistence/session-persistence/README.md

@@ -26,6 +26,8 @@ The two first-party backends were byte-identical (or same-algorithm) for ALL of
 
 `PersistenceCoordinator` owns that orchestration once. A first-party backend composes one (`new PersistenceCoordinator(ctx, this)`), implements the small `PersistenceBackend` hook interface, and delegates its four public service methods to the coordinator. This keeps the duplicated, correctness-heavy orchestration in a single place (it used to receive the same fixes twice).
 
+After an append or load-time repair commits, the coordinator emits the observe-only `session/persisted` notification described in the generated [Cordis event catalog](../../../docs/cordis-catalog/events.md). Its snapshotted header and seq range let derived read models invalidate safely; synchronous dispatch failures and rejected listeners are contained and never fail durability. Truncate-only HMR adoption emits no repair notification while the live session still owns the open turn.
+
 The `PersistenceBackend<TornMarker>` hooks (the only seam between the coordinator and storage):
 
 | Hook | Role |

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

@@ -27,7 +27,7 @@
 import { Context } from 'cordis'
 import { interruptedTurnClosers, SESSION_FORMAT_VERSION } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
-import { assertSerializable, seedCoversPrefix } from './index.ts'
+import { assertSerializable, seedCoversPrefix, type SessionPersistedChange } from './index.ts'
 
 /**
  * A stored session's durable prefix as read back from a backend: its
@@ -229,13 +229,13 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     // event inside it — before the op runs would otherwise have those changes
     // persisted. The clone is taken synchronously (at call time).
     const batch = events.map(e => structuredClone(e))
-    return this.serialize(id, () => this.appendCore(id, batch))
+    return this.serialize(id, () => this._appendCore(id, batch))
   }
 
-  private async appendCore(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
+  private async _appendCore(id: SessionId, events: readonly SessionEvent[]): Promise<void> {
     if (events.length === 0) return
     let state = this.states.get(id)
-    if (state === undefined) state = await this.adopt(id) // calls loadCore, not load
+    if (state === undefined) state = await this.adopt(id) // calls _loadCore, not load
 
     // Contiguity contract: each event's seq must continue the stored log.
     for (const [i, event] of events.entries()) {
@@ -247,8 +247,14 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     await this.backend.appendBatch(state.meta, events, state.materialized)
     // The durable write is the transaction: mark materialized + advance the
     // cursor as soon as it commits (uniform across backends).
+    const fromSeq = state.cursor
     state.materialized = true
     state.cursor += events.length
+    this._notifyPersisted(state.meta, {
+      kind: 'append',
+      fromSeq,
+      toSeq: state.cursor - 1,
+    })
   }
 
   /**
@@ -259,10 +265,10 @@ 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[] }> {
-    return this.serialize(id, () => this.loadCore(id))
+    return this.serialize(id, () => this._loadCore(id))
   }
 
-  private async loadCore(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
+  private async _loadCore(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
     const stored = await this.backend.loadStored(id)
     if (stored === undefined) throw new Error(`session "${id}" not found`)
     const { meta, events, tornMarker } = stored
@@ -281,10 +287,21 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     // there is no state-path ordering dependency (uniform across backends).
     if (tornMarker !== undefined || closers.length > 0) {
       await this.backend.commitRepair(meta, tornMarker, closers)
+      this._notifyPersisted(meta, {
+        kind: 'repair',
+        fromSeq: events.length,
+        toSeq: balanced.length - 1,
+      })
     }
-    // The state keeps its OWN copy of the meta; the returned value is separate so
-    // a consumer mutating loaded.meta cannot corrupt the backend's metadata.
-    this.states.set(id, { meta: { ...meta }, cursor: balanced.length, materialized: true })
+    const owner = this.states.get(id)?.owner
+    // The state keeps its OWN copy of the meta; preserve a live owner already
+    // bound to the id so a read-side load cannot downgrade adoption state.
+    this.states.set(id, {
+      meta: { ...meta },
+      cursor: balanced.length,
+      materialized: true,
+      ...owner !== undefined ? { owner } : {},
+    })
     return { meta, events: balanced }
   }
 
@@ -314,11 +331,11 @@ export class PersistenceCoordinator<TornMarker = unknown> {
 
   /** Build a state for a session discovered in storage but not yet in memory. */
   private async adopt(id: SessionId): Promise<SessionState> {
-    // loadCore (NOT load) — adopt runs inside an already-serialized op, so
+    // _loadCore (NOT load) — adopt runs inside an already-serialized op, so
     // re-entering the chain via the public load() would deadlock.
-    await this.loadCore(id)
+    await this._loadCore(id)
     const state = this.states.get(id)
-    /* v8 ignore next -- loadCore always sets the state for the id */
+    /* v8 ignore next -- _loadCore always sets the state for the id */
     if (!state) throw new Error(`failed to adopt session "${id}"`)
     return state
   }
@@ -475,7 +492,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     // resume.
     const live = await this.backend.loadLive(id, session.header.cwd)
     if (live !== undefined) {
-      // Do NOT route through loadCore(): that crash-repairs open turns as
+      // Do NOT route through _loadCore(): that crash-repairs open turns as
       // interrupted, which is wrong for HMR while the live Session is still the
       // authority and may append the real step/turn end later.
       await this.serialize(id, () => this.adoptLivePrefix(session, seed, live))
@@ -515,7 +532,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
       owner: session,
     })
     const suffix = seed.slice(events.length)
-    if (suffix.length > 0) await this.appendCore(session.header.id, suffix)
+    if (suffix.length > 0) await this._appendCore(session.header.id, suffix)
   }
 
   private async flush(session: Session): Promise<void> {
@@ -546,9 +563,20 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     /* v8 ignore next -- state is always set by the awaited init before flush */
     const cursor = state?.cursor ?? 0
     const fresh = batch.filter(e => e.seq >= cursor)
-    // appendCore (NOT the serialized append) — drain already runs inside the
+    // _appendCore (NOT the serialized append) — drain already runs inside the
     // per-session chain, so re-entering via append() would deadlock.
-    if (fresh.length > 0) await this.appendCore(session.header.id, fresh)
+    if (fresh.length > 0) await this._appendCore(session.header.id, fresh)
     buffer.splice(0, batch.length)
   }
+
+  /** Notify derived read models after source data commits. */
+  private _notifyPersisted(meta: SessionHeader, change: SessionPersistedChange): void {
+    const header = structuredClone(meta)
+    const snapshot = structuredClone(change)
+    void Promise.resolve()
+      .then(() => this.ctx.parallel('session/persisted', header, snapshot))
+      .catch((error: unknown) => {
+        this.ctx.logger.warn(`${this.backend.name}: session/persisted listener failed after ${change.kind} for "${meta.id}": ${String(error)}`)
+      })
+  }
 }

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

@@ -36,6 +36,29 @@ declare module 'cordis' {
   interface Context {
     sessionPersistence: SessionPersistence
   }
+
+  interface Events {
+    /**
+     * A persistence backend committed a canonical session-log change. This is
+     * an observe-only notification for derived read models: the durable write
+     * has already succeeded, and listener failures are contained rather than
+     * propagated into append, load, flush, or teardown.
+     * @param header - snapshotted persisted session metadata.
+     * @param change - committed seq range and whether it was an append or repair.
+     * @mode parallel
+     */
+    'session/persisted'(header: SessionHeader, change: SessionPersistedChange): Promise<void> | void
+  }
+}
+
+/** A committed persisted-log change observed by derived read models. */
+export interface SessionPersistedChange {
+  /** Whether ordinary append or load-time repair committed the change. */
+  kind: 'append' | 'repair'
+  /** First seq affected by the commit. */
+  fromSeq: number
+  /** Last seq appended; less than `fromSeq` when repair only removed a torn fragment. */
+  toSeq: number
 }
 
 /**

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

@@ -31,6 +31,7 @@ import { Context, type Fiber } from 'cordis'
 import SessionStore, { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
 import type { SessionPersistence } from '../src/index.ts'
+import type { SessionPersistedChange } from '../src/index.ts'
 import { meta, oneTurnLog, appendLog } from './contract.ts'
 
 /**
@@ -123,6 +124,40 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
       }
     })
 
+    it('announces committed append and repair ranges without coupling listener failures to writes', async () => {
+      const fix = await makeFixture()
+      const { ctx, fiber } = await freshCtx(fix)
+      const observed: Array<{ headerId: SessionId; change: SessionPersistedChange }> = []
+      ctx.on('session/persisted', (header, change) => {
+        observed.push({ headerId: header.id, change: structuredClone(change) })
+        header.createdAt = -1
+        return Promise.reject(new Error('derived read model failed'))
+      })
+      try {
+        const m = meta('notifications', WORK)
+        await ctx.sessionPersistence.create(m)
+        await expect(ctx.sessionPersistence.append(m.id, oneTurnLog())).resolves.toBeUndefined()
+        await ctx.sessionPersistence.append(m.id, [
+          { type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
+          { type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
+        ])
+        await expect(ctx.sessionPersistence.load(m.id)).resolves.toMatchObject({ meta: { createdAt: m.createdAt } })
+        await Promise.resolve()
+        await Promise.resolve()
+
+        expect(observed).toEqual([
+          { headerId: m.id, change: { kind: 'append', fromSeq: 0, toSeq: 5 } },
+          { headerId: m.id, change: { kind: 'append', fromSeq: 6, toSeq: 7 } },
+          { headerId: m.id, change: { kind: 'repair', fromSeq: 8, toSeq: 9 } },
+        ])
+        expect((await ctx.sessionPersistence.load(m.id)).meta.createdAt).toBe(m.createdAt)
+      } finally {
+        await fiber.dispose()
+        await ctx.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
@@ -368,6 +403,10 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
         // Crash-tail a torn fragment past the (open) committed turn, then reload.
         await first.dispose()
         if (fix.corruptTail) await fix.corruptTail(SessionId('hmr-open'), WORK)
+        const repairs: SessionPersistedChange[] = []
+        ctx.on('session/persisted', (_header, change) => {
+          if (change.kind === 'repair') repairs.push(structuredClone(change))
+        })
         const second = await fix.mount(ctx)
         // The live session is still the authority: it appends the REAL step/turn
         // end. Adoption must truncate the torn tail but NOT synthesize closers.
@@ -378,6 +417,7 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
         const loaded = await ctx.sessionPersistence.load(SessionId('hmr-open'))
         expect(loaded.events.map(e => e.type)).toEqual(['turn/start', 'step/start', 'step/end', 'turn/end'])
         expect(loaded.events.at(-1)).toMatchObject({ type: 'turn/end', data: { reason: { kind: 'completed' } } })
+        expect(repairs).toEqual([])
         await second.dispose()
       } finally {
         await ctx.fiber.dispose()
@@ -385,6 +425,34 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise<
       }
     })
 
+    it('a query-side load preserves the existing live owner binding', async () => {
+      const fix = await makeFixture()
+      const { ctx, fiber } = await freshCtx(fix)
+      let session!: Session
+      const liveFiber = await ctx.plugin(Object.assign((inner: Context) => {
+        session = inner.sessions.create(SessionId('load-owner'), { meta: { cwd: WORK } })
+        send(session, oneTurnLog())
+      }, { inject: ['sessions'] }))
+      try {
+        await ctx.parallel('session/flush', session)
+        const loaded = await ctx.sessionPersistence.load(session.id)
+        await liveFiber.dispose()
+
+        let replacement!: Session
+        await ctx.plugin(Object.assign((inner: Context) => {
+          replacement = inner.sessions.create(session.id, {
+            seed: loaded.events,
+            meta: { cwd: WORK, createdAt: loaded.meta.createdAt },
+          })
+        }, { inject: ['sessions'] }))
+        await expect(inits(ctx.sessionPersistence).get(replacement)).rejects.toThrow(/different live session|id collision/)
+      } finally {
+        await fiber.dispose()
+        await ctx.fiber.dispose()
+        await fix.cleanup()
+      }
+    })
+
     // --- collision / id reuse ---
 
     it('a NEW live session colliding on a persisted id is rejected, not silently adopted', async () => {