feat: add reusable session preparation

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

@@ -7,12 +7,26 @@
 
 import { Context } from 'cordis'
 import {
+  adoptSessionEvent,
   interruptedTurnClosers,
   SESSION_FORMAT_VERSION,
+  SessionPreparation,
   snapshotJsonValue,
   snapshotSessionEvent,
 } from '@deepseek-ai/dsh-session'
 import type { Session, SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
+import type { SessionInspection } from './index.ts'
+import { SessionPreparations } from './preparations.ts'
+import type { SessionPreparationReservation } from './preparations.ts'
+
+/** Default number of detached session preparations retained by a coordinator. */
+export const DEFAULT_PREPARED_SESSION_CACHE_SIZE = 5
+
+/** Coordinator policy supplied by a concrete persistence backend. */
+export interface PersistenceCoordinatorOptions {
+  /** Maximum completed unpublished preparations retained for reuse. */
+  readonly preparedSessionCacheSize: number
+}
 
 /**
  * A stored session's header, valid contiguous event prefix, and optional opaque
@@ -55,7 +69,9 @@ export interface PersistenceBackend<TornMarker = unknown> {
    * `undefined` if no stored artifact exists. Returned metadata must identify
    * `id` before repair or state publication. Used by resume/load, live adoption,
    * and — via `!== undefined` — the create-collision probe. The returned
-   * `tornMarker` is present iff there is a torn tail to truncate.
+   * `tornMarker` is present iff there is a torn tail to truncate. Every header
+   * and event graph must be fresh, mutually unaliased, and unretained by the
+   * backend because preparation freezes and publishes them in place.
    * @param id - persisted session id to resolve.
    * @param signal - optional cancellation for backend read work.
    */
@@ -138,6 +154,16 @@ interface LiveSessionState {
   flush: Promise<void> | undefined
 }
 
+/** One validated cold source and the exact unpublished Session built from it. */
+interface PreparedSessionSource<TornMarker> {
+  readonly inspection: SessionInspection
+  readonly session: Session
+  /** Session length after constructor-owned seed markers were appended. */
+  readonly sessionLength: number
+  readonly tornMarker: TornMarker | undefined
+  readonly closers: readonly SessionEvent[]
+}
+
 /** Collect the rejection reasons from a set of promises (none-throwing). */
 async function settledErrors(promises: Iterable<Promise<unknown>>): Promise<unknown[]> {
   const settled = await Promise.allSettled([...promises])
@@ -443,6 +469,19 @@ function snapshotStoredEvents(events: readonly SessionEvent[], id: SessionId): S
   })
 }
 
+/** Upgrade and validate an exclusively owned backend result without copying it. */
+function adoptStoredEvents(events: SessionEvent[], id: SessionId): SessionEvent[] {
+  assertSupportedEvents(events, id)
+  const messageIds = new Map<number, PersistedMessageId>()
+  for (const [index, event] of events.entries()) {
+    const adopted = adoptSessionEvent(migrateLegacyMessageEvent(event, id, messageIds))
+    events[index] = adopted
+    const messageId = eventMessageId(adopted)
+    if (messageId !== undefined) messageIds.set(adopted.seq, messageId)
+  }
+  return events
+}
+
 /**
  * Owns the backend-agnostic session write-path orchestration. A backend
  * constructs one (`new PersistenceCoordinator(ctx, this)`), implements
@@ -463,15 +502,26 @@ export class PersistenceCoordinator<TornMarker = unknown> {
   private live = new Map<Session, LiveSessionState>()
   /** Exact disposed lifecycles whose eager tail is still draining. */
   private retirements = new Map<SessionId, Promise<void>>()
-  /** Cold loads currently reserving an id across backend reads and repair writes. */
-  private coldLoads = new Set<SessionId>()
+  /** Shared cold reads, unpublished reservations, and completed LRU entries. */
+  private readonly preparations: SessionPreparations<PreparedSessionSource<TornMarker>, SessionState>
   /**
    * Per-session serialization: every operation chains onto the prior one for the
    * same id, so writes for one session never interleave. Keyed by session id.
    */
   private chains = new Map<SessionId, Promise<unknown>>()
 
-  constructor(private ctx: Context, private backend: PersistenceBackend<TornMarker>) {
+  constructor(
+    private ctx: Context,
+    private backend: PersistenceBackend<TornMarker>,
+    options: PersistenceCoordinatorOptions = {
+      preparedSessionCacheSize: DEFAULT_PREPARED_SESSION_CACHE_SIZE,
+    },
+  ) {
+    if (!Number.isSafeInteger(options.preparedSessionCacheSize)
+      || options.preparedSessionCacheSize < 1) {
+      throw new TypeError('preparedSessionCacheSize must be a positive safe integer')
+    }
+    this.preparations = new SessionPreparations(options.preparedSessionCacheSize)
     this.installWritePath()
   }
 
@@ -495,7 +545,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
 
   private async createCore(meta: SessionHeader): Promise<void> {
     // Do NOT clobber an existing session: the SessionId IS the identity.
-    if (this.states.has(meta.id)) {
+    if (this.states.has(meta.id) || this.preparations.has(meta.id)) {
       throw new Error(`session "${meta.id}" already exists in this backend`)
     }
     // A persisted artifact under this id (in ANY scope) blocks creation: load/
@@ -537,8 +587,9 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     // this same backend will refuse to load.
     assertSupportedEvents(events, id)
     if (events.length === 0) return
+    this.preparations.assertWritable(id)
     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)
 
     // Contiguity contract: each event's seq must continue the stored log.
     for (const [i, event] of events.entries()) {
@@ -552,68 +603,93 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     // cursor as soon as it commits (uniform across backends).
     state.materialized = true
     state.cursor += events.length
+    this.preparations.invalidate(id)
   }
 
   /**
-   * Reload a session: its {@link SessionHeader} plus the event log up to the last
-   * durable checkpoint, with any interrupted final turn durably closed (synthetic
-   * boundary events) during load.
-   * @param id - the persisted session to reload.
-   * @returns the header plus the event log, ending on a balanced `turn/end`.
+   * Prepare and reserve the exact unpublished Session used by resume.
+   * @param id - persisted session to prepare.
+   * @param signal - optional cancellation for reading and repair.
+   * @returns an owned preparation released after publication or rollback.
    */
-  async load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
-    await this.retirements.get(id)
-    const selected = await this.serialize(id, async () => {
-      const live = this.ctx.sessions.get(id)
-      if (live !== undefined) return { live }
-      this.coldLoads.add(id)
-      try {
-        return { loaded: await this.loadCore(id) }
-      } finally {
-        this.coldLoads.delete(id)
+  async prepare(id: SessionId, signal?: AbortSignal): Promise<SessionPreparation> {
+    for (;;) {
+      await this.waitForRetirement(id, signal)
+      if (this.ctx.sessions.get(id) !== undefined) {
+        throw new Error(`cannot prepare session "${id}" while it is live`)
       }
-    })
-    return 'loaded' in selected ? selected.loaded : this.loadLiveSnapshot(selected.live)
+      const reservation = await this.preparations.reserve(
+        id,
+        () => this.serialize(id, () => this.prepareCore(id, signal), signal),
+        source => this.serialize(id, () => this.commitPrepared(source), signal),
+        signal,
+      )
+      if (reservation === undefined) continue
+      if (this.ctx.sessions.get(id) !== undefined) {
+        this.preparations.release(reservation, false)
+        throw new Error(`cannot prepare session "${id}" while it is live`)
+      }
+      return SessionPreparation.create(reservation.source.session, {
+        release: () => {
+          this.preparations.release(
+            reservation,
+            reservation.state.owner === undefined
+              && reservation.source.session.events.length === reservation.source.sessionLength,
+          )
+        },
+      })
+    }
   }
 
   /**
-   * Read a detached valid stored prefix without recovery mutations or
-   * coordinator-state publication.
-   * @param id - persisted session to inspect.
-   * @param signal - optional cancellation for queued and backend read work.
-   * @returns stored header and events before any synthetic recovery closers.
+   * Commit recovery and return its immutable logical view without publication.
+   * @param id - persisted session to load.
+   * @returns prepared header and balanced events.
    */
-  inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
-    // Waiting for an in-flight retirement drain must honor cancellation too: a
-    // slow drain would otherwise pin a cancelled inspect until it finishes,
-    // past the documented boundary. serialize() already races the signal for
-    // the queued read; do the same for the retirement wait.
-    const retired = Promise.resolve(this.retirements.get(id))
-    const waited = signal === undefined ? retired : observeQueuedAbort(retired, signal, () => false)
-    return waited.then(() => this.serialize(id, () => this.inspectCore(id, signal), signal))
+  async load(id: SessionId): Promise<SessionInspection> {
+    for (;;) {
+      await this.waitForRetirement(id)
+      const live = this.ctx.sessions.get(id)
+      if (live !== undefined) return this.loadLiveSnapshot(live)
+      const reservation = await this.preparations.reserve(
+        id,
+        () => this.serialize(id, () => this.prepareCore(id)),
+        source => this.serialize(id, () => this.commitPrepared(source)),
+      )
+      if (reservation === undefined) continue
+      const attached = this.ctx.sessions.get(id)
+      if (attached !== undefined) {
+        this.preparations.discard(reservation)
+        return this.loadLiveSnapshot(attached)
+      }
+      this.preparations.discard(reservation)
+      return reservation.source.inspection
+    }
   }
 
-  private async inspectCore(
-    id: SessionId,
-    signal?: AbortSignal,
-  ): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
-    signal?.throwIfAborted()
-    let stored: StoredPrefix<TornMarker> | undefined
+  /**
+   * Inspect a logical session without publishing it or committing recovery.
+   * @param id - persisted session to inspect.
+   * @param signal - optional cancellation for preparation work.
+   * @returns immutable prepared metadata and balanced events.
+   */
+  async inspect(id: SessionId, signal?: AbortSignal): Promise<SessionInspection> {
+    await this.waitForRetirement(id, signal)
+    const live = this.ctx.sessions.get(id)
+    if (live !== undefined) return this.inspectLive(live)
     try {
-      stored = await this.backend.loadStored(id, signal)
+      const source = await this.preparations.inspect(
+        id,
+        () => this.serialize(id, () => this.prepareCore(id, signal), signal),
+        signal,
+      )
+      const attached = this.ctx.sessions.get(id)
+      return attached === undefined ? source.inspection : this.inspectLive(attached)
     } catch (error: unknown) {
-      if (signal?.aborted) signal.throwIfAborted()
+      const attached = this.ctx.sessions.get(id)
+      if (attached !== undefined) return this.inspectLive(attached)
       throw error
     }
-    signal?.throwIfAborted()
-    if (stored === undefined) throw new Error(`session "${id}" not found`)
-    this.assertStoredId(id, stored.meta)
-    this.assertVersion(stored.meta)
-    const events = snapshotStoredEvents(stored.events, id)
-    return {
-      meta: structuredClone(stored.meta),
-      events,
-    }
   }
 
   /**
@@ -660,45 +736,124 @@ export class PersistenceCoordinator<TornMarker = unknown> {
       }
       return { meta: structuredClone(suffix.meta), events: snapshotStoredEvents(suffix.events, id) }
     }
-    const whole = await this.inspectCore(id, signal)
+    const whole = await this.readStoredPrefix(id, signal)
     // Sequential fallback: contiguous seqs from 0 make the suffix an index slice.
     return { meta: whole.meta, events: whole.events.slice(fromSeq) }
   }
 
-  private async loadCore(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
-    const stored = await this.backend.loadStored(id)
+  /** Read one detached physical prefix without logical recovery or caching. */
+  private async readStoredPrefix(
+    id: SessionId,
+    signal?: AbortSignal,
+  ): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
+    signal?.throwIfAborted()
+    const stored = await this.backend.loadStored(id, signal)
+    signal?.throwIfAborted()
+    if (stored === undefined) throw new Error(`session "${id}" not found`)
+    this.assertStoredId(id, stored.meta)
+    this.assertVersion(stored.meta)
+    return {
+      meta: structuredClone(stored.meta),
+      events: snapshotStoredEvents(stored.events, id),
+    }
+  }
+
+  /** Read, repair in memory, validate, and freeze one cold source once. */
+  private async prepareCore(
+    id: SessionId,
+    signal?: AbortSignal,
+  ): Promise<PreparedSessionSource<TornMarker>> {
+    signal?.throwIfAborted()
+    let stored: StoredPrefix<TornMarker> | undefined
+    try {
+      stored = await this.backend.loadStored(id, signal)
+    } catch (error: unknown) {
+      if (signal?.aborted) signal.throwIfAborted()
+      throw error
+    }
+    signal?.throwIfAborted()
     if (stored === undefined) throw new Error(`session "${id}" not found`)
     const { meta, events, tornMarker } = stored
     this.assertStoredId(id, meta)
     this.assertVersion(meta)
-    const storedEvents = snapshotStoredEvents(events, id)
+    const storedEvents = adoptStoredEvents(events, id)
 
     // Preserve complete interrupted events and synthesize only missing closers.
-    const closers = interruptedTurnClosers(storedEvents).map(snapshotSessionEvent)
+    const closers = interruptedTurnClosers(storedEvents).map(adoptSessionEvent)
     const balanced = [...storedEvents, ...closers]
-
-    // Repair storage before publishing coordinator state.
-    if (tornMarker !== undefined || closers.length > 0) {
-      await this.backend.commitRepair(meta, tornMarker, closers)
+    const session = this.ctx.sessions.prepare(id, {
+      seed: balanced,
+      meta,
+      seedSource: 'persistence',
+    })
+    const inspection: SessionInspection = Object.freeze({
+      meta: session.header,
+      events: Object.freeze(balanced),
+    })
+    return {
+      inspection,
+      session,
+      sessionLength: session.events.length,
+      tornMarker,
+      closers,
     }
-    // Keep coordinator metadata detached from the returned record.
-    this.states.set(id, { meta: { ...meta }, cursor: balanced.length, materialized: true })
-    return { meta: structuredClone(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 events = session.events.map(snapshotSessionEvent)
+  /** Commit one prepared repair and establish its ownerless durable cursor. */
+  private async commitPrepared(
+    source: PreparedSessionSource<TornMarker>,
+  ): Promise<{ source: PreparedSessionSource<TornMarker>; state: SessionState }> {
+    const id = source.inspection.meta.id
+    const cursor = source.inspection.events.length
+    const existing = this.states.get(id)
+    if (existing?.owner !== undefined) {
+      throw new Error(`session "${id}" already has a live persistence owner`)
+    }
+    if (source.tornMarker !== undefined || source.closers.length > 0) {
+      await this.backend.commitRepair(source.inspection.meta, source.tornMarker, source.closers)
+    }
+    const state = existing ?? {
+      meta: source.inspection.meta,
+      cursor,
+      materialized: true,
+    }
+    state.meta = source.inspection.meta
+    state.cursor = cursor
+    state.materialized = true
+    this.states.set(id, state)
+    return {
+      source: source.tornMarker === undefined && source.closers.length === 0
+        ? source
+        : { ...source, tornMarker: undefined, closers: [] },
+      state,
+    }
+  }
+
+  /** Return one durable immutable view of an already-live Session. */
+  private async loadLiveSnapshot(session: Session): Promise<SessionInspection> {
+    const events = session.events
     await this.flush(session)
     const state = this.states.get(session.id)
     /* v8 ignore next -- successful flush always publishes this live session's durable state */
     if (state === undefined) throw new Error(`session "${session.id}" lost persistence state during load`)
-    const meta = structuredClone(state.meta)
     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 }
+    return Object.freeze({ meta: state.meta, events })
+  }
+
+  /** Borrow one immutable view from an already-live Session. */
+  private inspectLive(session: Session): SessionInspection {
+    return Object.freeze({ meta: session.header, events: session.events })
+  }
+
+  /** Await one retiring lifecycle with caller cancellation. */
+  private waitForRetirement(id: SessionId, signal?: AbortSignal): Promise<void> {
+    const retired = Promise.resolve(this.retirements.get(id))
+    return signal === undefined
+      ? retired
+      : observeQueuedAbort(retired, signal, () => false)
   }
 
   // Listing is a direct backend read and needs no coordinator state.
@@ -738,13 +893,10 @@ 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
-    // re-entering the chain via the public load() would deadlock.
-    await this.loadCore(id)
-    const state = this.states.get(id)
-    /* v8 ignore next -- loadCore always sets the state for the id */
-    if (!state) throw new Error(`failed to adopt session "${id}"`)
-    return state
+    // This runs inside the id's serialization chain, so it uses core helpers
+    // instead of re-entering through public prepare/load methods.
+    const source = this.preparations.takeReady(id) ?? await this.prepareCore(id)
+    return (await this.commitPrepared(source)).state
   }
 
   private assertVersion(meta: SessionHeader): void {
@@ -795,9 +947,6 @@ export class PersistenceCoordinator<TornMarker = unknown> {
 
     // Capture the header on creation and persist a fork's seed once.
     ctx.on('session/created', (session) => {
-      if (this.coldLoads.has(session.id)) {
-        throw new Error(`cannot publish session "${session.id}" while its persisted history is loading`)
-      }
       void this.initFor(session)
     })
 
@@ -847,6 +996,12 @@ export class PersistenceCoordinator<TornMarker = unknown> {
   private initFor(session: Session): LiveSessionState {
     const existing = this.live.get(session)
     if (existing) return existing
+    const reservation = this.preparations.reservationFor(session)
+    if (reservation !== undefined) {
+      const restored = this.attachPrepared(session, reservation)
+      this.live.set(session, restored)
+      return restored
+    }
     const seed = session.events.map(e => structuredClone(e))
     const live: LiveSessionState = { pending: [], init: Promise.resolve(), flush: undefined }
     this.live.set(session, live)
@@ -855,6 +1010,28 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     return live
   }
 
+  /** Bind one exact prepared Session and persist only its unpublished suffix. */
+  private attachPrepared(
+    session: Session,
+    reservation: SessionPreparationReservation<PreparedSessionSource<TornMarker>, SessionState>,
+  ): LiveSessionState {
+    const { source, state } = reservation
+    if (source.session !== session || state.owner !== undefined
+      || state.cursor !== source.inspection.events.length
+      || session.firstLiveSeq !== state.cursor) {
+      throw new Error(`session "${session.id}" preparation no longer matches its persistence state`)
+    }
+    const suffix = session.events.slice(state.cursor).map(event => structuredClone(event))
+    this.preparations.attach(reservation)
+    state.owner = session
+    const live: LiveSessionState = { pending: [], init: Promise.resolve(), flush: undefined }
+    if (suffix.length > 0) {
+      live.init = this.serialize(session.id, () => this.appendCore(session.id, suffix))
+      live.init.catch(() => { /* observed by flush/dispose through the controller */ })
+    }
+    return live
+  }
+
   /**
    * Whether a live session's `seed` reproduces the first `cursor` persisted
    * events. A `cursor` of 0 (nothing persisted yet) trivially matches. Used when
@@ -922,7 +1099,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
     // cwd mismatch before repair or state publication.
     const live = await this.backend.loadStored(id)
     if (live !== undefined) {
-      // Do NOT route through loadCore(): that crash-repairs open turns as
+      // Do NOT route through cold preparation: 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.adoptLivePrefix(session, seed, live)

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

@@ -6,6 +6,7 @@
  */
 
 import { Context, Service } from 'cordis'
+import { SessionPreparation } from '@deepseek-ai/dsh-session'
 import type { SessionEvent, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
 import type { SessionPersistenceRevision } from './revision.ts'
 
@@ -21,9 +22,22 @@ export interface SessionPersistenceSnapshot {
   revision: SessionPersistenceRevision
 }
 
+/** Immutable logical session prepared from persistence or a live owner. */
+export interface SessionInspection {
+  /** Validated immutable session metadata. */
+  readonly meta: SessionHeader
+  /** Validated contiguous logical event log. */
+  readonly events: readonly SessionEvent[]
+}
+
 // The backend-agnostic write-path orchestration first-party backends compose.
-export { PersistenceCoordinator } from './coordinator.ts'
-export type { PersistenceBackend, StoredPrefix, StoredSuffix } from './coordinator.ts'
+export { DEFAULT_PREPARED_SESSION_CACHE_SIZE, PersistenceCoordinator } from './coordinator.ts'
+export type {
+  PersistenceBackend,
+  PersistenceCoordinatorOptions,
+  StoredPrefix,
+  StoredSuffix,
+} from './coordinator.ts'
 
 declare module 'cordis' {
   interface Context {
@@ -83,46 +97,64 @@ export abstract class SessionPersistence extends Service {
   abstract append(id: SessionId, events: readonly SessionEvent[]): Promise<void>
 
   /**
-   * 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. Implementations
-   * MUST NOT crash-repair an identity still bound to a live Session: a balanced
-   * live log may return with its stored header as a durable snapshot, while an
-   * open live turn rejects.
-   * A coordinator-backed cold load reserves the identity across storage awaits,
-   * so concurrent publication of a same-id live Session rejects.
-   * Returned events are detached, and every identified message is deeply
-   * frozen. Coordinator-backed implementations upgrade supported pre-identity
-   * message events before validation; other malformed messages reject before
-   * any stored event is returned.
+   * Prepare the exact unpublished Session used by resume. Implementations may
+   * reuse object graphs retained by an earlier {@link inspect}; disposal
+   * releases an unpublished reservation.
+   * @param id - persisted session to prepare.
+   * @param signal - optional cancellation for preparation work.
+   * @returns one owned unpublished Session preparation.
+   */
+  async prepare(id: SessionId, signal?: AbortSignal): Promise<SessionPreparation> {
+    signal?.throwIfAborted()
+    const loaded = await this.load(id)
+    signal?.throwIfAborted()
+    const sessions = this.ctx.get('sessions')
+    if (sessions === undefined) {
+      throw new Error('cannot prepare a session: SessionStore is not configured')
+    }
+    return SessionPreparation.create(sessions.prepare(id, {
+      seed: loaded.events.map(event => structuredClone(event)),
+      meta: structuredClone(loaded.meta),
+      seedSource: 'persistence',
+    }))
+  }
+
+  /**
+   * Load an immutable balanced logical view and commit any required cold
+   * recovery. 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. 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. Returned values may be
+   * shared with immutable live or prepared state and must not be mutated.
    * @param id - the persisted session to reload.
    * @returns the header and a log ending on a balanced `turn/end`.
    */
-  abstract load(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }>
+  abstract load(id: SessionId): Promise<SessionInspection>
 
   /**
-   * Inspect a header and its valid contiguous stored prefix without repairing
-   * a torn tail, closing an interrupted turn, or publishing coordinator state.
-   * This read is serialized with writes for the same id and returns detached
-   * values with upgraded, deeply frozen identified messages, so observers
-   * cannot mutate message identity/content or backend-owned state. Other
-   * malformed messages reject.
+   * Inspect an immutable balanced logical session without committing recovery
+   * or publishing it. A complete interrupted turn receives synthetic closers
+   * in memory and a torn physical tail remains untouched. Coordinator-backed
+   * implementations retain the exact unpublished Session for bounded reuse by
+   * a later {@link prepare}; callers borrow only its immutable header and log.
    * @param id - the persisted session to inspect.
    * @param signal - optional cancellation for queued and backend read work.
-   * @returns the header and valid stored event prefix exactly as observed.
+   * @returns the validated header and balanced logical event log.
    */
-  abstract inspect(id: SessionId, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }>
+  abstract inspect(id: SessionId, signal?: AbortSignal): Promise<SessionInspection>
 
   /**
    * Read the stored events from `fromSeq` onward — the read-from-seq
    * primitive for read models that resume from a watermark (e.g. a persisted
-   * projection cache folding only the tail past its checkpoint). Like
-   * {@link inspect} it is non-mutating and detached: no torn-tail truncation,
-   * no synthetic closers, no coordinator-state publication; only events from
-   * the valid contiguous stored prefix are returned, so a torn fragment never
-   * reaches the caller. `fromSeq` at or beyond the stored prefix returns an
-   * empty event list (never an error). Backends whose medium can seek by seq
+   * projection cache folding only the tail past its checkpoint). Unlike
+   * {@link inspect}, it is a detached physical suffix read: no preparation
+   * cache, torn-tail truncation, synthetic closers, or coordinator-state
+   * publication. Only events from the valid contiguous stored prefix are
+   * returned, so a torn fragment never reaches the caller. `fromSeq` at or
+   * beyond the stored prefix returns an empty event list (never an error).
+   * Backends whose medium can seek by seq
    * (SQLite) read only the suffix; sequential media (JSONL, both encodings)
    * still parse the whole artifact and skip forward — the primitive bounds
    * what is RETURNED and refolded, not every backend's physical read.

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

@@ -0,0 +1,308 @@
+/**
+ * Bounded sharing and exclusive reservation of unpublished Sessions.
+ * @module @deepseek-ai/dsh-session-persistence/preparations
+ */
+
+import type { Session, SessionId } from '@deepseek-ai/dsh-session'
+
+interface PreparedSource {
+  readonly session: Session
+}
+
+type PreparationPhase = 'loading' | 'ready' | 'committing' | 'reserved'
+
+interface PreparationEntry<Source, CommitState> {
+  readonly id: SessionId
+  readonly result: Promise<Source>
+  phase: PreparationPhase
+  source?: Source
+  reservation?: SessionPreparationReservation<Source, CommitState>
+  reservationSettled?: Promise<void>
+  settleReservation?: () => void
+}
+
+/** One exclusively held prepared source and its committed persistence state. */
+export interface SessionPreparationReservation<Source, CommitState> {
+  readonly entry: PreparationEntry<Source, CommitState>
+  readonly source: Source
+  readonly state: CommitState
+}
+
+/** Per-coordinator cold-read sharing, exclusive reservation, and ready-entry LRU. */
+export class SessionPreparations<Source extends PreparedSource, CommitState> {
+  private readonly entries = new Map<SessionId, PreparationEntry<Source, CommitState>>()
+
+  constructor(private readonly capacity: number) {}
+
+  /**
+   * Whether this pool currently knows about an unpublished identity.
+   * @param id - session identity.
+   * @returns whether an entry exists for the identity.
+   */
+  has(id: SessionId): boolean {
+    return this.entries.has(id)
+  }
+
+  /**
+   * Observe one prepared source, sharing an in-flight read for the same id.
+   * @param id - session identity.
+   * @param load - cold loader used when no entry exists.
+   * @param signal - optional cancellation signal while waiting.
+   * @returns the shared prepared source.
+   */
+  async inspect(
+    id: SessionId,
+    load: () => Promise<Source>,
+    signal?: AbortSignal,
+  ): Promise<Source> {
+    const { entry, created } = this.entryFor(id, load)
+    const loaded = signal === undefined || created
+      ? await entry.result
+      : await observeQueuedAbort(entry.result, signal)
+    const source = entry.source ?? loaded
+    if (this.entries.get(id) === entry && entry.phase === 'ready') this.touch(entry)
+    return source
+  }
+
+  /**
+   * Reserve one ready source after committing its pending durable repair.
+   * @param id - session identity.
+   * @param load - cold loader used when no entry exists.
+   * @param commit - durable repair and cursor-state commit.
+   * @param signal - optional cancellation signal while waiting.
+   * @returns the exclusive reservation, or undefined if its entry was invalidated.
+   */
+  async reserve(
+    id: SessionId,
+    load: () => Promise<Source>,
+    commit: (source: Source) => Promise<{ source: Source; state: CommitState }>,
+    signal?: AbortSignal,
+  ): Promise<SessionPreparationReservation<Source, CommitState> | undefined> {
+    const { entry, created } = this.entryFor(id, load)
+    const loaded = signal === undefined || created
+      ? await entry.result
+      : await observeQueuedAbort(entry.result, signal)
+    while (this.entries.get(id) === entry && entry.phase !== 'ready') {
+      const settled = entry.reservationSettled
+      /* v8 ignore next -- committing/reserved transitions install this waiter synchronously. */
+      if (settled === undefined) throw new Error(`session "${id}" preparation lost its reservation waiter`)
+      if (signal === undefined) await settled
+      else await observeQueuedAbort(settled, signal)
+    }
+    if (this.entries.get(id) !== entry) return undefined
+    const source = entry.source ?? loaded
+    const reservationSettled = Promise.withResolvers<void>()
+    entry.phase = 'committing'
+    entry.reservationSettled = reservationSettled.promise
+    entry.settleReservation = reservationSettled.resolve
+    let committed: { source: Source; state: CommitState }
+    try {
+      committed = await commit(source)
+    } catch (error: unknown) {
+      this.remove(entry)
+      throw error
+    }
+    entry.source = committed.source
+    try {
+      signal?.throwIfAborted()
+    } catch (error: unknown) {
+      this.makeReady(entry)
+      throw error
+    }
+    const reservation: SessionPreparationReservation<Source, CommitState> = {
+      entry,
+      source: committed.source,
+      state: committed.state,
+    }
+    entry.phase = 'reserved'
+    entry.reservation = reservation
+    return reservation
+  }
+
+  /**
+   * Return the exact reservation for Session publication, rejecting aliases.
+   * @param session - exact Session candidate for publication.
+   * @returns its reservation, or undefined when no preparation exists.
+   */
+  reservationFor(session: Session): SessionPreparationReservation<Source, CommitState> | undefined {
+    const entry = this.entries.get(session.id)
+    if (entry === undefined) return undefined
+    if (entry.phase === 'reserved'
+      && entry.source?.session === session
+      && entry.reservation !== undefined) {
+      return entry.reservation
+    }
+    throw new Error(`cannot publish session "${session.id}" while a persisted preparation exists`)
+  }
+
+  /**
+   * Consume a reservation after its exact Session has attached.
+   * @param reservation - reservation to consume.
+   */
+  attach(reservation: SessionPreparationReservation<Source, CommitState>): void {
+    const { entry } = reservation
+    if (this.entries.get(entry.id) !== entry || entry.reservation !== reservation) {
+      throw new Error(`session "${entry.id}" preparation is no longer reserved`)
+    }
+    this.remove(entry)
+  }
+
+  /**
+   * Consume a reservation whose caller only needs the committed inspection.
+   * @param reservation - reservation to consume.
+   */
+  discard(reservation: SessionPreparationReservation<Source, CommitState>): void {
+    const { entry } = reservation
+    if (this.entries.get(entry.id) !== entry || entry.reservation !== reservation) return
+    this.remove(entry)
+  }
+
+  /**
+   * Return a reusable unpublished reservation to the ready LRU.
+   * @param reservation - reservation to release.
+   * @param reusable - whether the source remains valid for reuse.
+   */
+  release(
+    reservation: SessionPreparationReservation<Source, CommitState>,
+    reusable: boolean,
+  ): void {
+    const { entry } = reservation
+    if (this.entries.get(entry.id) !== entry
+      || entry.reservation !== reservation
+      || entry.phase !== 'reserved') return
+    if (!reusable) {
+      this.remove(entry)
+      return
+    }
+    delete entry.reservation
+    this.makeReady(entry)
+  }
+
+  /**
+   * Discard a prepared view after the durable log changes.
+   * @param id - changed session identity.
+   */
+  invalidate(id: SessionId): void {
+    const entry = this.entries.get(id)
+    if (entry !== undefined) this.remove(entry)
+  }
+
+  /**
+   * Reject writes while an unpublished Session exclusively reserves the id.
+   * @param id - session identity to check.
+   */
+  assertWritable(id: SessionId): void {
+    const phase = this.entries.get(id)?.phase
+    if (phase === 'committing' || phase === 'reserved') {
+      throw new Error(`cannot append session "${id}" while its persisted preparation is reserved`)
+    }
+  }
+
+  /**
+   * Remove a completed entry for an already-serialized append adoption.
+   * @param id - adopted session identity.
+   * @returns the prepared source, or undefined when no entry exists.
+   */
+  takeReady(id: SessionId): Source | undefined {
+    const entry = this.entries.get(id)
+    if (entry === undefined) return undefined
+    if (entry.phase !== 'ready' || entry.source === undefined) {
+      throw new Error(`cannot adopt session "${id}" while its preparation is pending`)
+    }
+    this.remove(entry)
+    return entry.source
+  }
+
+  private entryFor(
+    id: SessionId,
+    load: () => Promise<Source>,
+  ): { entry: PreparationEntry<Source, CommitState>; created: boolean } {
+    const existing = this.entries.get(id)
+    if (existing !== undefined) return { entry: existing, created: false }
+    const result = Promise.resolve().then(load)
+    const entry: PreparationEntry<Source, CommitState> = { id, result, phase: 'loading' }
+    this.entries.set(id, entry)
+    void result.then((source) => {
+      if (this.entries.get(id) !== entry) return
+      entry.source = source
+      entry.phase = 'ready'
+    }, () => {
+      this.remove(entry)
+    })
+    return { entry, created: true }
+  }
+
+  private makeReady(entry: PreparationEntry<Source, CommitState>): void {
+    if (this.entries.get(entry.id) !== entry) return
+    entry.phase = 'ready'
+    const settle = entry.settleReservation
+    delete entry.reservationSettled
+    delete entry.settleReservation
+    settle?.()
+    this.touch(entry)
+  }
+
+  private remove(entry: PreparationEntry<Source, CommitState>): void {
+    if (this.entries.get(entry.id) !== entry) return
+    this.entries.delete(entry.id)
+    const settle = entry.settleReservation
+    delete entry.reservationSettled
+    delete entry.settleReservation
+    settle?.()
+  }
+
+  private touch(entry: PreparationEntry<Source, CommitState>): void {
+    if (this.entries.get(entry.id) !== entry || entry.phase !== 'ready') return
+    this.entries.delete(entry.id)
+    this.entries.set(entry.id, entry)
+    let readyCount = 0
+    for (const candidate of this.entries.values()) {
+      if (candidate.phase === 'ready') readyCount += 1
+    }
+    if (readyCount <= this.capacity) return
+    for (const [id, candidate] of this.entries) {
+      if (candidate.phase !== 'ready') continue
+      this.entries.delete(id)
+      readyCount -= 1
+      if (readyCount <= this.capacity) break
+    }
+  }
+}
+
+/** Give a queued observer a prompt cancellation view without cancelling shared work. */
+function observeQueuedAbort<T>(operation: Promise<T>, signal: AbortSignal): Promise<T> {
+  return new Promise<T>((resolve, reject) => {
+    let settled = false
+    const finish = (callback: () => void): void => {
+      if (settled) return
+      settled = true
+      signal.removeEventListener('abort', onAbort)
+      callback()
+    }
+    const onAbort = (): void => {
+      finish(() => {
+        try {
+          signal.throwIfAborted()
+        } catch (reason: unknown) {
+          rejectPreparationObservation(reject, reason)
+          return
+        }
+        /* v8 ignore next -- a native AbortSignal emits abort only after becoming aborted. */
+        reject(new Error('preparation observation abort event lacked an aborted signal'))
+      })
+    }
+    signal.addEventListener('abort', onAbort, { once: true })
+    operation.then(
+      (value) => { finish(() => { resolve(value) }) },
+      (reason: unknown) => {
+        finish(() => { rejectPreparationObservation(reject, reason) })
+      },
+    )
+    if (signal.aborted) onAbort()
+  })
+}
+
+/** Preserve an exact loader or AbortSignal reason, including legacy non-Error values. */
+function rejectPreparationObservation(reject: (reason?: unknown) => void, reason: unknown): void {
+  reject(reason)
+}