feat(session-query): add shared projected observations

packages/session-query/session-query-sqlite/tests/sqlite.spec.ts

@@ -91,6 +91,10 @@ class TestPersistence extends SessionPersistence {
     return undefined
   }
 
+  borrowSession(_id: SessionIdType, _signal?: AbortSignal): ReturnType<SessionPersistence['borrowSession']> {
+    return Promise.reject(new Error('not used'))
+  }
+
   static reset(entries: readonly { meta: SessionHeader; events: SessionEvent[] }[] = []): void {
     this.entries = new Map()
     this.revisions = new Map()

packages/session-query/session-query/package.json

@@ -39,11 +39,19 @@
     "@deepseek-ai/dsh-session-title": "workspace:^",
     "@deepseek-ai/dsh-tool-todo": "workspace:^",
     "@deepseek-ai/dsh-session-persistence": "workspace:^",
+    "@deepseek-ai/dsh-session-projection": "workspace:^",
+    "@deepseek-ai/dsh-session-projection-cache": "workspace:^",
     "@deepseek-ai/cordis": "workspace:^"
   },
   "peerDependenciesMeta": {
     "@deepseek-ai/dsh-session-persistence": {
       "optional": true
+    },
+    "@deepseek-ai/dsh-session-projection": {
+      "optional": true
+    },
+    "@deepseek-ai/dsh-session-projection-cache": {
+      "optional": true
     }
   },
   "devDependencies": {
@@ -54,6 +62,8 @@
     "@deepseek-ai/dsh-session-title": "workspace:^",
     "@deepseek-ai/dsh-tool-todo": "workspace:^",
     "@deepseek-ai/dsh-session-persistence": "workspace:^",
+    "@deepseek-ai/dsh-session-projection": "workspace:^",
+    "@deepseek-ai/dsh-session-projection-cache": "workspace:^",
     "@deepseek-ai/cordis": "workspace:^"
   }
 }

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

@@ -37,6 +37,11 @@ import {
   type Config,
 } from './config.ts'
 import { SessionCorpus } from './corpus.ts'
+import {
+  SessionObservationReader,
+  type SessionObservation,
+  type SessionObservationOptions,
+} from './observation.ts'
 import { buildSessionEventSearchDocuments } from './documents.ts'
 import {
   filterSessionEventDocuments,
@@ -64,6 +69,7 @@ export {
   materializeSessionResultFilters,
 } from './filters.ts'
 export { assertSessionHeadersCompatible } from './sources.ts'
+export type { SessionObservation, SessionObservationOptions } from './observation.ts'
 
 declare module '@deepseek-ai/cordis' {
   interface Context {
@@ -83,6 +89,7 @@ export abstract class SessionQueryEngine extends Service {
 
   private readonly _readWindowMax: number
   private readonly _corpus: SessionCorpus
+  private readonly _observations: SessionObservationReader
 
   constructor(ctx: Context, config: Config = {}) {
     super(ctx, 'sessionQuery')
@@ -102,6 +109,20 @@ export abstract class SessionQueryEngine extends Service {
       )
     }
     this._corpus = new SessionCorpus(ctx, persistedInspectConcurrency)
+    this._observations = new SessionObservationReader(ctx)
+  }
+
+  /**
+   * Observe one exact live or prepared Session without a persistence listing preflight.
+   * @param sessionId - logical Session identity.
+   * @param options - cancellation and projection selection for this read.
+   * @returns a caller-owned observation lease.
+   */
+  observeSession(
+    sessionId: SessionId,
+    options: SessionObservationOptions = {},
+  ): Promise<SessionObservation> {
+    return this._observations.read(sessionId, options)
   }
 
   /**

packages/session-query/session-query/src/observation.ts

@@ -0,0 +1,213 @@
+/** Shared live/prepared observations for Session page and lifecycle consumers. */
+
+import type { Context } from '@deepseek-ai/cordis'
+import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
+import type {
+  BorrowedSessionSource,
+  SessionPersistenceRevision,
+} from '@deepseek-ai/dsh-session-persistence'
+import type { ProjectionSnapshot } from '@deepseek-ai/dsh-session-projection'
+import type {} from '@deepseek-ai/dsh-session-projection-cache'
+import { SessionQueryError } from './config.ts'
+
+/** One exact immutable Session cut retained for the caller's read lifetime. */
+export interface SessionObservation extends Disposable {
+  /** Whether the cut came from an attached Session or a retained preparation. */
+  readonly source: 'live' | 'prepared'
+  /** Immutable Session identity metadata. */
+  readonly header: SessionHeader
+  /** Immutable contiguous events at {@link cursor}. */
+  readonly events: readonly SessionEvent[]
+  /** Last observed event seq, or -1 for an empty log. */
+  readonly cursor: number
+  /** Durable source revision for a cold prepared observation. */
+  readonly revision?: SessionPersistenceRevision
+  /** Exact projection baseline at {@link cursor}, when the registry is mounted. */
+  readonly projections?: ProjectionSnapshot
+  /**
+   * Retain the same immutable cut for another Host owner.
+   * @returns an independently disposable lease over this observation.
+   */
+  retain(): SessionObservation
+}
+
+/** Projection work and cancellation requested for one exact observation. */
+export interface SessionObservationOptions {
+  /** Optional cancellation while resolving a cold source. */
+  readonly signal?: AbortSignal
+  /** Whether to compute every projection or leave projection state untouched. */
+  readonly projectionMode?: 'all' | 'none'
+}
+
+/** Builds point observations without a corpus listing preflight. */
+export class SessionObservationReader {
+  /** @param ctx - context carrying Session and optional persistence/projection services. */
+  constructor(private readonly ctx: Context) {}
+
+  /**
+   * Observe one live-preferred Session and retain a cold preparation until disposal.
+   * @param sessionId - logical Session identity.
+   * @param options - cancellation and all-or-none projection computation for this read.
+   * @returns one exact immutable observation.
+   */
+  async read(
+    sessionId: SessionId,
+    options: SessionObservationOptions = {},
+  ): Promise<SessionObservation> {
+    const { signal, projectionMode = 'all' } = options
+    for (;;) {
+      throwIfObservationAborted(signal)
+      const live = this.ctx.sessions.get(sessionId)
+      if (live !== undefined) return this.live(live, projectionMode)
+      const persistence = this.ctx.get('sessionPersistence')
+      if (persistence === undefined) throw notFound(sessionId)
+
+      let borrowed: BorrowedSessionSource
+      try {
+        borrowed = await persistence.borrowSession(sessionId, signal)
+      } catch (error: unknown) {
+        throwIfObservationAborted(signal)
+        if (hasErrorName(error, 'SessionPersistenceNotFoundError')) throw notFound(sessionId, error)
+        if (hasErrorName(error, 'SessionPersistenceCorruptionError')) {
+          throw new SessionQueryError(
+            `stored session "${sessionId}" is corrupt: ${error.message}`,
+            'SESSION_QUERY_CORRUPT_SESSION',
+            { cause: error },
+          )
+        }
+        throw new SessionQueryError(
+          `failed to observe session "${sessionId}": ${errorMessage(error)}`,
+          'SESSION_QUERY_PERSISTENCE_FAILED',
+          { cause: error },
+        )
+      }
+
+      try {
+        throwIfObservationAborted(signal)
+        if (borrowed.inspection.meta.id !== sessionId) {
+          throw new SessionQueryError(
+            `session persistence returned "${borrowed.inspection.meta.id}" for "${sessionId}"`,
+            'SESSION_QUERY_SOURCE_CONFLICT',
+          )
+        }
+        const attached = this.ctx.sessions.get(sessionId)
+        if (attached !== undefined) {
+          const liveObservation = this.live(attached, projectionMode)
+          borrowed[Symbol.dispose]()
+          return liveObservation
+        }
+        if (borrowed.source === 'live') {
+          // The live Session disappeared between persistence's race check and
+          // this read. Retry against its now-cold durable identity.
+          borrowed[Symbol.dispose]()
+          continue
+        }
+        const prepared = borrowed
+        const events = prepared.inspection.events
+        let projections: ProjectionSnapshot | undefined
+        try {
+          projections = projectionMode === 'none'
+            ? undefined
+            : this.preparedProjections(prepared, events)
+        } catch (error: unknown) {
+          throw new SessionQueryError(
+            `failed to project session "${sessionId}": ${errorMessage(error)}`,
+            'SESSION_QUERY_CORRUPT_SESSION',
+            { cause: error },
+          )
+        }
+        let references = 1
+        const lease = (): SessionObservation => {
+          let disposed = false
+          return {
+            source: 'prepared',
+            header: prepared.inspection.meta,
+            events,
+            cursor: events.at(-1)?.seq ?? -1,
+            revision: prepared.revision,
+            ...projections === undefined ? {} : { projections },
+            retain: () => {
+              if (disposed || references === 0) throw new Error(`session observation "${sessionId}" is disposed`)
+              references += 1
+              return lease()
+            },
+            [Symbol.dispose]: () => {
+              if (disposed) return
+              disposed = true
+              references -= 1
+              if (references === 0) prepared[Symbol.dispose]()
+            },
+          }
+        }
+        return lease()
+      } catch (error: unknown) {
+        borrowed[Symbol.dispose]()
+        throw error
+      }
+    }
+  }
+
+  private live(
+    session: Session,
+    projectionMode: NonNullable<SessionObservationOptions['projectionMode']>,
+  ): SessionObservation {
+    const events = Object.freeze([...session.events])
+    const projections = projectionMode === 'none'
+      ? undefined
+      : this.ctx.get('sessionProjections')?.snapshot(session)
+    const lease = (): SessionObservation => {
+      let disposed = false
+      return {
+        source: 'live',
+        header: session.header,
+        events,
+        cursor: events.at(-1)?.seq ?? -1,
+        ...projections === undefined ? {} : { projections },
+        retain: () => {
+          if (disposed) throw new Error(`session observation "${session.id}" is disposed`)
+          return lease()
+        },
+        [Symbol.dispose]: () => { disposed = true },
+      }
+    }
+    return lease()
+  }
+
+  private preparedProjections(
+    observation: Extract<BorrowedSessionSource, { readonly source: 'prepared' }>,
+    events: readonly SessionEvent[],
+  ): ProjectionSnapshot | undefined {
+    const registry = this.ctx.get('sessionProjections')
+    if (registry === undefined) return undefined
+    const prepared = observation.preparedSession
+    const cache = this.ctx.get('sessionProjectionCache')
+    return cache === undefined
+      ? registry.hydrate(prepared, {}, events, 0)
+      : cache.hydratePrepared(prepared, observation.inspection.meta, events)
+  }
+}
+
+function throwIfObservationAborted(signal: AbortSignal | undefined): void {
+  if (signal?.aborted !== true) return
+  throw new SessionQueryError(
+    'session observation was aborted',
+    'SESSION_QUERY_ABORTED',
+    { cause: signal.reason },
+  )
+}
+
+function notFound(sessionId: SessionId, cause?: unknown): SessionQueryError {
+  return new SessionQueryError(
+    `session "${sessionId}" not found`,
+    'SESSION_QUERY_SESSION_NOT_FOUND',
+    cause === undefined ? undefined : { cause },
+  )
+}
+
+function errorMessage(error: unknown): string {
+  return error instanceof Error ? error.message : 'unknown error'
+}
+
+function hasErrorName(error: unknown, name: string): error is Error {
+  return error instanceof Error && error.name === name
+}

packages/session-query/session-query/tests/observation.spec.ts

@@ -0,0 +1,141 @@
+import { Context } from '@deepseek-ai/cordis'
+import SessionStore, { Session, SessionId } from '@deepseek-ai/dsh-session'
+import type { SessionHeader } from '@deepseek-ai/dsh-session'
+import { SessionPersistenceRevision } from '@deepseek-ai/dsh-session-persistence'
+import type { BorrowedSessionSource } from '@deepseek-ai/dsh-session-persistence'
+import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
+import { describe, expect, it, vi } from 'vitest'
+import { SessionObservationReader } from '../src/observation.ts'
+
+function header(id: string): SessionHeader {
+  return { version: 0, id: SessionId(id), createdAt: 1, cwd: '/workspace' }
+}
+
+function preparedSource(
+  meta: SessionHeader,
+  dispose = vi.fn(),
+): BorrowedSessionSource {
+  const preparedSession = Session.create(meta.id, [], meta)
+  return {
+    source: 'prepared',
+    inspection: { meta: preparedSession.header, events: preparedSession.events },
+    revision: SessionPersistenceRevision(`fixture:${meta.id}`),
+    preparedSession,
+    [Symbol.dispose]: dispose,
+  }
+}
+
+describe('SessionObservationReader', () => {
+  it('prefers a live Session that attaches while a prepared source is borrowed', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const meta = header('attached-during-borrow')
+    const dispose = vi.fn()
+    const prepared = preparedSource(meta, dispose)
+    ctx.provide('sessionPersistence', {
+      borrowSession: () => {
+        ctx.sessions.create(meta.id, { meta })
+        return Promise.resolve(prepared)
+      },
+    } as never)
+
+    using observed = await new SessionObservationReader(ctx).read(meta.id, { projectionMode: 'none' })
+
+    expect(observed.source).toBe('live')
+    expect(dispose).toHaveBeenCalledOnce()
+    await ctx.fiber.dispose()
+  })
+
+  it('releases a borrowed source once when the winning live projection fails', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    await ctx.plugin(SessionProjectionRegistry)
+    const meta = header('attached-projection-failure')
+    const dispose = vi.fn()
+    const prepared = preparedSource(meta, dispose)
+    ctx.provide('sessionPersistence', {
+      borrowSession: () => {
+        ctx.sessions.create(meta.id, { meta })
+        return Promise.resolve(prepared)
+      },
+    } as never)
+    vi.spyOn(ctx.sessionProjections, 'snapshot').mockImplementation(() => {
+      throw new Error('projection failed')
+    })
+
+    await expect(new SessionObservationReader(ctx).read(meta.id)).rejects.toThrow('projection failed')
+    expect(dispose).toHaveBeenCalledOnce()
+    await ctx.fiber.dispose()
+  })
+
+  it('retries when persistence reports a live source that has already detached', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const meta = header('detached-live-source')
+    const disposeLive = vi.fn()
+    const prepared = preparedSource(meta)
+    const borrowSession = vi.fn()
+      .mockResolvedValueOnce({
+        source: 'live', inspection: { meta, events: [] }, [Symbol.dispose]: disposeLive,
+      } satisfies BorrowedSessionSource)
+      .mockResolvedValueOnce(prepared)
+    ctx.provide('sessionPersistence', { borrowSession } as never)
+
+    using observed = await new SessionObservationReader(ctx).read(meta.id, { projectionMode: 'none' })
+
+    expect(observed.source).toBe('prepared')
+    expect(borrowSession).toHaveBeenCalledTimes(2)
+    expect(disposeLive).toHaveBeenCalledOnce()
+    await ctx.fiber.dispose()
+  })
+
+  it('reference-counts prepared leases and rejects retention after disposal', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const meta = header('prepared-leases')
+    const dispose = vi.fn()
+    ctx.provide('sessionPersistence', {
+      borrowSession: () => Promise.resolve(preparedSource(meta, dispose)),
+    } as never)
+    const observed = await new SessionObservationReader(ctx).read(meta.id, { projectionMode: 'none' })
+    const retained = observed.retain()
+
+    observed[Symbol.dispose]()
+    observed[Symbol.dispose]()
+    expect(dispose).not.toHaveBeenCalled()
+    expect(() => observed.retain()).toThrow('is disposed')
+    retained[Symbol.dispose]()
+    expect(dispose).toHaveBeenCalledOnce()
+    await ctx.fiber.dispose()
+  })
+
+  it('creates independent live leases and rejects retention after disposal', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const session = ctx.sessions.create(SessionId('live-leases'), { meta: { cwd: '/workspace' } })
+    const reader = new SessionObservationReader(ctx)
+    const observed = await reader.read(session.id, { projectionMode: 'none' })
+    const retained = observed.retain()
+
+    observed[Symbol.dispose]()
+    expect(() => observed.retain()).toThrow('is disposed')
+    expect(retained.source).toBe('live')
+    retained[Symbol.dispose]()
+    await ctx.fiber.dispose()
+  })
+
+  it('contains a non-Error persistence rejection', async () => {
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    ctx.provide('sessionPersistence', {
+      // Exercise containment of a backend that violates the Error rejection convention.
+      borrowSession: () => Promise.reject('offline'), // oxlint-disable-line typescript/prefer-promise-reject-errors
+    } as never)
+
+    await expect(new SessionObservationReader(ctx).read(SessionId('failed'))).rejects.toMatchObject({
+      code: 'SESSION_QUERY_PERSISTENCE_FAILED',
+      message: expect.stringContaining('unknown error') as string,
+    })
+    await ctx.fiber.dispose()
+  })
+})

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

@@ -64,6 +64,10 @@ class TestPersistence extends SessionPersistence {
     return undefined
   }
 
+  borrowSession(_id: SessionIdType, _signal?: AbortSignal): ReturnType<SessionPersistence['borrowSession']> {
+    return Promise.reject(new Error('not used'))
+  }
+
   create(meta: SessionHeader): Promise<void> {
     TestPersistence.entries.set(meta.id, { meta: structuredClone(meta), events: [] })
     return Promise.resolve()

packages/session-query/session-query/tests/tracing.spec.ts

@@ -54,6 +54,10 @@ class TracePersistence extends SessionPersistence {
     return undefined
   }
 
+  borrowSession(_id: SessionIdType, _signal?: AbortSignal): ReturnType<SessionPersistence['borrowSession']> {
+    return Promise.reject(new Error('not used'))
+  }
+
   create(meta: SessionHeader): Promise<void> {
     TracePersistence.entries.set(meta.id, { meta: structuredClone(meta), events: [] })
     return Promise.resolve()

packages/session-query/session-query/tsconfig.json

@@ -32,6 +32,12 @@
     {
       "path": "../../session/session-persistence"
     },
+    {
+      "path": "../../session/session-projection"
+    },
+    {
+      "path": "../../session/session-projection-cache"
+    },
     {
       "path": "../../runtime-diagnostics/invariants"
     }

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

@@ -19,7 +19,11 @@ import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-
 // Empty type import: applies the package's cordis Context merge
 // (`ctx.sessionPersistence`), which this service reads on the cold path.
 import type {} from '@deepseek-ai/dsh-session-persistence'
-import type { ProjectionCheckpoint, ProjectionSnapshot } from '@deepseek-ai/dsh-session-projection'
+import type {
+  ProjectionCheckpoint,
+  ProjectionSnapshot,
+  SessionProjectionMap,
+} from '@deepseek-ai/dsh-session-projection'
 import type { KvTable } from '@deepseek-ai/dsh-storage-domain'
 import { projectionCacheDomainSpec } from './spec.ts'
 import type { CheckpointIdentity, CheckpointRecord } from './spec.ts'
@@ -113,22 +117,54 @@ export class SessionProjectionCache extends Service {
    * paths (the history tail baseline, {@link coldSnapshot}) supersede these
    * values whenever a session is actually opened.
    * @param meta - the listed session's header (identity witness; no log read).
+   * @param keys - optional projection keys required by the caller's audience.
    * @returns the cut (`asOfSeq` = lowest served-row watermark), or
    *   `undefined` when no usable row exists for this lifecycle.
    */
-  cachedSnapshot(meta: SessionHeader): ProjectionSnapshot | undefined {
+  cachedSnapshot(
+    meta: SessionHeader,
+    keys?: readonly Extract<keyof SessionProjectionMap, string>[],
+  ): ProjectionSnapshot | undefined {
     const record = this.recordFor(meta.id, identityOf(meta))
     if (record === undefined) return undefined
-    const values = this.ctx.sessionProjections.viewCheckpoint(record.rows)
-    const keys = Object.keys(values)
-    if (keys.length === 0) return undefined
+    const values = this.ctx.sessionProjections.viewCheckpoint(record.rows, keys)
+    const servedKeys = Object.keys(values)
+    if (servedKeys.length === 0) return undefined
     // The block carries ONE cut: the lowest served watermark is the seq every
     // value is at least current as of (under-claiming is safe under
     // higher-seq-wins; over-claiming would let a stale value outrank pushes).
-    const asOfSeq = Math.min(...keys.map(key => (record.rows[key] as { seq: number }).seq))
+    const asOfSeq = Math.min(...servedKeys.map(key => (record.rows[key] as { seq: number }).seq))
     return { asOfSeq, values }
   }
 
+  /**
+   * Hydrate projection cells for an already-prepared Session without another
+   * persistence read. The cache seeds matching rows; the supplied exact log
+   * advances every unit to the observation cut. No checkpoint is written
+   * because the logical observation may contain recovery events not yet durable.
+   * @param session - exact unpublished Session retained by persistence.
+   * @param meta - observed lifecycle header.
+   * @param events - exact logical event prefix represented by the observation.
+   * @returns all projection values at the event cut.
+   */
+  hydratePrepared(
+    session: Session,
+    meta: SessionHeader,
+    events: readonly SessionEvent[],
+  ): ProjectionSnapshot {
+    const record = this.recordFor(meta.id, identityOf(meta))
+    if (record === undefined) {
+      return this.ctx.sessionProjections.hydrate(session, {}, events, 0)
+    }
+    try {
+      return this.ctx.sessionProjections.hydrate(session, record.rows, events, 0)
+    } catch {
+      // Cached rows are disposable derived data. Retry from the exact log so a
+      // stale schema cannot make a valid Session unreadable.
+      return this.ctx.sessionProjections.hydrate(session, {}, events, 0)
+    }
+  }
+
   /**
    * Durably checkpoint one live session NOW (both mandatory points call
    * this; tests and carriers may too). The registry cut is snapshotted at
@@ -183,13 +219,13 @@ export class SessionProjectionCache extends Service {
     const related = record === undefined || identityMatches(record.identity, identityOf(tail.meta))
     try {
       if (!related) throw new Error('unrelated log identity')
-      restored = this.ctx.sessionProjections.restore(cached, tail.events, floor)
+      restored = this.ctx.sessionProjections.restore(cached, tail.events, floor, tail.meta)
     } catch {
       // Recoverable failures are an unrelated record, a row outside the
       // supplied suffix or log end, and stateSchema rejection. The full read
       // removes every checkpoint seed and lets each unit refold from init.
       const whole = await persistence.readFrom(id, 0, signal)
-      restored = this.ctx.sessionProjections.restore({}, whole.events, 0)
+      restored = this.ctx.sessionProjections.restore({}, whole.events, 0, whole.meta)
     }
     await this.putSoft(id, identityOf(tail.meta), restored.checkpoint, 'cold-read write-back')
     return restored.snapshot

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

@@ -11,8 +11,8 @@ import { Context } from '@deepseek-ai/cordis'
 import { z } from 'zod'
 import Storage from '@deepseek-ai/dsh-storage'
 import { DomainFacility } from '@deepseek-ai/dsh-storage-domain'
-import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
-import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
+import SessionStore, { Session, SessionId } from '@deepseek-ai/dsh-session'
+import type { SessionEvent } from '@deepseek-ai/dsh-session'
 import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
 import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
 import { MemoryMediaPool, MemoryStorageBackend } from '../../../storage/storage-domain/tests/helpers/memory-backend.ts'
@@ -250,6 +250,21 @@ describe('SessionProjectionCache cold read', () => {
     })
   }
 
+  it('retries prepared hydration without a malformed cached checkpoint', async () => {
+    const pool = new MemoryMediaPool()
+    const id = SessionId('prepared-cache-fallback')
+    seedRow(pool, id, { ver: 1, seq: 1, val: { marks: 'malformed' } })
+    const events = storedLog([['fresh']])
+    const { cache } = await harness({ pool })
+    const meta = headerOf(id)
+    const session = Session.create(id, events, meta)
+
+    expect(cache.hydratePrepared(session, meta, events)).toEqual({
+      asOfSeq: 2,
+      values: { 'cache-test/marks': { marks: ['fresh'] } },
+    })
+  })
+
   it('serves a cold session from the cache row plus a bounded tail read, and writes the refresh back', async () => {
     const pool = new MemoryMediaPool()
     const logs = new Map([['cold', storedLog([['a'], ['a', 'b']])]])

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

@@ -19,7 +19,7 @@
 
 import { Context, Service } from '@deepseek-ai/cordis'
 import type { ZodType } from 'zod'
-import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
+import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
 
 declare module '@deepseek-ai/cordis' {
   interface Context {
@@ -48,10 +48,11 @@ export interface ProjectionDefinition<
   /** Validates persisted state before it seeds a fold. */
   stateSchema: ZodType<S>
   /**
-   * State for the empty log.
+   * State for the empty log and its immutable Session metadata.
+   * @param header - immutable metadata for the Session being projected.
    * @returns the initial state.
    */
-  init(): NoInfer<S>
+  init(header: SessionHeader): NoInfer<S>
   /**
    * Pure transition: previous state + one committed event → next state. A
    * unit uninterested in an event MUST return the same state reference — an
@@ -129,7 +130,7 @@ export type ProjectionCheckpoint = Record<string, ProjectionCheckpointRow>
 interface ErasedDefinition {
   key: string
   stateSchema: { parse(value: unknown): unknown }
-  init(): unknown
+  init(header: SessionHeader): unknown
   apply(state: unknown, event: SessionEvent): unknown
   wire: { viewSchema: { parse(value: unknown): unknown }; view(state: unknown): unknown } | undefined
   stateVersion: number
@@ -187,6 +188,16 @@ export class SessionProjectionRegistry extends Service {
    */
   constructor(ctx: Context) {
     super(ctx, 'sessionProjections')
+    ctx.on('session/created', (session: Session) => {
+      if (session.seq !== 0) return
+      for (const registration of this.registrations.values()) {
+        if (registration.cells.has(session)) continue
+        registration.cells.set(session, {
+          state: registration.def.init(session.header),
+          observedSeq: -1,
+        })
+      }
+    })
     ctx.on('session/event', (session: Session, event: SessionEvent) => {
       this.drive(session, event)
     })
@@ -230,7 +241,7 @@ export class SessionProjectionRegistry extends Service {
     const erased: ErasedDefinition = {
       key: definition.key,
       stateSchema: definition.stateSchema,
-      init: () => definition.init(),
+      init: header => definition.init(header),
       apply: (state, event) => definition.apply(state as S, event),
       wire: wire === undefined
         ? undefined
@@ -279,7 +290,8 @@ export class SessionProjectionRegistry extends Service {
   }
 
   /**
-   * Read one unit's current host state without computing unrelated views.
+   * Read one unit's current host state after materializing every registered
+   * unit at the Session cursor. Unrelated wire views are not produced.
    * The returned value is live; callers must not mutate it.
    * @param session - the session whose state is read.
    * @param key - the registered unit key.
@@ -291,6 +303,7 @@ export class SessionProjectionRegistry extends Service {
   ): SessionProjectionStateMap[K] | undefined {
     const registration = this.registrations.get(key)
     if (registration === undefined) return undefined
+    this.materializeCells(session)
     return this.cellFor(registration, session).state as SessionProjectionStateMap[K]
   }
 
@@ -300,18 +313,53 @@ export class SessionProjectionRegistry extends Service {
    * Fully synchronous — every value and `asOfSeq` reflect the same log
    * position. Each value passes its unit's `viewSchema` before leaving.
    * @param session - the session whose projection values are read.
-   * @returns the snapshot; `values` is empty when no client-visible unit is registered.
+   * @param keys - optional client-visible outputs; state materialization remains complete.
+   * @returns the snapshot; `values` is empty when no selected client-visible unit is registered.
    */
-  snapshot(session: Session): ProjectionSnapshot {
+  snapshot(
+    session: Session,
+    keys?: readonly Extract<keyof SessionProjectionMap, string>[],
+  ): ProjectionSnapshot {
     const values: Record<string, unknown> = {}
+    const selected = keys === undefined ? undefined : new Set<string>(keys)
+    this.materializeCells(session)
     for (const registration of this.registrations.values()) {
       if (registration.def.wire === undefined) continue
+      if (selected !== undefined && !selected.has(registration.def.key)) continue
       const cell = this.cellFor(registration, session)
-      values[registration.def.key] = registration.def.wire.viewSchema.parse(registration.def.wire.view(cell.state))
+      values[registration.def.key] = this.viewCell(registration, cell)
     }
     return { asOfSeq: session.seq - 1, values }
   }
 
+  /**
+   * Read only already-materialized client-visible cells without folding history.
+   * Values may trail the live Session and are therefore hints, not a complete
+   * baseline. Missing cells are omitted.
+   * @param session - attached Session whose cached cells are inspected.
+   * @param keys - optional wire keys to view.
+   * @returns the lowest common cached cut, or `undefined` when no wire cell exists.
+   */
+  cachedSnapshot(
+    session: Session,
+    keys?: readonly Extract<keyof SessionProjectionMap, string>[],
+  ): ProjectionSnapshot | undefined {
+    const values: Record<string, unknown> = {}
+    let asOfSeq: number | undefined
+    const selected = keys === undefined ? undefined : new Set<string>(keys)
+    for (const registration of this.registrations.values()) {
+      if (registration.def.wire === undefined) continue
+      if (selected !== undefined && !selected.has(registration.def.key)) continue
+      const cell = registration.cells.get(session)
+      if (cell === undefined) continue
+      values[registration.def.key] = this.viewCell(registration, cell)
+      asOfSeq = asOfSeq === undefined
+        ? cell.observedSeq
+        : Math.min(asOfSeq, cell.observedSeq)
+    }
+    return asOfSeq === undefined ? undefined : { asOfSeq, values }
+  }
+
   /**
    * State-level checkpoint of every persisted unit for one session, read
    * from the watermark cache (missing cells fold lazily over the in-memory
@@ -375,13 +423,19 @@ export class SessionProjectionRegistry extends Service {
    * fuller read path refolds it). The zero-I/O rung of the read ladder —
    * values are as stale as their rows, never wrong.
    * @param checkpoint - persisted rows for one session (possibly stale or empty).
+   * @param keys - optional wire keys to view.
    * @returns whole values per key with a usable row; empty when none.
    */
-  viewCheckpoint(checkpoint: ProjectionCheckpoint): Partial<SessionProjectionMap> {
+  viewCheckpoint(
+    checkpoint: ProjectionCheckpoint,
+    keys?: readonly Extract<keyof SessionProjectionMap, string>[],
+  ): Partial<SessionProjectionMap> {
     const values: Record<string, unknown> = {}
+    const selected = keys === undefined ? undefined : new Set<string>(keys)
     for (const registration of this.registrations.values()) {
       const def = registration.def
       if (def.wire === undefined) continue
+      if (selected !== undefined && !selected.has(def.key)) continue
       const row = checkpoint[def.key]
       if (row === undefined || row.ver !== def.stateVersion) continue
       let state: unknown
@@ -413,6 +467,7 @@ export class SessionProjectionRegistry extends Service {
    * @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).
+   * @param header - immutable metadata for the Session being restored.
    * @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.
@@ -421,6 +476,7 @@ export class SessionProjectionRegistry extends Service {
     checkpoint: ProjectionCheckpoint,
     events: readonly SessionEvent[],
     baseSeq: number,
+    header: SessionHeader,
   ):
   { snapshot: ProjectionSnapshot; checkpoint: ProjectionCheckpoint } {
     const endSeq = events.at(-1)?.seq ?? baseSeq - 1
@@ -439,10 +495,16 @@ export class SessionProjectionRegistry extends Service {
           + 'its checkpoint row is missing, version-mismatched, or beyond the supplied log end; re-read from seq 0',
         )
       }
-      let state = usable ? def.stateSchema.parse(row.val) : def.init()
+      let state = usable ? def.stateSchema.parse(row.val) : def.init(header)
       const from = usable ? row.seq : baseSeq - 1
-      for (const event of events) {
-        if (event.seq > from) state = def.apply(state, event)
+      const startIndex = from - baseSeq + 1
+      for (let index = startIndex; index < events.length; index++) {
+        const event = events[index]
+        const expectedSeq = baseSeq + index
+        if (event === undefined || event.seq !== expectedSeq) {
+          throw new Error(`session projection ${JSON.stringify(def.key)} cannot restore across missing seq ${String(expectedSeq)}`)
+        }
+        state = def.apply(state, event)
       }
       if (def.wire !== undefined) values[def.key] = def.wire.viewSchema.parse(def.wire.view(state))
       refreshed[def.key] = { ver: def.stateVersion, seq: endSeq, val: state }
@@ -453,9 +515,66 @@ export class SessionProjectionRegistry extends Service {
     }
   }
 
+  /**
+   * Restore an exact cut and install its states on the supplied prepared Session.
+   * A later publication reuses these cells; ordinary live reads and event drive
+   * advance any constructor-owned suffix exactly once.
+   * @param session - exact prepared Session that owns the restored log prefix.
+   * @param checkpoint - persisted rows for this Session lifecycle.
+   * @param events - exact events at the observation cut.
+   * @param baseSeq - first supplied event sequence.
+   * @returns all projection values at the supplied cut.
+   */
+  hydrate(
+    session: Session,
+    checkpoint: ProjectionCheckpoint,
+    events: readonly SessionEvent[],
+    baseSeq: number,
+  ): ProjectionSnapshot {
+    const endSeq = events.at(-1)?.seq ?? baseSeq - 1
+    let complete = true
+    for (const registration of this.registrations.values()) {
+      const current = registration.cells.get(session)
+      if (current?.observedSeq !== endSeq) {
+        complete = false
+        break
+      }
+    }
+    if (complete) {
+      const values: Record<string, unknown> = {}
+      for (const registration of this.registrations.values()) {
+        if (registration.def.wire === undefined) continue
+        const current = registration.cells.get(session) as UnitCell
+        values[registration.def.key] = this.viewCell(registration, current)
+      }
+      return { asOfSeq: endSeq, values }
+    }
+    const restored = this.restore(checkpoint, events, baseSeq, session.header)
+    for (const registration of this.registrations.values()) {
+      const row = restored.checkpoint[registration.def.key]
+      if (row === undefined) continue
+      const current = registration.cells.get(session)
+      if (current !== undefined && current.observedSeq > row.seq) continue
+      registration.cells.set(session, {
+        state: row.val,
+        observedSeq: row.seq,
+      })
+    }
+    return restored.snapshot
+  }
+
+  /** Materialize every registered unit cell at the Session's current cursor. */
+  private materializeCells(session: Session): void {
+    for (const registration of this.registrations.values()) this.cellFor(registration, session)
+  }
+
   /** 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()
+  private buildCell(
+    def: ErasedDefinition,
+    header: SessionHeader,
+    events: readonly SessionEvent[],
+  ): UnitCell {
+    let state = def.init(header)
     for (const event of events) state = def.apply(state, event)
     return { state, observedSeq: (events.at(-1)?.seq ?? -1) }
   }
@@ -464,34 +583,65 @@ export class SessionProjectionRegistry extends Service {
   private cellFor(registration: Registration, session: Session): UnitCell {
     let cell = registration.cells.get(session)
     if (cell === undefined) {
-      cell = this.buildCell(registration.def, session.events)
+      cell = this.buildCell(registration.def, session.header, session.events)
       registration.cells.set(session, cell)
+    } else {
+      this.advanceCell(registration.def, cell, session.events, session.seq - 1)
     }
     return cell
   }
 
+  /** Advance one existing cell through a contiguous Session prefix. */
+  private advanceCell(
+    def: ErasedDefinition,
+    cell: UnitCell,
+    events: readonly SessionEvent[],
+    throughSeq: number,
+  ): void {
+    if (cell.observedSeq >= throughSeq) return
+    for (let seq = cell.observedSeq + 1; seq <= throughSeq; seq++) {
+      const event = events[seq]
+      if (event === undefined || event.seq !== seq) {
+        throw new Error(`session projection ${JSON.stringify(def.key)} cannot advance across missing seq ${String(seq)}`)
+      }
+      const next = def.apply(cell.state, event)
+      cell.state = next
+      cell.observedSeq = seq
+    }
+  }
+
   /** Eager drive: pass one committed event through every registered unit; notify on changed references. */
   private drive(session: Session, event: SessionEvent): void {
     for (const registration of this.registrations.values()) {
       let cell = registration.cells.get(session)
+      if (cell !== undefined && cell.observedSeq >= event.seq) continue
       if (cell === undefined) {
         // Late build mid-stream: fold history before this event (seq = log
         // index, so the prefix slice is exact), then take the normal gate.
-        cell = this.buildCell(registration.def, session.events.slice(0, event.seq))
+        cell = this.buildCell(registration.def, session.header, session.events.slice(0, event.seq))
         registration.cells.set(session, cell)
+      } else {
+        this.advanceCell(registration.def, cell, session.events, event.seq - 1)
       }
       const next = registration.def.apply(cell.state, event)
       const changed = !Object.is(next, cell.state)
       cell.state = next
       cell.observedSeq = event.seq
       if (changed && registration.def.wire !== undefined && this.listeners.size > 0) {
-        const value = registration.def.wire.viewSchema.parse(registration.def.wire.view(next))
+        const value = this.viewCell(registration, cell)
         for (const listener of this.listeners) {
           listener(session, registration.def.key as Extract<keyof SessionProjectionMap, string>, value, event.seq)
         }
       }
     }
   }
+
+  /** Return one schema-validated wire value. */
+  private viewCell(registration: Registration, cell: UnitCell): unknown {
+    const wire = registration.def.wire
+    if (wire === undefined) throw new Error(`session projection ${JSON.stringify(registration.def.key)} has no wire view`)
+    return wire.viewSchema.parse(wire.view(cell.state))
+  }
 }
 
 export default SessionProjectionRegistry

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

@@ -10,8 +10,8 @@
 import { describe, expect, it } from 'vitest'
 import { Context } from '@deepseek-ai/cordis'
 import { z } from 'zod'
-import SessionStore from '@deepseek-ai/dsh-session'
-import type { Session, SessionEvent } from '@deepseek-ai/dsh-session'
+import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
+import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
 import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
 import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
 
@@ -33,6 +33,11 @@ declare module '@deepseek-ai/dsh-session/types' {
 }
 
 type MarksState = { marks: string[] } | null
+const RESTORE_HEADER: SessionHeader = {
+  version: 0,
+  id: SessionId('projection-restore'),
+  createdAt: 0,
+}
 /** Whole-value unit: latest test/mark event wins; unrelated events return the same reference. */
 const marksUnit = (): Omit<ProjectionDefinition<'test/marks', MarksState>, 'wire'>
   & { wire: NonNullable<ProjectionDefinition<'test/marks', MarksState>['wire']> } => ({
@@ -280,7 +285,7 @@ describe('SessionProjectionRegistry drive', () => {
     expect(() => ctx.sessionProjections.restore({
       'test/marks': { ver: 1, seq: 2, val: { marks: ['old'] } },
       'test/count': { ver: 99, seq: 2, val: 3 },
-    }, tail, 3)).toThrow(/re-read from seq 0/)
+    }, tail, 3, RESTORE_HEADER)).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 } },
@@ -291,7 +296,7 @@ describe('SessionProjectionRegistry drive', () => {
     const { snapshot, checkpoint } = ctx.sessionProjections.restore({
       'test/marks': { ver: 1, seq: 2, val: { marks: ['old', '2'] } },
       'test/count': { ver: 99, seq: 2, val: 3 },
-    }, full, 0)
+    }, full, 0, RESTORE_HEADER)
     expect(snapshot.asOfSeq).toBe(4)
     expect(snapshot.values['test/marks']).toEqual({ marks: ['new'] })
     expect('test/count' in snapshot.values).toBe(false)
@@ -312,7 +317,7 @@ describe('SessionProjectionRegistry drive', () => {
       { type: 'turn/start', seq: 3, time: 3, data: { turn: 2 } },
       { type: 'turn/end', seq: 4, time: 4, data: { turn: 2, reason: { kind: 'completed' } } },
     ]
-    const { snapshot, checkpoint } = ctx.sessionProjections.restore(rows, tail, 3)
+    const { snapshot, checkpoint } = ctx.sessionProjections.restore(rows, tail, 3, RESTORE_HEADER)
     expect(snapshot.asOfSeq).toBe(4)
     // marks already covers the tail (watermark 4): nothing re-applied.
     expect(snapshot.values['test/marks']).toEqual({ marks: ['done'] })
@@ -324,7 +329,7 @@ describe('SessionProjectionRegistry drive', () => {
     const { snapshot: current, checkpoint: currentCheckpoint } = ctx.sessionProjections.restore({
       'test/marks': { ver: 1, seq: 4, val: { marks: ['done'] } },
       'test/count': { ver: 1, seq: 4, val: 5 },
-    }, [], 5)
+    }, [], 5, RESTORE_HEADER)
     expect(current.asOfSeq).toBe(4)
     expect('test/count' in current.values).toBe(false)
     expect(currentCheckpoint['test/count']).toEqual({ ver: 1, seq: 4, val: 5 })
@@ -355,7 +360,7 @@ describe('SessionProjectionRegistry drive', () => {
       'test/marks': { marks: ['stored'] },
     })
 
-    const restored = ctx.sessionProjections.restore(rows, [], 5)
+    const restored = ctx.sessionProjections.restore(rows, [], 5, RESTORE_HEADER)
     expect(restored.snapshot.values).toEqual({
       'test/marks': { marks: ['stored'] },
     })
@@ -370,7 +375,7 @@ describe('SessionProjectionRegistry drive', () => {
     }
 
     expect(ctx.sessionProjections.viewCheckpoint(drifted)).toEqual({})
-    expect(() => ctx.sessionProjections.restore(drifted, [], 3)).toThrow()
+    expect(() => ctx.sessionProjections.restore(drifted, [], 3, RESTORE_HEADER)).toThrow()
   })
 
   it('restore rejects a row claiming events past the supplied log end (shrunk log ⇒ re-read)', async () => {
@@ -383,18 +388,18 @@ describe('SessionProjectionRegistry drive', () => {
     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' } } }
-    const anchored = ctx.sessionProjections.restore(rows, [anchor], 9)
+    const anchored = ctx.sessionProjections.restore(rows, [anchor], 9, RESTORE_HEADER)
     expect(anchored.snapshot.values).toEqual({})
     expect(anchored.checkpoint['test/count']).toEqual({ ver: 1, seq: 9, val: 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/)
+    expect(() => ctx.sessionProjections.restore(rows, [], 9, RESTORE_HEADER)).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 } },
       { type: 'turn/end', seq: 1, time: 1, data: { turn: 1, reason: { kind: 'completed' } } },
     ]
-    const { snapshot, checkpoint } = ctx.sessionProjections.restore(rows, events, 0)
+    const { snapshot, checkpoint } = ctx.sessionProjections.restore(rows, events, 0, RESTORE_HEADER)
     expect(snapshot.asOfSeq).toBe(1)
     expect(snapshot.values).toEqual({})
     expect(checkpoint['test/count']).toEqual({ ver: 1, seq: 1, val: 2 })