feat(session-projection): checkpoint/restore faces for the persisted projection cache

The registry grows the state-level read ladder the persisted cache builds
on: checkpoint(session) snapshots every unit's {stateVersion, observedSeq,
state} row from the watermark cache; restoreFloor(checkpoint) anchors the
tail read one event below the lowest usable watermark (so a shrunk log is
provable); restore(checkpoint, events, baseSeq) refolds each unit from its
usable row (or from init over a full read), rejects rows a tail read cannot
fix (version mismatch / overreach with baseSeq > 0 => re-read from 0), and
returns both the snapshot and the refreshed rows for durable write-back.
ProjectionCheckpointRow/ProjectionCheckpoint are the persisted-row types
minus the record keys.

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

@@ -97,6 +97,27 @@ export interface ProjectionSnapshot {
   values: Partial<SessionProjectionMap>
 }
 
+/**
+ * One unit's checkpoint: its internal state (plain JSON by the unit
+ * contract), the seq of the last event folded into it, and the
+ * `stateVersion` that produced it — the persisted projection-cache row
+ * `(sessionId, key, stateVersion, observedSeq, state)` minus the two outer
+ * keys. A row is never authoritative, only a fold shortcut: `restore`
+ * discards it on a `stateVersion` mismatch or when it claims events past the
+ * stored log end.
+ */
+export interface ProjectionCheckpointRow {
+  /** The registering unit's `stateVersion` at fold time. */
+  stateVersion: number
+  /** Seq of the last event folded into `state`; -1 for the empty log. */
+  observedSeq: number
+  /** The unit's internal state — plain JSON per the unit contract. */
+  state: unknown
+}
+
+/** Checkpoint rows keyed by projection key (one session's persisted cache value). */
+export type ProjectionCheckpoint = Record<string, ProjectionCheckpointRow>
+
 /** Type-erased unit view the drive machinery works with (the register seam already proved the typed contract). */
 interface ErasedDefinition {
   key: string
@@ -206,6 +227,113 @@ export class SessionProjectionRegistry extends Service {
     return { asOfSeq: session.seq - 1, values: values }
   }
 
+  /**
+   * State-level checkpoint of every registered unit for one session, read
+   * from the watermark cache (missing cells fold lazily over the in-memory
+   * log). This is the write side of the persisted projection cache: the
+   * returned rows are the `(key → {stateVersion, observedSeq, state})` part
+   * of the durable `(sessionId, key, stateVersion, observedSeq, state)`
+   * rows. States are the units' live references — plain JSON by the unit
+   * contract, treated as immutable; a durable writer snapshots them at its
+   * own boundary.
+   * @param session - the session whose unit states are checkpointed.
+   * @returns one row per registered key; empty when no unit is registered.
+   */
+  checkpoint(session: Session): ProjectionCheckpoint {
+    const rows: ProjectionCheckpoint = {}
+    for (const registration of this.registrations.values()) {
+      const cell = this.cellFor(registration, session)
+      rows[registration.def.key] = {
+        stateVersion: registration.def.stateVersion,
+        observedSeq: cell.observedSeq,
+        state: cell.state,
+      }
+    }
+    return rows
+  }
+
+  /**
+   * The stored seq a {@link restore} tail read over `checkpoint` must start
+   * at: one event BELOW the lowest usable watermark (a row is usable when
+   * its `stateVersion` matches the live unit; an absent or mismatched row
+   * pulls the floor to `0` — that key must refold the full log). The
+   * one-below anchor is load-bearing: the tail then proves how far the
+   * stored log still extends, so {@link restore} can detect a log that
+   * shrank below a row's watermark (crash-repair truncation) instead of
+   * serving the stale row as current — an empty tail read from the anchor
+   * yields an end below every watermark and the restore rejects for a full
+   * re-read.
+   * @param checkpoint - persisted rows for one session (possibly stale or empty).
+   * @returns the seq to hand the persistence `readFrom`, or `undefined`
+   *   when no unit is registered (no read needed — {@link restore} would
+   *   serve empty values regardless).
+   */
+  restoreFloor(checkpoint: ProjectionCheckpoint): number | undefined {
+    let floor: number | undefined
+    for (const registration of this.registrations.values()) {
+      const row = checkpoint[registration.def.key]
+      const need = row !== undefined && row.stateVersion === registration.def.stateVersion
+        ? Math.max(row.observedSeq + 1, 0)
+        : 0
+      floor = floor === undefined ? need : Math.min(floor, need)
+    }
+    return floor === undefined ? undefined : Math.max(floor - 1, 0)
+  }
+
+  /**
+   * Cold read: fold every registered unit over a stored log suffix, seeding
+   * each from its checkpoint row when usable — the one read recipe (cached
+   * state + forward tail replay + `view`) applied without a live `Session`.
+   * Call with the events returned by a persistence
+   * `readFrom(id, restoreFloor(checkpoint))` and that same floor as
+   * `baseSeq`; the floor's one-below anchor makes the supplied end honest,
+   * so a shrunk log is detected here. A row is usable iff its
+   * `stateVersion` matches the live unit, it does not predate `baseSeq`
+   * (`observedSeq >= baseSeq - 1`), and it does not claim events past the
+   * supplied end (`observedSeq <= endSeq`); an unusable row is discarded
+   * and its key refolds from `init` — which is only sound over the full
+   * log, so a discarded row with `baseSeq > 0` throws (the caller re-reads
+   * from seq 0, e.g. after a crash-repair truncation shrank the log below
+   * a row's watermark).
+   * @param checkpoint - persisted rows for one session (possibly stale or empty).
+   * @param events - the stored events with `seq >= baseSeq`, in seq order.
+   * @param baseSeq - the seq `events` starts at (its first event's seq when non-empty).
+   * @returns the snapshot cut at the supplied log end (`asOfSeq` is the last
+   *   supplied event's seq, `baseSeq - 1` for an empty tail) plus the
+   *   refreshed checkpoint rows at that cut, ready for a durable write-back.
+   */
+  restore(checkpoint: ProjectionCheckpoint, events: readonly SessionEvent[], baseSeq: number):
+  { snapshot: ProjectionSnapshot; checkpoint: ProjectionCheckpoint } {
+    const endSeq = events.at(-1)?.seq ?? baseSeq - 1
+    const values: Record<string, unknown> = {}
+    const refreshed: ProjectionCheckpoint = {}
+    for (const registration of this.registrations.values()) {
+      const def = registration.def
+      const row = checkpoint[def.key]
+      const usable = row !== undefined
+        && row.stateVersion === def.stateVersion
+        && row.observedSeq >= baseSeq - 1
+        && row.observedSeq <= endSeq
+      if (!usable && baseSeq > 0) {
+        throw new Error(
+          `session projection ${JSON.stringify(def.key)} cannot restore from seq ${baseSeq}: `
+          + 'its checkpoint row is missing, version-mismatched, or beyond the supplied log end; re-read from seq 0',
+        )
+      }
+      let state = usable ? row.state : def.init()
+      const from = usable ? row.observedSeq : baseSeq - 1
+      for (const event of events) {
+        if (event.seq > from) state = def.apply(state, event)
+      }
+      values[def.key] = def.schema.parse(def.view(state))
+      refreshed[def.key] = { stateVersion: def.stateVersion, observedSeq: endSeq, state }
+    }
+    return {
+      snapshot: { asOfSeq: endSeq, values: values as ProjectionSnapshot['values'] },
+      checkpoint: refreshed,
+    }
+  }
+
   /** Fold one unit from init over `events`, producing a cell watermarked at the last folded event. */
   private buildCell(def: ErasedDefinition, events: readonly SessionEvent[]): UnitCell {
     let state = def.init()

packages/session-projection/session-projection/tests/registry.spec.ts

@@ -169,6 +169,128 @@ describe('SessionProjectionRegistry drive', () => {
     expect(ctx.sessionProjections.snapshot(session).values).toEqual({})
   })
 
+  it('checkpoints every registered unit with its stateVersion and per-cell watermark', async () => {
+    const { ctx, session } = await harness()
+    ctx.sessionProjections.register(marksUnit())
+    ctx.sessionProjections.register({ ...countUnit(), stateVersion: 7 })
+    const markEvent = mark(session, ['a'])
+    const rows = ctx.sessionProjections.checkpoint(session)
+    expect(rows['test/marks']).toEqual({ stateVersion: 1, observedSeq: markEvent.seq, state: { marks: ['a'] } })
+    expect(rows['test/count']).toEqual({ stateVersion: 7, observedSeq: markEvent.seq, state: 1 })
+    // Empty log: init-derived state at watermark -1.
+    const fresh = ctx.sessions.create()
+    expect(ctx.sessionProjections.checkpoint(fresh)['test/marks']).toEqual({ stateVersion: 1, observedSeq: -1, state: null })
+  })
+
+  it('restoreFloor anchors one below the lowest usable watermark and at 0 for missing or mismatched rows', async () => {
+    const { ctx } = await harness()
+    expect(ctx.sessionProjections.restoreFloor({})).toBeUndefined() // no unit registered
+    ctx.sessionProjections.register(marksUnit())
+    ctx.sessionProjections.register(countUnit())
+    expect(ctx.sessionProjections.restoreFloor({})).toBe(0)
+    // Lowest usable watermark is count's 5 → the anchored tail starts AT 5
+    // (one below the first needed seq 6), so the read proves seq 5 still exists.
+    expect(ctx.sessionProjections.restoreFloor({
+      'test/marks': { stateVersion: 1, observedSeq: 10, state: { marks: [] } },
+      'test/count': { stateVersion: 1, observedSeq: 5, state: 6 },
+    })).toBe(5)
+    // A version-mismatched row forces that key back to a full refold.
+    expect(ctx.sessionProjections.restoreFloor({
+      'test/marks': { stateVersion: 2, observedSeq: 10, state: { marks: [] } },
+      'test/count': { stateVersion: 1, observedSeq: 5, state: 6 },
+    })).toBe(0)
+    // A fresh (-1) row still needs the whole tail from 0.
+    expect(ctx.sessionProjections.restoreFloor({
+      'test/marks': { stateVersion: 1, observedSeq: -1, state: null },
+      'test/count': { stateVersion: 1, observedSeq: -1, state: 0 },
+    })).toBe(0)
+  })
+
+  it('restore folds the tail past each usable row and refolds from init on version mismatch', async () => {
+    const { ctx } = await harness()
+    ctx.sessionProjections.register(marksUnit())
+    ctx.sessionProjections.register(countUnit())
+    const tail: SessionEvent[] = [
+      { type: 'test/mark', seq: 3, time: 3, data: { marks: ['new'] } } as SessionEvent,
+      { type: 'turn/end', seq: 4, time: 4, data: { turn: 1, reason: { kind: 'completed' } } },
+    ]
+    // marks row usable (watermark 2, tail starts at 3); count row mismatched — but
+    // a mismatch with baseSeq > 0 cannot silently refold: it throws for a re-read.
+    expect(() => ctx.sessionProjections.restore({
+      'test/marks': { stateVersion: 1, observedSeq: 2, state: { marks: ['old'] } },
+      'test/count': { stateVersion: 99, observedSeq: 2, state: 3 },
+    }, tail, 3)).toThrow(/re-read from seq 0/)
+    // The full-log re-read (baseSeq 0) refolds the mismatched key from init.
+    const full: SessionEvent[] = [
+      { type: 'turn/start', seq: 0, time: 0, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
+      { type: 'test/mark', seq: 1, time: 1, data: { marks: ['old'] } } as SessionEvent,
+      { type: 'test/mark', seq: 2, time: 2, data: { marks: ['old', '2'] } } as SessionEvent,
+      ...tail,
+    ]
+    const { snapshot, checkpoint } = ctx.sessionProjections.restore({
+      'test/marks': { stateVersion: 1, observedSeq: 2, state: { marks: ['old', '2'] } },
+      'test/count': { stateVersion: 99, observedSeq: 2, state: 3 },
+    }, full, 0)
+    expect(snapshot.asOfSeq).toBe(4)
+    expect(snapshot.values['test/marks']).toEqual({ marks: ['new'] })
+    expect(snapshot.values['test/count']).toBe(5) // refolded from init over all 5 events
+    // The refreshed rows sit at the served cut, ready for a durable write-back.
+    expect(checkpoint['test/marks']).toEqual({ stateVersion: 1, observedSeq: 4, state: { marks: ['new'] } })
+    expect(checkpoint['test/count']).toEqual({ stateVersion: 1, observedSeq: 4, state: 5 })
+  })
+
+  it('restore over a suffix folds only past each row watermark and serves an exact empty-tail cut', async () => {
+    const { ctx } = await harness()
+    ctx.sessionProjections.register(marksUnit())
+    ctx.sessionProjections.register(countUnit())
+    const rows = {
+      'test/marks': { stateVersion: 1, observedSeq: 4, state: { marks: ['done'] } },
+      'test/count': { stateVersion: 1, observedSeq: 2, state: 3 },
+    }
+    const tail: SessionEvent[] = [
+      { type: 'turn/start', seq: 3, time: 3, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
+      { type: 'turn/end', seq: 4, time: 4, data: { turn: 2, reason: { kind: 'completed' } } },
+    ]
+    const { snapshot } = ctx.sessionProjections.restore(rows, tail, 3)
+    expect(snapshot.asOfSeq).toBe(4)
+    // marks already covers the tail (watermark 4): nothing re-applied.
+    expect(snapshot.values['test/marks']).toEqual({ marks: ['done'] })
+    // count folds exactly seqs 3 and 4 on top of its checkpoint.
+    expect(snapshot.values['test/count']).toBe(5)
+
+    // Empty tail (checkpoint is current): the cut sits at baseSeq - 1.
+    const { snapshot: current } = ctx.sessionProjections.restore({
+      'test/marks': { stateVersion: 1, observedSeq: 4, state: { marks: ['done'] } },
+      'test/count': { stateVersion: 1, observedSeq: 4, state: 5 },
+    }, [], 5)
+    expect(current.asOfSeq).toBe(4)
+    expect(current.values['test/count']).toBe(5)
+  })
+
+  it('restore rejects a row claiming events past the supplied log end (shrunk log ⇒ re-read)', async () => {
+    const { ctx } = await harness()
+    ctx.sessionProjections.register(countUnit())
+    const rows = { 'test/count': { stateVersion: 1, observedSeq: 9, state: 10 } }
+    // The anchored floor sits ON the watermark, so the tail read must return
+    // at least seq 9 from an intact log…
+    const floor = ctx.sessionProjections.restoreFloor(rows)
+    expect(floor).toBe(9)
+    // …an intact log serves the anchor event and the checkpoint stands as-is.
+    const anchor: SessionEvent = { type: 'turn/end', seq: 9, time: 9, data: { turn: 2, reason: { kind: 'completed' } } }
+    expect(ctx.sessionProjections.restore(rows, [anchor], 9).snapshot.values['test/count']).toBe(10)
+    // …while a log crash-repaired down to fewer events returns an empty tail:
+    // the row overreaches the proven end and a tail read cannot fix this key.
+    expect(() => ctx.sessionProjections.restore(rows, [], 9)).toThrow(/re-read from seq 0/)
+    // The full re-read discards the overreaching row and refolds from init.
+    const events: SessionEvent[] = [
+      { type: 'turn/start', seq: 0, time: 0, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
+      { type: 'turn/end', seq: 1, time: 1, data: { turn: 1, reason: { kind: 'completed' } } },
+    ]
+    const { snapshot } = ctx.sessionProjections.restore(rows, events, 0)
+    expect(snapshot.asOfSeq).toBe(1)
+    expect(snapshot.values['test/count']).toBe(2)
+  })
+
   it('fails loud when a unit view violates its own schema (async unit output is unrepresentable)', async () => {
     const { ctx, session } = await harness()
     ctx.sessionProjections.register({