perf(session-persistence): prepare historical reads before publication

benchmarks/session-open/session-open.bench.ts

@@ -173,7 +173,7 @@ class SessionOpenBenchmarkSuite {
     this.legacySourcePath = this.facts.path
     // Produce one real post-upgrade directory outside every measured interval.
     const templateRoot = await this.createRoot('first-open', 'post-upgrade-template')
-    requireReport(await runWorker(templateRoot, 'phase-migrate'), 'phase-migrate')
+    requireReport(await runWorker(templateRoot, 'agent-resume'), 'agent-resume')
     this.currentSourcePath = join(
       templateRoot,
       SYNTHETIC_SESSION_DIRECTORY,

packages/session/session-persistence-jsonl/src/generation.ts

@@ -22,8 +22,11 @@ import { basename, dirname, join } from 'node:path'
 import { performance } from 'node:perf_hooks'
 import { pipeline, Readable } from 'node:stream'
 import { scheduler } from 'node:timers/promises'
+import { isDeepStrictEqual } from 'node:util'
 import { constants, createZstdCompress } from 'node:zlib'
 import { Session } from '@deepseek-ai/dsh-session'
+import type { SessionEvent } from '@deepseek-ai/dsh-session'
+import { BlockAssembler, expandAssistantStream } from '@deepseek-ai/dsh-llm'
 import type {
   SessionFormatArtifact,
   SessionFormatJsonValue,
@@ -62,8 +65,8 @@ export interface JsonlGenerationFormatAdapter {
   isUnsupportedMigrationError?(error: unknown): error is Error
 }
 
-/** Inputs for ensuring one already-resolved generation has a current successor. */
-export interface EnsureJsonlGenerationOptions {
+/** Inputs for preparing one historical generation and publishing its current successor later. */
+export interface PrepareJsonlMigrationOptions {
   /** Immutable generation selected by the backend resolver. */
   readonly sourcePath: string
   /** Version selected from the source filename and independently checked against its header. */
@@ -101,36 +104,24 @@ export interface JsonlExpectedPrefix {
   readonly digest: string
 }
 
-/** Result of current classification or exclusive publication. */
-export type EnsureJsonlGenerationResult =
-  | {
-    readonly status: 'current'
-    readonly version: number
-    readonly path: string
-    readonly snapshot: JsonlPhysicalSnapshot
-  }
-  | {
-    readonly status: 'migrated'
-    readonly fromVersion: number
-    readonly toVersion: number
-    readonly path: string
-    readonly sourcePath: string
-    readonly snapshot: JsonlPhysicalSnapshot
-  }
+/** A historical source changed after its single decode and migration pass. */
+export class JsonlGenerationSourceChangedError extends Error {
+  override readonly name = 'JsonlGenerationSourceChangedError'
 
-/** A future physical header was readable, but this writer cannot interpret it. */
-export class JsonlGenerationNewerVersionError extends Error {
-  override readonly name = 'JsonlGenerationNewerVersionError'
-
-  constructor(
-    readonly storedVersion: number,
-    readonly currentVersion: number,
-    readonly storedId: string,
-  ) {
-    super(`session log format v${storedVersion} is newer than current v${currentVersion}`)
+  /** @param path - historical generation whose revision changed. */
+  constructor(readonly path: string) {
+    super(`historical session generation changed during migration: "${path}"`)
   }
 }
 
+/** Current logical state prepared independently from durable publication. */
+export interface PreparedJsonlMigration {
+  readonly sourceIdentity: JsonlPhysicalIdentity
+  readonly artifact: SessionFormatArtifact
+  /** Encode, verify, and exclusively publish once; every call shares the same success or failure. */
+  publish(): Promise<JsonlPhysicalIdentity>
+}
+
 /** A historical artifact is intact, but the format edge refuses its contents. */
 export class JsonlGenerationUnsupportedMigrationError extends Error {
   override readonly name = 'JsonlGenerationUnsupportedMigrationError'
@@ -172,23 +163,12 @@ export interface JsonlPhysicalIdentity {
   readonly ctimeNs: bigint
 }
 
-/** One revision-stable physical artifact returned to the immediate backend decoder. */
-export interface JsonlPhysicalSnapshot extends StablePhysicalFile {
-  readonly headerValue: Record<string, unknown>
-  readonly headerRecord: Buffer
-}
-
 /** Exact bytes of one stable file revision together with the stat identity that proved it stable. */
 export interface StablePhysicalFile {
   readonly bytes: Buffer
   readonly identity: JsonlPhysicalIdentity
 }
 
-interface JsonlPhysicalHeader {
-  readonly value: Record<string, unknown>
-  readonly record: Buffer
-}
-
 interface GenerationFileSystem {
   open(path: string, flags: string, mode?: number): Promise<FileHandle>
   readFile(path: string, signal?: AbortSignal): Promise<Buffer>
@@ -219,7 +199,7 @@ export type JsonlGenerationRuntimeOverrides = Partial<Omit<JsonlGenerationIntern
 /** Bound generation operations used by production defaults and deterministic tests. */
 export interface JsonlGenerationRuntime {
   readStable(path: string, signal?: AbortSignal): Promise<StablePhysicalFile>
-  ensure(options: EnsureJsonlGenerationOptions): Promise<EnsureJsonlGenerationResult>
+  prepare(options: PrepareJsonlMigrationOptions): Promise<PreparedJsonlMigration>
   verify(
     path: string,
     compression: JsonlCompression,
@@ -260,10 +240,6 @@ function identity(value: JsonlPhysicalIdentity): string {
   return [value.dev, value.ino, value.size, value.mtimeNs, value.ctimeNs].join(':')
 }
 
-function fingerprint(value: JsonlPhysicalIdentity, bytes: Buffer): string {
-  return `${identity(value)}:${createHash('sha256').update(bytes).digest('hex')}`
-}
-
 /**
  * Read one stable revision of a JSONL file with a single retry. If an append
  * overlaps both reads, return the second read's committed pre-read prefix
@@ -313,10 +289,6 @@ function storedVersion(header: unknown): number {
   return version as number
 }
 
-function storedId(header: unknown): string {
-  return String((header as { id?: unknown }).id)
-}
-
 function parseJson(text: string, subject: string): unknown {
   try {
     return JSON.parse(text)
@@ -398,10 +370,15 @@ interface StartedMigrationStream {
 
 async function startMigrationStream(
   headerRecord: Buffer,
+  sourceVersion: number,
   format: JsonlGenerationFormatAdapter,
-  validateHistoricalHeader?: EnsureJsonlGenerationOptions['validateHistoricalHeader'],
+  validateHistoricalHeader?: PrepareJsonlMigrationOptions['validateHistoricalHeader'],
 ): Promise<StartedMigrationStream> {
   const value = parseJson(headerRecord.subarray(0, -1).toString('utf8'), 'header line')
+  const version = storedVersion(value)
+  if (version !== sourceVersion) {
+    throw new Error(`resolved JSONL source filename identifies v${sourceVersion}, but its header identifies v${version}`)
+  }
   const header = value as Record<string, unknown>
   const validation = validateHistoricalHeader?.(header)
   if (validation !== undefined) await validation
@@ -430,17 +407,18 @@ async function consumeMigrationBytes(
 async function decodeStreamingMigration(
   bytes: Buffer,
   compression: JsonlCompression,
+  sourceVersion: number,
   format: JsonlGenerationFormatAdapter,
-  validateHistoricalHeader: EnsureJsonlGenerationOptions['validateHistoricalHeader'],
+  validateHistoricalHeader: PrepareJsonlMigrationOptions['validateHistoricalHeader'],
   signal?: AbortSignal,
 ): Promise<SessionFormatArtifact> {
   signal?.throwIfAborted()
   if (compression === 'none') {
     const headerEnd = bytes.indexOf(0x0A)
-    /* v8 ignore next -- ensureCurrent's physical-header preflight already requires this newline. */
     if (headerEnd === -1) throw new Error('empty or header-less session log')
     const stream = await startMigrationStream(
       bytes.subarray(0, headerEnd + 1),
+      sourceVersion,
       format,
       validateHistoricalHeader,
     )
@@ -457,7 +435,6 @@ async function decodeStreamingMigration(
   }
 
   const { frames, tornStart } = scanZstdFrames(bytes)
-  /* v8 ignore next -- ensureCurrent's physical-header preflight already requires a complete header frame. */
   if (frames.length === 0) throw new Error('empty or header-less Zstandard session log')
   const decoder = createZstdFrameDecoder()
   try {
@@ -468,6 +445,7 @@ async function decodeStreamingMigration(
     assertIndependentHeaderFrame(first.value)
     const stream = await startMigrationStream(
       first.value,
+      sourceVersion,
       format,
       validateHistoricalHeader,
     )
@@ -557,7 +535,9 @@ async function verifyCurrentGeneration(
     generation.events,
     generation.meta,
     generation.inheritedEventCount,
+    'detached',
   )
+  assertCurrentAssistantStreams(generation.events)
   return {
     identity: snapshot.identity,
     bytes: snapshot.bytes.length,
@@ -565,6 +545,32 @@ async function verifyCurrentGeneration(
   }
 }
 
+/** Fully replay embedded streams only inside isolated current-generation verification. */
+function assertCurrentAssistantStreams(events: readonly SessionEvent[]): void {
+  for (const [index, event] of events.entries()) {
+    if (event.type !== 'assistant/message' && event.type !== 'assistant/attempt') continue
+    const assembler = new BlockAssembler()
+    let timed: ReturnType<typeof expandAssistantStream>
+    try {
+      timed = expandAssistantStream(event.data.stream)
+      for (const member of timed) assembler.push(member.chunk)
+    } catch (error: unknown) {
+      throw new Error(`seed ${event.type} at index ${index} has an invalid embedded stream`, { cause: error })
+    }
+    if (event.type === 'assistant/attempt' || timed.length === 0) continue
+    const content = event.data.interrupted === true ? assembler.interruptedBlocks() : assembler.blocks()
+    if (!isDeepStrictEqual(event.data.message.content, content)) {
+      throw new Error(`seed assistant/message at index ${index} content disagrees with its embedded stream`)
+    }
+    if (!isDeepStrictEqual(event.data.usage, assembler.usage)) {
+      throw new Error(`seed assistant/message at index ${index} usage disagrees with its embedded stream`)
+    }
+    if (!isDeepStrictEqual(event.data.message.source.replayState, assembler.replayState)) {
+      throw new Error(`seed assistant/message at index ${index} replay state disagrees with its embedded stream`)
+    }
+  }
+}
+
 function decodeCurrentGeneration(
   bytes: Buffer,
   compression: JsonlCompression,
@@ -620,45 +626,6 @@ function assertIndependentHeaderFrame(plaintext: Buffer): void {
   }
 }
 
-function readRawHeader(bytes: Buffer): JsonlPhysicalHeader {
-  const newline = bytes.indexOf(0x0A)
-  if (newline === -1) throw new Error('empty or header-less session log')
-  const record = bytes.subarray(0, newline + 1)
-  const value = parseJson(record.subarray(0, -1).toString('utf8'), 'header line')
-  storedVersion(value)
-  return { value: value as Record<string, unknown>, record }
-}
-
-function readZstdHeader(bytes: Buffer, signal?: AbortSignal): JsonlPhysicalHeader {
-  signal?.throwIfAborted()
-  const first = scanZstdFrames(bytes, 1).frames[0]
-  if (first === undefined) throw new Error('empty or header-less Zstandard session log')
-  const decoder = createZstdFrameDecoder()
-  const decodedFrames = decoder.decode(bytes, [first])
-  try {
-    const decoded = decodedFrames.next()
-    /* v8 ignore next -- one complete frame yields once or the decoder throws. */
-    if (decoded.done) throw new Error('empty or header-less Zstandard session log')
-    signal?.throwIfAborted()
-    assertIndependentHeaderFrame(decoded.value)
-    const record = Buffer.from(decoded.value)
-    const value = parseJson(record.subarray(0, -1).toString('utf8'), 'header line')
-    storedVersion(value)
-    return { value: value as Record<string, unknown>, record }
-  } finally {
-    decodedFrames.return()
-    decoder.close()
-  }
-}
-
-function readPhysicalHeader(
-  bytes: Buffer,
-  compression: JsonlCompression,
-  signal: AbortSignal | undefined,
-): JsonlPhysicalHeader {
-  return compression === 'zstd' ? readZstdHeader(bytes, signal) : readRawHeader(bytes)
-}
-
 function assertGenerationPaths(
   sourcePath: string,
   sourceVersion: number,
@@ -823,7 +790,6 @@ async function removeTemporary(
   try {
     await internals.fs.rm(path)
   } catch (cleanupFailure: unknown) {
-    if (primaryFailure === undefined) throw cleanupFailure
     throw new AggregateError(
       [primaryFailure, cleanupFailure],
       `failed to clean migration temporary "${path}" after an earlier failure`,
@@ -879,19 +845,16 @@ function asError(error: unknown): Error {
 
 async function inspectExpectedCurrent<T>(
   currentPath: string,
-  checkCanonicalTargetName: boolean,
   internals: JsonlGenerationInternals,
   inspect: () => Promise<T>,
 ): Promise<T> {
   try {
-    if (checkCanonicalTargetName) {
-      const expectedName = basename(currentPath)
-      const names = await internals.fs.readdir(dirname(currentPath))
-      if (!names.includes(expectedName)) {
-        const noncanonical = names.find(name => name.toLowerCase() === expectedName.toLowerCase())
-        if (noncanonical !== undefined) {
-          throw new Error(`target resolves to noncanonical directory entry "${noncanonical}"`)
-        }
+    const expectedName = basename(currentPath)
+    const names = await internals.fs.readdir(dirname(currentPath))
+    if (!names.includes(expectedName)) {
+      const noncanonical = names.find(name => name.toLowerCase() === expectedName.toLowerCase())
+      if (noncanonical !== undefined) {
+        throw new Error(`target resolves to noncanonical directory entry "${noncanonical}"`)
       }
     }
     const info = await internals.fs.lstat(currentPath)
@@ -913,39 +876,71 @@ function withOverrides(overrides: JsonlGenerationRuntimeOverrides): JsonlGenerat
   }
 }
 
-async function reopenExpectedCurrent(
-  currentPath: string,
-  staged: StreamedMigrationStage,
-  compression: JsonlCompression,
-  expectedId: string,
-  expectedEventCount: number,
-  verifyCurrentFile: EnsureJsonlGenerationOptions['verifyCurrentFile'],
-  signal: AbortSignal | undefined,
-  checkCanonicalTargetName: boolean,
+async function publishPreparedMigration(
+  options: PrepareJsonlMigrationOptions,
+  suffix: string,
+  artifact: SessionFormatArtifact,
+  sourceIdentity: JsonlPhysicalIdentity,
   internals: JsonlGenerationInternals,
-): Promise<JsonlPhysicalSnapshot> {
-  return inspectExpectedCurrent(currentPath, checkCanonicalTargetName, internals, async () => {
-    const verified = await verifyCurrentFile(
-      currentPath,
+): Promise<JsonlPhysicalIdentity> {
+  await scheduler.yield()
+  const { sourcePath, currentPath, compression, verifyCurrentFile } = options
+  const eventCount = artifact.events.length
+  let staged = await writeSyncedTemp(currentPath, suffix, compression, artifact, options.format, undefined, internals)
+  try {
+    const verifiedStage = await verifyCurrentFile(
+      staged.path,
       compression,
-      expectedId,
-      expectedEventCount,
-      staged,
-      signal,
+      artifact.header.id,
+      eventCount,
     )
-    if (verified.bytes !== staged.bytes || verified.digest !== staged.digest) {
-      throw new Error('target bytes differ from the migrated generation')
+    if (verifiedStage.bytes !== staged.bytes || verifiedStage.digest !== staged.digest) {
+      throw new Error('staged session generation changed during verification')
     }
-    const snapshot = await readStableSnapshot(currentPath, signal, internals.fs)
-    const header = readPhysicalHeader(snapshot.bytes, compression, signal)
-    return { ...snapshot, headerValue: header.value, headerRecord: header.record }
-  })
+    await internals.barrier('before-source-check', 1)
+    const beforePublish = await internals.fs.stat(sourcePath)
+    if (identity(beforePublish) !== identity(sourceIdentity)) {
+      throw new JsonlGenerationSourceChangedError(sourcePath)
+    }
+    const published = await publishCurrentExclusive(staged.path, currentPath, internals)
+    if (published && internals.platform === 'win32') staged = { ...staged, path: '' }
+    await internals.barrier('after-publication', 1)
+    let currentIdentity: JsonlPhysicalIdentity
+    if (published) {
+      if (staged.path !== '') {
+        await removeCommittedTemporary(staged.path, internals)
+        staged = { ...staged, path: '' }
+      }
+      currentIdentity = await internals.fs.stat(currentPath)
+    } else {
+      const winner = await inspectExpectedCurrent(currentPath, internals, async () => {
+        const candidate = await verifyCurrentFile(
+          currentPath,
+          compression,
+          artifact.header.id,
+          eventCount,
+          staged,
+        )
+        if (candidate.bytes !== staged.bytes || candidate.digest !== staged.digest) {
+          throw new Error('target bytes differ from the migrated generation')
+        }
+        return candidate
+      })
+      currentIdentity = winner.identity
+      await removeCommittedTemporary(staged.path, internals)
+      staged = { ...staged, path: '' }
+    }
+    return currentIdentity
+  } catch (error: unknown) {
+    if (staged.path !== '') await removeTemporary(staged.path, error, internals)
+    throw error
+  }
 }
 
-async function ensureCurrent(
-  options: EnsureJsonlGenerationOptions,
+async function prepareMigration(
+  options: PrepareJsonlMigrationOptions,
   internals: JsonlGenerationInternals,
-): Promise<EnsureJsonlGenerationResult> {
+): Promise<PreparedJsonlMigration> {
   const { sourcePath, sourceVersion, currentPath, compression, format, signal } = options
   const suffix = assertGenerationPaths(
     sourcePath,
@@ -954,124 +949,58 @@ async function ensureCurrent(
     format.currentVersion,
     compression,
   )
-  let attempt = 0
-  for (;;) {
-    attempt += 1
-    signal?.throwIfAborted()
-    const source = await readStableSnapshot(sourcePath, signal, internals.fs)
-    const quickHeader = readPhysicalHeader(source.bytes, compression, signal)
-    const quickVersion = storedVersion(quickHeader.value)
-    if (quickVersion !== sourceVersion) {
-      throw new Error(
-        `resolved JSONL source filename identifies v${sourceVersion}, but its header identifies v${quickVersion}: `
-        + sourcePath,
-      )
-    }
-    if (sourceVersion > format.currentVersion) {
-      throw new JsonlGenerationNewerVersionError(
-        sourceVersion,
-        format.currentVersion,
-        storedId(quickHeader.value),
-      )
-    }
-    if (sourceVersion === format.currentVersion) {
-      return {
-        status: 'current',
-        version: quickVersion,
-        path: sourcePath,
-        snapshot: {
-          ...source,
-          headerValue: quickHeader.value,
-          headerRecord: quickHeader.record,
-        },
-      }
-    }
-    let artifact: SessionFormatArtifact
-    try {
-      artifact = await decodeStreamingMigration(
-        source.bytes,
-        compression,
-        format,
-        options.validateHistoricalHeader,
-        signal,
-      )
-    } catch (error: unknown) {
-      if (format.isUnsupportedMigrationError?.(error) === true) {
-        throw new JsonlGenerationUnsupportedMigrationError(sourceVersion, error)
-      }
-      throw error
-    }
-    if (artifact.header.version !== format.currentVersion) {
-      throw new Error(`format migration returned v${artifact.header.version}, expected v${format.currentVersion}`)
+  if (sourceVersion >= format.currentVersion) {
+    throw new Error(`migration preparation requires a historical source, got v${sourceVersion}`)
+  }
+  const source = await readStableSnapshot(sourcePath, signal, internals.fs)
+  let artifact: SessionFormatArtifact
+  try {
+    artifact = await decodeStreamingMigration(
+      source.bytes,
+      compression,
+      sourceVersion,
+      format,
+      options.validateHistoricalHeader,
+      signal,
+    )
+  } catch (error: unknown) {
+    if (format.isUnsupportedMigrationError?.(error) === true) {
+      throw new JsonlGenerationUnsupportedMigrationError(sourceVersion, error)
     }
-
-    await scheduler.yield()
-    signal?.throwIfAborted()
-    const sourceFingerprint = fingerprint(source.identity, source.bytes)
-    const eventCount = artifact.events.length
-    let staged = await writeSyncedTemp(currentPath, suffix, compression, artifact, format, signal, internals)
-    let failure: unknown
-    try {
-      const verifiedStage = await options.verifyCurrentFile(
-        staged.path,
-        compression,
-        artifact.header.id,
-        eventCount,
-        undefined,
-        signal,
-      )
-      if (verifiedStage.bytes !== staged.bytes || verifiedStage.digest !== staged.digest) {
-        throw new Error('staged session generation changed during verification')
-      }
-      await internals.barrier('before-source-check', attempt)
-      const beforePublish = await readStableSnapshot(sourcePath, signal, internals.fs)
-      if (fingerprint(beforePublish.identity, beforePublish.bytes) !== sourceFingerprint) continue
-
-      const published = await publishCurrentExclusive(staged.path, currentPath, internals)
-      if (published && internals.platform === 'win32') staged = { ...staged, path: '' }
-      await internals.barrier('after-publication', attempt)
-      signal?.throwIfAborted()
-      const committed = await reopenExpectedCurrent(
-        currentPath,
-        staged,
-        compression,
-        artifact.header.id,
-        eventCount,
-        options.verifyCurrentFile,
-        signal,
-        !published,
-        internals,
-      )
-      if (staged.path !== '') {
-        await removeCommittedTemporary(staged.path, internals)
-        staged = { ...staged, path: '' }
-      }
-      return {
-        status: 'migrated',
-        fromVersion: sourceVersion,
-        toVersion: format.currentVersion,
-        path: currentPath,
-        sourcePath,
-        snapshot: committed,
+    throw error
+  }
+  if (artifact.header.version !== format.currentVersion) {
+    throw new Error(`format migration returned v${artifact.header.version}, expected v${format.currentVersion}`)
+  }
+  const sourceIdentity = source.identity
+  let publication: Promise<JsonlPhysicalIdentity> | undefined
+  return {
+    sourceIdentity,
+    artifact,
+    publish() {
+      if (publication === undefined) {
+        publication = publishPreparedMigration(
+          options,
+          suffix,
+          artifact,
+          sourceIdentity,
+          internals,
+        )
       }
-    } catch (error: unknown) {
-      failure = error
-      throw error
-    } finally {
-      if (staged.path !== '') await removeTemporary(staged.path, failure, internals)
-    }
+      return publication
+    },
   }
 }
 
 /**
- * Ensure one resolved generation has a current-format successor before returning.
- * @param options - resolved source, current target, format adapter, verification, and cancellation.
- * @returns the current source or the verified and reopened migrated successor.
+ * Decode and migrate one historical generation without writing its successor.
+ * @param options - resolved source, current target, format adapter, and load cancellation.
+ * @returns the current artifact and an idempotent explicit publication operation.
  */
-export function ensureJsonlGenerationCurrent(
-  options: EnsureJsonlGenerationOptions,
-): Promise<EnsureJsonlGenerationResult> {
-  return defaultGenerationRuntime.ensure(options)
+export function prepareJsonlMigration(
+  options: PrepareJsonlMigrationOptions,
+): Promise<PreparedJsonlMigration> {
+  return defaultGenerationRuntime.prepare(options)
 }
 
 /**
@@ -1085,7 +1014,7 @@ export function createJsonlGenerationRuntime(
   const internals = withOverrides(overrides)
   return {
     readStable: (path, signal) => readStableSnapshot(path, signal, internals.fs),
-    ensure: options => ensureCurrent(options, internals),
+    prepare: options => prepareMigration(options, internals),
     verify: (path, compression, expectedId, expectedEventCount, expectedPrefix) => verifyCurrentGeneration(
       path, compression, expectedId, expectedEventCount, internals.fs, expectedPrefix,
     ),

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

@@ -24,12 +24,13 @@ import {
   SessionAlreadyExistsError, SessionPersistenceNotFoundError,
   assertStoredId, materializeCreateHeader, sessionFormatVersionRefusal, validateStoredEvents,
   type SessionAccess, type SessionHandle,
+  type SessionHandleReadResult,
   type SessionLocation, type SessionPersistenceCreateOptions,
   type SessionPersistenceListOptions, type SessionPersistenceOpenOptions,
   type SessionPersistenceSnapshot, type SessionPersistenceStatOptions,
   type SessionPersistenceRevision as PersistenceRevision,
 } from '@deepseek-ai/dsh-session-persistence'
-import { JsonlBackendTracker, JsonlSessionHandle } from './storage.ts'
+import { JsonlBackendTracker, JsonlSessionHandle, type StorageHandleState } from './storage.ts'
 import { SessionWriteLease } from './lease.ts'
 import { SESSION_FORMAT_VERSION, SessionId as makeSessionId, SessionLogOffset } from '@deepseek-ai/dsh-session'
 import type { SessionEvent, SessionId, SessionHeader, SessionLogOffset as SessionLogOffsetType } from '@deepseek-ai/dsh-session'
@@ -44,13 +45,13 @@ import {
 import { ensureDurableDirectoryWin32, publishNewFileWin32 } from './win32.ts'
 import { verifyCurrentGenerationInWorker } from './migration-verifier.ts'
 import {
-  ensureJsonlGenerationCurrent,
-  JsonlGenerationNewerVersionError,
+  JsonlGenerationSourceChangedError,
   JsonlGenerationUnsupportedMigrationError,
+  prepareJsonlMigration,
   readStableJsonlFile,
-  type EnsureJsonlGenerationResult,
   type JsonlGenerationFormatAdapter,
   type JsonlPhysicalIdentity,
+  type PreparedJsonlMigration,
 } from './generation.ts'
 
 export type { JsonlCompression } from './format.ts'
@@ -97,11 +98,14 @@ export interface Config {
   compression?: JsonlCompression
 }
 
-/** A parsed, validated stored log: header, logical events, and any torn-tail repair state. */
-interface StoredLog {
+/** One stored event graph whose producer has established immutable sharing. */
+interface FrozenStoredEvents extends SessionHandleReadResult {
+  readonly eventState: 'shared-frozen'
+}
+
+/** State shared by prepared historical and published current logs. */
+interface StoredLogBase extends FrozenStoredEvents {
   readonly meta: SessionHeader
-  /** The logical log, including any events recovered from a torn final frame. */
-  readonly events: SessionEvent[]
   readonly tornTruncateTo: number | undefined
   /** Complete events recovered from the torn final frame; the write path rewrites them durably. */
   readonly recoveredTail: SessionEvent[]
@@ -110,6 +114,45 @@ interface StoredLog {
   readonly revision: PersistenceRevision
 }
 
+/** A decoded current generation that is already durable. */
+interface CurrentStoredLog extends StoredLogBase {
+  readonly status: 'current'
+}
+
+/** A migrated historical generation retained until an explicit write open publishes it. */
+interface PreparedStoredLog extends StoredLogBase {
+  readonly status: 'prepared'
+  readonly publication: {
+    readonly source: ResolvedJsonlGeneration
+    readonly value: PreparedJsonlMigration
+  }
+}
+
+/** A validated logical log, either durable current state or prepared historical state. */
+type StoredLog = CurrentStoredLog | PreparedStoredLog
+
+/** Deep-freeze one acyclic stored JSON event without recursive calls. */
+function freezeStoredEvent(event: SessionEvent): void {
+  const pending: object[] = [event]
+  while (pending.length > 0) {
+    // The non-empty check proves an object remains to visit.
+    // oxlint-disable-next-line typescript/no-non-null-assertion
+    const current = pending.pop()!
+    Object.freeze(current)
+    for (const key in current) {
+      const child = (current as Record<string, unknown>)[key]
+      if (child !== null && typeof child === 'object') pending.push(child)
+    }
+  }
+}
+
+/** Establish immutable sharing for one decoded event graph and report that state. */
+function freezeStoredEvents(events: SessionEvent[]): FrozenStoredEvents {
+  for (const event of events) freezeStoredEvent(event)
+  Object.freeze(events)
+  return { eventState: 'shared-frozen', events }
+}
+
 /** One authoritative immutable generation selected from a Session directory. */
 interface ResolvedJsonlGeneration {
   readonly sourcePath: string
@@ -117,6 +160,16 @@ interface ResolvedJsonlGeneration {
   readonly currentPath: string
 }
 
+/** One backend-owned historical preparation shared by its current callers. */
+interface MigrationPreparation {
+  readonly sourcePath: string
+  readonly sourceRevision: PersistenceRevision
+  readonly controller: AbortController
+  readonly promise: Promise<PreparedStoredLog>
+  settled: boolean
+  waiters: number
+}
+
 /** Build the stat-derived best-effort change token shared by full and lightweight reads. */
 function fileRevision(identity: JsonlPhysicalIdentity): PersistenceRevision {
   return SessionPersistenceRevision([
@@ -138,6 +191,41 @@ function isErrnoException(error: unknown): error is NodeJS.ErrnoException {
   return typeof (error as NodeJS.ErrnoException | null)?.code === 'string'
 }
 
+/** Preserve an Error abort reason and normalize hostile non-Error reasons. */
+function abortError(signal: AbortSignal): Error {
+  return signal.reason instanceof Error
+    ? signal.reason
+    : new Error('session migration preparation aborted', { cause: signal.reason })
+}
+
+/** Let one caller stop waiting without transferring cancellation ownership to shared work. */
+function waitWithAbort<T>(operation: Promise<T>, signal?: AbortSignal): Promise<T> {
+  if (signal === undefined) return operation
+  /* v8 ignore next -- requireStoredLog synchronously rechecks the signal immediately before waiting. */
+  if (signal.aborted) return Promise.reject(abortError(signal))
+  return new Promise<T>((resolve, reject) => {
+    const stopWaiting = (): void => {
+      reject(abortError(signal))
+    }
+    signal.addEventListener('abort', stopWaiting, { once: true })
+    void operation.then(
+      (value) => {
+        signal.removeEventListener('abort', stopWaiting)
+        resolve(value)
+      },
+      (error: unknown) => {
+        signal.removeEventListener('abort', stopWaiting)
+        /* v8 ignore else -- the preparation owner normalizes every rejection before this waiter sees it. */
+        if (error instanceof Error) {
+          reject(error)
+        } else {
+          reject(new Error('session migration preparation failed', { cause: error }))
+        }
+      },
+    )
+  })
+}
+
 /**
  * The JSONL persistence backend. Load as a plugin; it registers as
  * `ctx.sessionPersistence`. Sessions materialize lazily: a created session is
@@ -166,6 +254,8 @@ class JsonlSessionPersistence extends SessionPersistence {
    * revision guard.
    */
   private readonly coldLogMemo = new Map<SessionId, StoredLog>()
+  /** One joinable decode/migration operation per selected historical Session file revision. */
+  private readonly migrationPreparations = new Map<SessionId, MigrationPreparation>()
 
   constructor(ctx: Context, public config: Config) {
     super(ctx)
@@ -253,11 +343,22 @@ class JsonlSessionPersistence extends SessionPersistence {
         return this.tracker.adopt(new JsonlSessionHandle(this, id, pending.header, 'read', { cursor: 0, materialized: false, inheritedEventCount: pending.inheritedEventCount }))
       }
       const stored = await this.requireStoredLog(id, options?.signal)
-      return this.tracker.adopt(new JsonlSessionHandle(this, id, stored.meta, 'read', {
-        cursor: 0,
-        materialized: true,
-        inheritedEventCount: stored.inheritedEventCount,
-      }))
+      let state: StorageHandleState
+      if (stored.status === 'prepared') {
+        state = {
+          cursor: 0,
+          materialized: true,
+          inheritedEventCount: stored.inheritedEventCount,
+          primed: stored,
+        }
+      } else {
+        state = {
+          cursor: 0,
+          materialized: true,
+          inheritedEventCount: stored.inheritedEventCount,
+        }
+      }
+      return this.tracker.adopt(new JsonlSessionHandle(this, id, stored.meta, 'read', state))
     }
     // A pending entry always belongs to an ACTIVE creator handle (close erases
     // it), so the claim below rejects that case as already owned.
@@ -267,14 +368,22 @@ class JsonlSessionPersistence extends SessionPersistence {
       const resolved = await this.findLog(id, options?.signal)
       if (resolved === undefined) throw new SessionPersistenceNotFoundError(id)
       lease = await this.acquireLease(id, undefined, dirname(resolved.currentPath))
-      const stored = await this.requireStoredLog(id, options?.signal)
+      const prepared = await this.requireStoredLog(id, options?.signal)
+      options?.signal?.throwIfAborted()
+      let stored: CurrentStoredLog
+      if (prepared.status === 'prepared') {
+        stored = await this.publishStoredMigration(id, prepared)
+      } else {
+        stored = prepared
+      }
+      options?.signal?.throwIfAborted()
       return this.tracker.adopt(new JsonlSessionHandle(this, id, stored.meta, 'write', {
         cursor: stored.events.length,
         materialized: true,
         tornTruncateTo: stored.tornTruncateTo,
         recoveredTail: stored.recoveredTail,
         inheritedEventCount: stored.inheritedEventCount,
-        primed: stored.events,
+        primed: stored,
       }, lease))
     } catch (error) {
       // Free the in-process claim no matter how the kernel-lock release
@@ -386,34 +495,115 @@ class JsonlSessionPersistence extends SessionPersistence {
   private async requireStoredLog(id: SessionId, signal?: AbortSignal): Promise<StoredLog> {
     const selected = await this.findLog(id, signal)
     if (selected === undefined) throw new SessionPersistenceNotFoundError(id)
-    if (selected.sourceVersion === SESSION_FORMAT_VERSION) {
-      const probe = fileRevision(await stat(selected.sourcePath, { bigint: true }))
-      const memoized = this.coldLogMemo.get(id)
-      if (memoized !== undefined && memoized.revision === probe) {
-        this.coldLogMemo.delete(id)
-        this.coldLogMemo.set(id, memoized)
-        return memoized
+    if (selected.sourceVersion < SESSION_FORMAT_VERSION) {
+      const sourceRevision = fileRevision(await stat(selected.sourcePath, { bigint: true }))
+      signal?.throwIfAborted()
+      let preparation = this.migrationPreparations.get(id)
+      if (preparation === undefined
+        || preparation.sourcePath !== selected.sourcePath
+        || preparation.sourceRevision !== sourceRevision) {
+        const controller = new AbortController()
+        const promise = this.loadStoredMigration(id, selected, sourceRevision, controller.signal)
+        preparation = {
+          sourcePath: selected.sourcePath,
+          sourceRevision,
+          controller,
+          promise,
+          settled: false,
+          waiters: 0,
+        }
+        this.migrationPreparations.set(id, preparation)
+        const created = preparation
+        const release = (): void => {
+          created.settled = true
+          if (this.migrationPreparations.get(id) === created) {
+            this.migrationPreparations.delete(id)
+          }
+        }
+        void promise.then(release, release)
       }
+      signal?.throwIfAborted()
+      return this.waitForPreparation(id, preparation, signal)
+    }
+    if (selected.sourceVersion > SESSION_FORMAT_VERSION) {
+      const header = await this.readGenerationHeader(selected, id, signal)
+      /* v8 ignore else -- a readable future header is rejected inside readGenerationHeader. */
+      if (header === undefined) {
+        throw new SessionPersistenceCorruptionError(
+          `session "${id}": stored log has a malformed header (raw log: ${selected.sourcePath})`,
+          { cause: new Error('malformed Session header') },
+        )
+      }
+      /* v8 ignore next -- readGenerationHeader rejects every future version. */
+      throw new SessionFormatUnsupportedError(
+        `${sessionFormatVersionRefusal(id, selected.sourceVersion)} (raw log: ${selected.sourcePath})`,
+        { kind: 'jsonl', path: selected.sourcePath },
+      )
     }
-    const current = await this.ensureCurrentLog(id, signal, selected)
+    const probe = fileRevision(await stat(selected.sourcePath, { bigint: true }))
+    const memoized = this.coldLogMemo.get(id)
+    if (memoized?.status === 'current' && memoized.revision === probe) {
+      this.coldLogMemo.delete(id)
+      this.coldLogMemo.set(id, memoized)
+      return memoized
+    }
+    const current = await readStableJsonlFile(selected.sourcePath, signal)
     return this.decodeStoredLog(
-      current.path,
+      selected.sourcePath,
       id,
-      current.snapshot.bytes,
-      fileRevision(current.snapshot.identity),
+      current.bytes,
+      fileRevision(current.identity),
       signal,
     )
   }
 
-  /** Select and, when required, publish one immutable current generation. */
-  private async ensureCurrentLog(
+  /** Probe the memo and otherwise decode one historical generation under backend cancellation. */
+  private async loadStoredMigration(
     id: SessionId,
-    signal: AbortSignal | undefined,
     selected: ResolvedJsonlGeneration,
-  ): Promise<EnsureJsonlGenerationResult> {
-    signal?.throwIfAborted()
+    sourceRevision: PersistenceRevision,
+    signal: AbortSignal,
+  ): Promise<PreparedStoredLog> {
+    signal.throwIfAborted()
+    const memoized = this.coldLogMemo.get(id)
+    if (memoized?.status === 'prepared' && memoized.revision === sourceRevision) {
+      this.coldLogMemo.delete(id)
+      this.coldLogMemo.set(id, memoized)
+      return memoized
+    }
+    return this.prepareStoredMigration(id, selected, signal)
+  }
+
+  /** Await shared preparation for one caller and abort it only after its last waiter leaves. */
+  private async waitForPreparation(
+    id: SessionId,
+    preparation: MigrationPreparation,
+    signal?: AbortSignal,
+  ): Promise<PreparedStoredLog> {
+    preparation.waiters += 1
     try {
-      return await ensureJsonlGenerationCurrent({
+      return await waitWithAbort(preparation.promise, signal)
+    } finally {
+      preparation.waiters -= 1
+      if (preparation.waiters === 0 && !preparation.settled) {
+        /* v8 ignore else -- a newer selected source may already own this id's preparation slot. */
+        if (this.migrationPreparations.get(id) === preparation) {
+          this.migrationPreparations.delete(id)
+        }
+        preparation.controller.abort()
+      }
+    }
+  }
+
+  /** Decode one historical generation without publishing a successor. */
+  private async prepareStoredMigration(
+    id: SessionId,
+    selected: ResolvedJsonlGeneration,
+    signal: AbortSignal,
+  ): Promise<PreparedStoredLog> {
+    let prepared: Awaited<ReturnType<typeof prepareJsonlMigration>>
+    try {
+      prepared = await prepareJsonlMigration({
         sourcePath: selected.sourcePath,
         sourceVersion: selected.sourceVersion,
         currentPath: selected.currentPath,
@@ -426,32 +616,75 @@ class JsonlSessionPersistence extends SessionPersistence {
           id,
           signal,
         ),
-        ...(signal === undefined ? {} : { signal }),
+        signal,
       })
     } catch (error: unknown) {
-      signal?.throwIfAborted()
-      if (error instanceof JsonlGenerationNewerVersionError) {
-        const reason = sessionFormatVersionRefusal(error.storedId, error.storedVersion)
-        throw new SessionFormatUnsupportedError(
-          `${reason} (raw log: ${selected.sourcePath})`,
-          { kind: 'jsonl', path: selected.sourcePath },
-        )
-      }
-      if (error instanceof JsonlGenerationUnsupportedMigrationError) {
-        throw new SessionFormatUnsupportedError(
-          `${error.message}; source v${error.fromVersion} artifact remains unchanged (raw log: ${selected.sourcePath})`,
-          { kind: 'jsonl', path: selected.sourcePath },
-        )
-      }
-      if (error instanceof SessionFormatUnsupportedError
-        || error instanceof SessionPersistenceCorruptionError
-        || isErrnoException(error)
-        || error instanceof DOMException && error.name === 'AbortError') throw error
-      throw new SessionPersistenceCorruptionError(
-        `session "${id}": stored log is corrupt: ${String(error)} (raw log: ${selected.sourcePath})`,
-        { cause: error },
+      throw this.generationFailure(id, selected, error)
+    }
+    const meta = this.currentHeader(prepared.artifact.header)
+    assertStoredId(id, meta)
+    const events = prepared.artifact.events as SessionEvent[]
+    validateStoredEvents(meta, events, { kind: 'jsonl', path: selected.sourcePath })
+    const stored: PreparedStoredLog = {
+      status: 'prepared',
+      meta,
+      ...freezeStoredEvents(events),
+      tornTruncateTo: undefined,
+      recoveredTail: [],
+      inheritedEventCount: SessionLogOffset(prepared.artifact.inheritedEventCount),
+      revision: fileRevision(prepared.sourceIdentity),
+      publication: { source: selected, value: prepared },
+    }
+    this.memoizeStoredLog(id, stored)
+    return stored
+  }
+
+  /** Publish a prepared historical log before granting write access. */
+  private async publishStoredMigration(id: SessionId, stored: PreparedStoredLog): Promise<CurrentStoredLog> {
+    const migration = stored.publication
+    let identity: JsonlPhysicalIdentity
+    try {
+      identity = await migration.value.publish()
+    } catch (error: unknown) {
+      /* v8 ignore else -- a newer preparation may have replaced this stale cache entry. */
+      if (this.coldLogMemo.get(id) === stored) this.coldLogMemo.delete(id)
+      throw this.generationFailure(id, migration.source, error)
+    }
+    const published: CurrentStoredLog = {
+      status: 'current',
+      meta: stored.meta,
+      eventState: stored.eventState,
+      events: stored.events,
+      tornTruncateTo: stored.tornTruncateTo,
+      recoveredTail: stored.recoveredTail,
+      inheritedEventCount: stored.inheritedEventCount,
+      revision: fileRevision(identity),
+    }
+    this.memoizeStoredLog(id, published)
+    return published
+  }
+
+  /** Translate generation-layer failures into the persistence seam's error vocabulary. */
+  private generationFailure(
+    id: SessionId,
+    selected: ResolvedJsonlGeneration,
+    error: unknown,
+  ): Error {
+    if (error instanceof JsonlGenerationUnsupportedMigrationError) {
+      return new SessionFormatUnsupportedError(
+        `${error.message}; source v${error.fromVersion} artifact remains unchanged (raw log: ${selected.sourcePath})`,
+        { kind: 'jsonl', path: selected.sourcePath },
       )
     }
+    if (error instanceof JsonlGenerationSourceChangedError) return error
+    if (error instanceof SessionFormatUnsupportedError
+      || error instanceof SessionPersistenceCorruptionError
+      || isErrnoException(error)
+      || error instanceof DOMException && error.name === 'AbortError') return error
+    return new SessionPersistenceCorruptionError(
+      `session "${id}": stored log is corrupt: ${String(error)} (raw log: ${selected.sourcePath})`,
+      { cause: error },
+    )
   }
 
   /**
@@ -461,11 +694,11 @@ class JsonlSessionPersistence extends SessionPersistence {
    * @param signal - optional cancellation for the stat/read/decode work.
    * @returns the validated stored log with any torn-tail truncation point.
    */
-  async readStoredLog(path: string, expectedId: SessionId, signal?: AbortSignal): Promise<StoredLog> {
+  async readStoredLog(path: string, expectedId: SessionId, signal?: AbortSignal): Promise<CurrentStoredLog> {
     signal?.throwIfAborted()
     const probe = fileRevision(await stat(path, { bigint: true }))
     const memoized = this.coldLogMemo.get(expectedId)
-    if (memoized !== undefined && memoized.revision === probe) {
+    if (memoized?.status === 'current' && memoized.revision === probe) {
       this.coldLogMemo.delete(expectedId)
       this.coldLogMemo.set(expectedId, memoized)
       return memoized
@@ -481,7 +714,7 @@ class JsonlSessionPersistence extends SessionPersistence {
     buffer: Buffer,
     revision: PersistenceRevision,
     signal?: AbortSignal,
-  ): Promise<StoredLog> {
+  ): Promise<CurrentStoredLog> {
     let parsed: {
       meta: SessionHeader
       inheritedEventCount: SessionLogOffsetType
@@ -523,30 +756,45 @@ class JsonlSessionPersistence extends SessionPersistence {
     assertStoredId(expectedId, parsed.meta)
     const location = this.locate(parsed.meta)
     validateStoredEvents(parsed.meta, parsed.events, location)
-    const stored: StoredLog = { ...parsed, revision }
-    this.coldLogMemo.delete(expectedId)
-    this.coldLogMemo.set(expectedId, stored)
+    const { events, ...rest } = parsed
+    const stored: CurrentStoredLog = {
+      status: 'current',
+      ...rest,
+      ...freezeStoredEvents(events),
+      revision,
+    }
+    this.memoizeStoredLog(expectedId, stored)
+    return stored
+  }
+
+  /** Insert one parsed log into the bounded handoff cache. */
+  private memoizeStoredLog(id: SessionId, stored: StoredLog): void {
+    this.coldLogMemo.delete(id)
+    this.coldLogMemo.set(id, stored)
     for (const oldest of this.coldLogMemo.keys()) {
       if (this.coldLogMemo.size <= COLD_LOG_MEMO_MAX_ENTRIES) break
       this.coldLogMemo.delete(oldest)
     }
-    return stored
   }
 
   /**
-   * Resolve a session's unique log path.
+   * Resolve a session's current-generation log path.
    * @param id - the stored session to locate.
    * @param signal - optional cancellation for the directory scans.
-   * @returns the artifact path, or `undefined` when absent.
+   * @returns the current artifact path, or `undefined` while only a historical generation exists.
    */
-  async resolveLog(id: SessionId, signal?: AbortSignal): Promise<string | undefined> {
+  async resolveCurrentLog(id: SessionId, signal?: AbortSignal): Promise<string | undefined> {
     await this.ensureRootEncoding()
     signal?.throwIfAborted()
     const selected = await this.findLog(id, signal)
     if (selected === undefined) return undefined
     if (selected.sourceVersion === SESSION_FORMAT_VERSION) return selected.sourcePath
-    const current = await this.ensureCurrentLog(id, signal, selected)
-    return current.path
+    if (selected.sourceVersion < SESSION_FORMAT_VERSION) return undefined
+    const reason = sessionFormatVersionRefusal(id, selected.sourceVersion)
+    throw new SessionFormatUnsupportedError(
+      `${reason} (raw log: ${selected.sourcePath})`,
+      { kind: 'jsonl', path: selected.sourcePath },
+    )
   }
 
   /**

packages/session/session-persistence-jsonl/src/storage.ts

@@ -28,6 +28,7 @@ import type {
   SessionHandleAppendOptions,
   SessionHandleFlushOptions,
   SessionHandleReadOptions,
+  SessionHandleReadResult,
 } from '@deepseek-ai/dsh-session-persistence'
 import type { SessionWriteLease } from './lease.ts'
 
@@ -47,10 +48,10 @@ export interface JsonlHandleStorage {
   persistHeader(header: SessionHeader, inheritedEventCount: SessionLogOffset): Promise<void>
   /** Truncate a torn physical tail before the first new append lands. */
   truncateTornTail(header: SessionHeader, truncateTo: number): Promise<void>
-  /** Resolve the session's artifact path, or `undefined` before materialization. */
-  resolveLog(id: SessionId, signal?: AbortSignal): Promise<string | undefined>
-  /** Read and validate the stored log at `path`. */
-  readStoredLog(path: string, expectedId: SessionId, signal?: AbortSignal): Promise<{ events: SessionEvent[] }>
+  /** Resolve the current-generation artifact path, or `undefined` when absent. */
+  resolveCurrentLog(id: SessionId, signal?: AbortSignal): Promise<string | undefined>
+  /** Read and validate the stored log at `path`, including its established event aliasing state. */
+  readStoredLog(path: string, expectedId: SessionId, signal?: AbortSignal): Promise<SessionHandleReadResult>
   /** Whether the id is still a created-but-unmaterialized session here. */
   hasPendingSession(id: SessionId): boolean
   /** Acquire the session's cross-process write lock in its artifact directory. */
@@ -72,7 +73,7 @@ export interface StorageHandleState {
   /** Exact fork-inherited prefix length stored with the log; `0` when unseeded. */
   inheritedEventCount: SessionLogOffset
   /** The validated stored prefix from a write open, served to reads until the first append. */
-  primed?: SessionEvent[] | undefined
+  primed?: SessionHandleReadResult | undefined
 }
 
 /**
@@ -112,9 +113,9 @@ export class JsonlSessionHandle implements SessionHandle {
    * @param offset - first logical seq to include (default 0).
    * @param length - maximum events returned (default: the rest).
    * @param options - optional cancellation.
-   * @returns the requested slice.
+   * @returns a slice carrying the aliasing state established by its producer.
    */
-  async read(offset = 0, length = Number.MAX_SAFE_INTEGER, options?: SessionHandleReadOptions): Promise<readonly SessionEvent[]> {
+  async read(offset = 0, length = Number.MAX_SAFE_INTEGER, options?: SessionHandleReadOptions): Promise<SessionHandleReadResult> {
     // Closed-handle refusal precedes argument validation: a closed handle
     // rejects SessionHandleClosedError regardless of the arguments.
     this.assertOpen('read')
@@ -125,24 +126,57 @@ export class JsonlSessionHandle implements SessionHandle {
       throw new TypeError(`read length must be a non-negative safe integer, got ${String(length)}`)
     }
     options?.signal?.throwIfAborted()
-    if (this.state.primed !== undefined) {
-      this.observedLength = Math.max(this.observedLength, this.state.primed.length)
-      return this.state.primed.slice(offset, offset + length)
+    let result: SessionHandleReadResult
+    const primed = this.state.primed
+    if (primed !== undefined) {
+      if (this.access === 'write') {
+        result = this.readPrimed(primed, offset, length)
+      } else {
+        const currentPath = await this.storage.resolveCurrentLog(this.id, options?.signal)
+        if (currentPath === undefined) {
+          result = this.readPrimed(primed, offset, length)
+        } else {
+          this.state.primed = undefined
+          result = await this.readCurrent(currentPath, offset, length, options?.signal)
+        }
+      }
+    } else if (this.access === 'write' && !this.state.materialized) {
+      result = { eventState: 'detached', events: [] }
+    } else {
+      const currentPath = await this.storage.resolveCurrentLog(this.id, options?.signal)
+      if (currentPath !== undefined) {
+        result = await this.readCurrent(currentPath, offset, length, options?.signal)
+      } else if (this.storage.hasPendingSession(this.id)) {
+        result = { eventState: 'detached', events: [] }
+      } else {
+        throw new SessionPersistenceNotFoundError(this.id)
+      }
     }
-    // A write handle knows its own materialization; a read handle asks the
-    // backend so a writer's later materialization becomes visible here.
-    if (this.access === 'write' && !this.state.materialized) return []
-    const path = await this.storage.resolveLog(this.id, options?.signal)
-    if (path === undefined) {
-      if (this.storage.hasPendingSession(this.id)) return []
-      throw new SessionPersistenceNotFoundError(this.id)
+    return result
+  }
+
+  /** Read one slice from the prepared historical prefix retained by this handle. */
+  private readPrimed(source: SessionHandleReadResult, offset: number, length: number): SessionHandleReadResult {
+    this.observedLength = Math.max(this.observedLength, source.events.length)
+    return { eventState: source.eventState, events: source.events.slice(offset, offset + length) }
+  }
+
+  /** Read one current physical generation and enforce this handle's monotonic view. */
+  private async readCurrent(
+    path: string,
+    offset: number,
+    length: number,
+    signal?: AbortSignal,
+  ): Promise<SessionHandleReadResult> {
+    const source = await this.storage.readStoredLog(path, this.id, signal)
+    if (source.events.length < this.observedLength) {
+      throw new Error(`session "${this.id}": stored log shrank below a previously observed prefix (${source.events.length} < ${this.observedLength})`)
     }
-    const { events } = await this.storage.readStoredLog(path, this.id, options?.signal)
-    if (events.length < this.observedLength) {
-      throw new Error(`session "${this.id}": stored log shrank below a previously observed prefix (${events.length} < ${this.observedLength})`)
+    this.observedLength = source.events.length
+    return {
+      eventState: source.eventState,
+      events: source.events.slice(offset, offset + length),
     }
-    this.observedLength = events.length
-    return events.slice(offset, offset + length)
   }
 
   /**

packages/session/session-persistence-jsonl/tests/built-migration-worker.e2e.ts

@@ -27,7 +27,7 @@ describe.skipIf(!built)('built migration verifier (plain node)', () => {
           type: 'session', version: 0, id, createdAt: 1, delegationDepth: 0,
         }) + '\\n')
         await ctx.plugin(JsonlSessionPersistence, { root, compression: 'none' })
-        const handle = await ctx.sessionPersistence.open(id, 'read')
+        const handle = await ctx.sessionPersistence.open(id, 'write')
         await handle.close()
         await ctx.sessionPersistence.flush()
         const header = JSON.parse((await readFile(join(directory, 'session.v2.jsonl'), 'utf8')).trim())

packages/session/session-persistence-jsonl/tests/generation.spec.ts

@@ -17,19 +17,24 @@ import {
 import { tmpdir } from 'node:os'
 import { basename, join } from 'node:path'
 import { performance } from 'node:perf_hooks'
-import { scheduler } from 'node:timers/promises'
 import {
-  ensureJsonlGenerationCurrent as ensureJsonlGenerationCurrentProduction,
+  JsonlGenerationSourceChangedError,
   JsonlGenerationTargetConflictError,
   JsonlGenerationUnsupportedMigrationError,
+  prepareJsonlMigration,
   verifyJsonlCurrentGeneration,
-  type EnsureJsonlGenerationOptions,
   type JsonlGenerationFormatAdapter,
+  type PrepareJsonlMigrationOptions,
 } from '../src/generation.ts'
 import { createJsonlGenerationTestRuntime } from '../src/testing/generation.ts'
 import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from '../src/zstd.ts'
 import type { JsonlCompression } from '../src/format.ts'
-import type { SessionFormatArtifact, SessionFormatRestore } from '@deepseek-ai/dsh-session-format'
+import type {
+  SessionFormatArtifact,
+  SessionFormatEvent,
+  SessionFormatJsonValue,
+  SessionFormatRestore,
+} from '@deepseek-ai/dsh-session-format'
 
 const roots: string[] = []
 
@@ -78,6 +83,61 @@ function header(version: number, id = 'generation-test'): Record<string, unknown
 
 const event0 = { type: 'turn/start', seq: 0, time: 2, data: { turn: 1 } }
 const event1 = { type: 'turn/end', seq: 1, time: 3, data: { turn: 1, reason: { kind: 'completed' } } }
+const assistantUsage = { inputTokens: 3, outputTokens: 2 }
+const assistantReplayState = { response: { id: 'response' } }
+
+function assistantData(
+  overrides: {
+    readonly content?: readonly SessionFormatJsonValue[]
+    readonly stream?: SessionFormatJsonValue
+    readonly usage?: SessionFormatJsonValue
+    readonly replayState?: SessionFormatJsonValue
+    readonly interrupted?: true
+  } = {},
+): SessionFormatJsonValue {
+  const replayState = overrides.replayState === undefined
+    ? assistantReplayState
+    : overrides.replayState
+  return {
+    turn: 1,
+    step: 1,
+    message: {
+      id: 'assistant',
+      role: 'assistant',
+      content: overrides.content ?? [{ type: 'text', text: 'hello' }],
+      source: {
+        kind: 'model', provider: 'mock', model: 'mock',
+        ...(replayState === null ? {} : { replayState }),
+      },
+    },
+    stream: overrides.stream ?? [
+      { type: 'text-chunks', time0: 3, index: 0, dt: [], texts: ['hello'] },
+      { type: 'chunk', time: 4, chunk: { type: 'usage', usage: assistantUsage } },
+      { type: 'chunk', time: 5, chunk: { type: 'finish', reason: { kind: 'stop' }, replayState: assistantReplayState } },
+    ],
+    ...(overrides.usage === null ? {} : { usage: overrides.usage ?? assistantUsage }),
+    ...(overrides.interrupted === undefined ? {} : { interrupted: overrides.interrupted }),
+  }
+}
+
+function assistantLifecycle(
+  type: 'assistant/message' | 'assistant/attempt',
+  data: SessionFormatJsonValue,
+): SessionFormatEvent[] {
+  return [
+    { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } },
+    { type: 'step/start', seq: 1, time: 2, data: { turn: 1, step: 1 } },
+    {
+      type,
+      seq: 2,
+      time: 5,
+      data,
+      ...(type === 'assistant/message' ? { surfaceOp: 'append' as const } : {}),
+    },
+    { type: 'step/end', seq: 3, time: 6, data: { turn: 1, step: 1 } },
+    { type: 'turn/end', seq: 4, time: 7, data: { turn: 1, reason: { kind: 'completed' } } },
+  ]
+}
 
 interface TestGenerationFormatAdapter extends JsonlGenerationFormatAdapter {
   createRestore(header: Record<string, unknown>): SessionFormatRestore
@@ -120,12 +180,12 @@ function streamingAdapter(): JsonlGenerationFormatAdapter & {
   return adapter()
 }
 
-function verifier(): EnsureJsonlGenerationOptions['verifyCurrentFile'] {
+function verifier(): PrepareJsonlMigrationOptions['verifyCurrentFile'] {
   return (path, compression, expectedId, expectedEventCount, expectedPrefix) =>
     verifyJsonlCurrentGeneration(path, compression, expectedId, expectedEventCount, expectedPrefix)
 }
 
-const byteVerifier: EnsureJsonlGenerationOptions['verifyCurrentFile'] = async (path) => {
+const byteVerifier: PrepareJsonlMigrationOptions['verifyCurrentFile'] = async (path) => {
   const [bytes, identity] = await Promise.all([readFile(path), stat(path, { bigint: true })])
   return {
     identity,
@@ -144,7 +204,7 @@ function options(
   compression: JsonlCompression = 'none',
   format: JsonlGenerationFormatAdapter = adapter(),
   sourceVersion = 0,
-): Omit<EnsureJsonlGenerationOptions, 'verifyCurrentFile'> {
+): Omit<PrepareJsonlMigrationOptions, 'verifyCurrentFile'> {
   return {
     sourcePath: generationPath(root, sourceVersion, compression),
     sourceVersion,
@@ -156,7 +216,7 @@ function options(
 
 type TestMigrationOptions = ReturnType<typeof options> & {
   readonly signal?: AbortSignal
-  readonly verifyCurrentFile?: EnsureJsonlGenerationOptions['verifyCurrentFile']
+  readonly verifyCurrentFile?: PrepareJsonlMigrationOptions['verifyCurrentFile']
 }
 type TestGenerationOverrides = Parameters<typeof createJsonlGenerationTestRuntime>[0]
 
@@ -175,17 +235,24 @@ async function ensureWithOverrides(
         expectedPrefix,
       )
   )
-  return runtime.ensure({
+  const prepared = await runtime.prepare({
     ...request,
     verifyCurrentFile,
   })
+  const identity = await prepared.publish()
+  const bytes = await readFile(request.currentPath)
+  return {
+    status: 'migrated' as const,
+    fromVersion: request.sourceVersion,
+    toVersion: request.format.currentVersion,
+    path: request.currentPath,
+    sourcePath: request.sourcePath,
+    snapshot: { identity, bytes },
+  }
 }
 
 function ensureJsonlGenerationCurrent(request: TestMigrationOptions) {
-  return ensureJsonlGenerationCurrentProduction({
-    ...request,
-    verifyCurrentFile: request.verifyCurrentFile ?? verifier(),
-  })
+  return ensureWithOverrides(request, {})
 }
 
 async function encodeZstd(version: number, rows: readonly unknown[]): Promise<Buffer> {
@@ -207,7 +274,7 @@ async function decodeZstdJsonl(path: string): Promise<string> {
 }
 
 describe('JSONL immutable generation publication', () => {
-  it('does not return until verification and publication complete', async () => {
+  it('returns migrated events while publication is still waiting for verification', async () => {
     const root = await tempRoot()
     const request = options(root, 'none', streamingAdapter())
     const boundaryBase = { ...event0, data: { turn: 1, text: '' } }
@@ -223,7 +290,7 @@ describe('JSONL immutable generation publication', () => {
     const entered = Promise.withResolvers<undefined>()
     const release = Promise.withResolvers<undefined>()
 
-    const migration = ensureJsonlGenerationCurrent({
+    const prepared = await prepareJsonlMigration({
       ...request,
       verifyCurrentFile: async (path, compression, expectedId, expectedEventCount) => {
         entered.resolve(undefined)
@@ -231,11 +298,14 @@ describe('JSONL immutable generation publication', () => {
         return verifyJsonlCurrentGeneration(path, compression, expectedId, expectedEventCount)
       },
     })
+    expect(prepared.artifact.events).toEqual([boundaryEvent, largeEvent, finalEvent])
+    const publication = prepared.publish()
+    expect(prepared.publish()).toBe(publication)
     await entered.promise
     await expect(readFile(request.currentPath)).rejects.toMatchObject({ code: 'ENOENT' })
 
     release.resolve(undefined)
-    await migration
+    await publication
     const [writtenHeader, ...writtenEvents] = (await readFile(request.currentPath, 'utf8')).trimEnd().split('\n')
     expect(JSON.parse(writtenHeader as string)).toEqual({ ...header(2), isSeeded: false })
     expect(writtenEvents.map(row => JSON.parse(row) as unknown)).toEqual([boundaryEvent, largeEvent, finalEvent])
@@ -252,15 +322,16 @@ describe('JSONL immutable generation publication', () => {
     const events = widths.map((_, seq) => ({ ...event0, seq }))
     await writeFile(request.sourcePath, line(header(0)) + events.map(line).join(''))
 
-    await ensureJsonlGenerationCurrent({
+    const prepared = await prepareJsonlMigration({
       ...request,
       verifyCurrentFile: byteVerifier,
     })
+    await prepared.publish()
 
     expect((await stat(request.currentPath)).size).toBeGreaterThan(8 * mib)
   })
 
-  it('retries migration when the source changes before publication', async () => {
+  it('fails publication without rerunning migration when the source changes', async () => {
     const root = await tempRoot()
     const base = streamingAdapter()
     const sourceStreams = vi.fn()
@@ -274,21 +345,18 @@ describe('JSONL immutable generation publication', () => {
     const source = line(header(0)) + line(event0)
     await writeFile(request.sourcePath, source)
 
-    let verifications = 0
-    await ensureJsonlGenerationCurrent({
+    const prepared = await prepareJsonlMigration({
       ...request,
       verifyCurrentFile: async (path, compression, expectedId, expectedEventCount) => {
         const verified = await verifyJsonlCurrentGeneration(path, compression, expectedId, expectedEventCount)
-        if (++verifications === 1) await writeFile(request.sourcePath, source + line(event1))
+        await writeFile(request.sourcePath, source + line(event1))
         return verified
       },
     })
 
-    expect(sourceStreams).toHaveBeenCalledTimes(2)
-    expect(verifications).toBe(3)
-    expect(await readFile(request.currentPath, 'utf8')).toBe(
-      line(header(2)) + line(event0) + line(event1),
-    )
+    await expect(prepared.publish()).rejects.toBeInstanceOf(JsonlGenerationSourceChangedError)
+    expect(sourceStreams).toHaveBeenCalledOnce()
+    await expect(readFile(request.currentPath)).rejects.toMatchObject({ code: 'ENOENT' })
   })
 
   it('refuses malformed streaming inputs before publication', async () => {
@@ -296,23 +364,24 @@ describe('JSONL immutable generation publication', () => {
     const request = options(root, 'none', streamingAdapter())
 
     await writeFile(request.sourcePath, '')
-    await expect(ensureJsonlGenerationCurrent({ ...request, verifyCurrentFile: vi.fn() }))
+    await expect(prepareJsonlMigration({ ...request, verifyCurrentFile: vi.fn() }))
       .rejects.toThrow('empty or header-less')
 
     await writeFile(request.sourcePath, line(header(1)))
-    await expect(ensureJsonlGenerationCurrent({ ...request, verifyCurrentFile: vi.fn() }))
+    await expect(prepareJsonlMigration({ ...request, verifyCurrentFile: vi.fn() }))
       .rejects.toThrow(/filename identifies v0.*header identifies v1/)
 
     await writeFile(request.sourcePath, line(header(0)) + '{bad json}\n' + line(event1))
-    await expect(ensureJsonlGenerationCurrent({ ...request, verifyCurrentFile: vi.fn() }))
+    await expect(prepareJsonlMigration({ ...request, verifyCurrentFile: vi.fn() }))
       .rejects.toThrow('row 1 is not valid JSON')
 
     await writeFile(request.sourcePath, line(header(0)) + '{bad json}\n' + line(event0))
-    await ensureJsonlGenerationCurrent({
+    const dropped = await prepareJsonlMigration({
       ...request,
       verifyCurrentFile: verifier(),
     })
-    expect((await readFile(request.currentPath, 'utf8')).trimEnd().split('\n')).toHaveLength(1)
+    expect(dropped.artifact.events).toEqual([])
+    await dropped.publish()
 
   })
 
@@ -346,11 +415,52 @@ describe('JSONL immutable generation publication', () => {
       .rejects.toThrow('torn physical tail')
   })
 
+  it('keeps complete Assistant stream checks in current-generation verification', async () => {
+    const root = await tempRoot()
+    const path = generationPath(root, 2, 'none')
+    const verify = async (events: readonly SessionFormatEvent[]) => {
+      await writeFile(path, line(header(2)) + events.map(line).join(''))
+      return verifyJsonlCurrentGeneration(path, 'none', 'generation-test', events.length)
+    }
+
+    const valid = [
+      assistantLifecycle('assistant/message', assistantData()),
+      assistantLifecycle('assistant/message', assistantData({
+        interrupted: true,
+        stream: [{ type: 'text-chunks', time0: 3, index: 0, dt: [], texts: ['hello'] }],
+        usage: null,
+        replayState: null,
+      })),
+      assistantLifecycle('assistant/message', assistantData({
+        content: [], stream: [], usage: null, replayState: null,
+      })),
+      assistantLifecycle('assistant/attempt', {
+        turn: 1,
+        step: 1,
+        stream: [{ type: 'text-chunks', time0: 3, index: 0, dt: [1], texts: ['a', 'b'] }],
+      }),
+    ]
+    for (const events of valid) expect((await verify(events)).bytes).toBeGreaterThan(0)
+
+    await expect(verify(assistantLifecycle('assistant/attempt', {
+      turn: 1, step: 1, stream: [{ type: 'future' }],
+    }))).rejects.toThrow(/invalid embedded stream/)
+    await expect(verify(assistantLifecycle('assistant/message', assistantData({
+      content: [{ type: 'text', text: 'different' }],
+    })))).rejects.toThrow(/content disagrees/)
+    await expect(verify(assistantLifecycle('assistant/message', assistantData({
+      usage: { inputTokens: 9, outputTokens: 2 },
+    })))).rejects.toThrow(/usage disagrees/)
+    await expect(verify(assistantLifecycle('assistant/message', assistantData({
+      replayState: { response: { id: 'different' } },
+    })))).rejects.toThrow(/replay state disagrees/)
+  })
+
   it('accepts an identical publication winner and rejects different bytes', async () => {
     const identicalRoot = await tempRoot()
     const identical = options(identicalRoot, 'none', streamingAdapter())
     await writeFile(identical.sourcePath, line(header(0)) + line(event0))
-    await ensureJsonlGenerationCurrent({
+    const prepared = await prepareJsonlMigration({
       ...identical,
       verifyCurrentFile: async (path, compression, expectedId, expectedEventCount) => {
         const verified = await verifyJsonlCurrentGeneration(path, compression, expectedId, expectedEventCount)
@@ -358,16 +468,17 @@ describe('JSONL immutable generation publication', () => {
         return verified
       },
     })
-    expect((await stat(identical.currentPath, { bigint: true })).size).toBeGreaterThan(0n)
+    expect((await prepared.publish()).size).toBeGreaterThan(0n)
 
     const differentRoot = await tempRoot()
     const different = options(differentRoot, 'none', streamingAdapter())
     await writeFile(different.sourcePath, line(header(0)) + line(event0))
     await writeFile(different.currentPath, line({ ...header(2), isSeeded: false }) + line({ ...event0, time: 99 }))
-    await expect(ensureJsonlGenerationCurrent({
+    const conflicted = await prepareJsonlMigration({
       ...different,
       verifyCurrentFile: verifier(),
-    })).rejects.toBeInstanceOf(JsonlGenerationTargetConflictError)
+    })
+    await expect(conflicted.publish()).rejects.toBeInstanceOf(JsonlGenerationTargetConflictError)
 
     const uncheckedRoot = await tempRoot()
     const unchecked = options(uncheckedRoot, 'none', streamingAdapter())
@@ -376,17 +487,19 @@ describe('JSONL immutable generation publication', () => {
       unchecked.currentPath,
       line({ ...header(2), isSeeded: false }) + line({ ...event0, time: 99 }),
     )
-    await expect(ensureJsonlGenerationCurrent({
+    const uncheckedPublication = await prepareJsonlMigration({
       ...unchecked,
       verifyCurrentFile: byteVerifier,
-    })).rejects.toThrow(/target bytes differ from the migrated generation/)
+    })
+    await expect(uncheckedPublication.publish())
+      .rejects.toThrow(/target bytes differ from the migrated generation/)
   })
 
   it('handles empty, incomplete-record, and torn Zstandard migration sources', async () => {
     const emptyRoot = await tempRoot()
     const empty = options(emptyRoot, 'zstd', streamingAdapter())
     await writeFile(empty.sourcePath, Buffer.alloc(0))
-    await expect(ensureJsonlGenerationCurrent({ ...empty, verifyCurrentFile: vi.fn() }))
+    await expect(prepareJsonlMigration({ ...empty, verifyCurrentFile: vi.fn() }))
       .rejects.toThrow('empty or header-less Zstandard')
 
     const incompleteRoot = await tempRoot()
@@ -395,7 +508,7 @@ describe('JSONL immutable generation publication', () => {
       await compressZstdFrame(line(header(0))),
       await compressZstdFrame(JSON.stringify(event0)),
     ]))
-    await expect(ensureJsonlGenerationCurrent({ ...incomplete, verifyCurrentFile: vi.fn() }))
+    await expect(prepareJsonlMigration({ ...incomplete, verifyCurrentFile: vi.fn() }))
       .rejects.toThrow('complete frame contains a torn JSONL record')
 
     const tornRoot = await tempRoot()
@@ -405,11 +518,12 @@ describe('JSONL immutable generation publication', () => {
       await compressZstdFrame(line(header(0))),
       tornBody.subarray(0, -3),
     ]))
-    await ensureJsonlGenerationCurrent({
+    const recovered = await prepareJsonlMigration({
       ...torn,
       verifyCurrentFile: verifier(),
     })
-    expect((await decodeZstdJsonl(torn.currentPath)).trimEnd().split('\n')).toHaveLength(3)
+    expect(recovered.artifact.events).toEqual([event0, event1])
+    await recovered.publish()
 
     const emptyTailRoot = await tempRoot()
     const emptyTail = options(emptyTailRoot, 'zstd', streamingAdapter())
@@ -417,11 +531,12 @@ describe('JSONL immutable generation publication', () => {
       await compressZstdFrame(line(header(0))),
       tornBody.subarray(0, 8),
     ]))
-    await ensureJsonlGenerationCurrent({
+    const withoutTail = await prepareJsonlMigration({
       ...emptyTail,
       verifyCurrentFile: verifier(),
     })
-    expect((await decodeZstdJsonl(emptyTail.currentPath)).trimEnd().split('\n')).toHaveLength(1)
+    expect(withoutTail.artifact.events).toEqual([])
+    await withoutTail.publish()
   })
 
   it('checks migration and verification identities exactly', async () => {
@@ -441,13 +556,18 @@ describe('JSONL immutable generation publication', () => {
     const mismatchRoot = await tempRoot()
     const mismatch = options(mismatchRoot, 'none', streamingAdapter())
     await writeFile(mismatch.sourcePath, line(header(0)))
-    await expect(ensureJsonlGenerationCurrent({
+    const mismatched = await prepareJsonlMigration({
       ...mismatch,
       verifyCurrentFile: async (path, compression, expectedId, expectedEventCount) => ({
         ...await verifyJsonlCurrentGeneration(path, compression, expectedId, expectedEventCount),
         digest: 'different',
       }),
-    })).rejects.toThrow('changed during verification')
+    })
+    await expect(mismatched.publish()).rejects.toThrow('changed during verification')
+
+    const current = options(await tempRoot(), 'none', streamingAdapter(), 2)
+    await expect(prepareJsonlMigration({ ...current, verifyCurrentFile: vi.fn() }))
+      .rejects.toThrow('requires a historical source')
 
     const wrongRoot = await tempRoot()
     const wrongFormat = streamingAdapter()
@@ -464,7 +584,7 @@ describe('JSONL immutable generation publication', () => {
       }),
     })
     await writeFile(wrong.sourcePath, line(header(0)))
-    await expect(ensureJsonlGenerationCurrent({ ...wrong, verifyCurrentFile: vi.fn() }))
+    await expect(prepareJsonlMigration({ ...wrong, verifyCurrentFile: vi.fn() }))
       .rejects.toThrow('migration returned v0')
   })
 
@@ -478,12 +598,13 @@ describe('JSONL immutable generation publication', () => {
       sync: async () => {},
       close: async () => { throw new Error('close failed') },
     } as unknown as FileHandle
-    await expect(createJsonlGenerationTestRuntime({
+    const failedPreparation = await createJsonlGenerationTestRuntime({
       fs: { open: async () => failedHandle },
-    }).ensure({
+    }).prepare({
       ...failed,
       verifyCurrentFile: vi.fn(),
-    })).rejects.toBeInstanceOf(AggregateError)
+    })
+    await expect(failedPreparation.publish()).rejects.toBeInstanceOf(AggregateError)
   })
 
   it('propagates a streamed encoder failure through the Zstandard pipeline', async () => {
@@ -494,77 +615,26 @@ describe('JSONL immutable generation publication', () => {
     }))
     await writeFile(request.sourcePath, await encodeZstd(0, [event0]))
 
-    await expect(ensureJsonlGenerationCurrent({
+    const prepared = await prepareJsonlMigration({
       ...request,
       verifyCurrentFile: vi.fn(),
-    })).rejects.toBe(failure)
-    expect(await readdir(root)).toEqual(['session.jsonl.zstd'])
-  })
-
-  it('observes cancellation at the existing encode yield boundary', async () => {
-    const root = await tempRoot()
-    const controller = new AbortController()
-    const reason = new Error('cancelled during encoding')
-    const request = { ...options(root), signal: controller.signal }
-    const payload = 'x'.repeat(600 * 1024)
-    await writeFile(request.sourcePath, line(header(0)) + line({
-      ...event0, data: { turn: 1, payload },
-    }) + line({
-      ...event1, data: { turn: 1, reason: { kind: 'completed' }, payload },
-    }))
-    vi.spyOn(performance, 'now').mockReturnValue(0)
-    let yields = 0
-    vi.spyOn(scheduler, 'yield').mockImplementation(async () => {
-      yields += 1
-      if (yields === 2) controller.abort(reason)
     })
-
-    await expect(ensureJsonlGenerationCurrent({
-      ...request,
-      verifyCurrentFile: vi.fn(),
-    })).rejects.toBe(reason)
-    expect(yields).toBe(2)
-    expect(await readdir(root)).toEqual(['session.jsonl'])
-  })
-
-  it('forwards cancellation to staged verification', async () => {
-    const root = await tempRoot()
-    const controller = new AbortController()
-    const reason = new Error('cancelled during verification')
-    const request = { ...options(root), signal: controller.signal }
-    await writeFile(request.sourcePath, line(header(0)) + line(event0))
-    const verifyCurrentFile: EnsureJsonlGenerationOptions['verifyCurrentFile'] = async (
-      _path,
-      _compression,
-      _expectedId,
-      _expectedEventCount,
-      _expectedPrefix,
-      signal,
-    ) => {
-      expect(signal).toBe(controller.signal)
-      controller.abort(reason)
-      signal?.throwIfAborted()
-      throw new Error('unreachable')
-    }
-
-    await expect(ensureJsonlGenerationCurrent({
-      ...request,
-      verifyCurrentFile,
-    })).rejects.toBe(reason)
-    expect(await readdir(root)).toEqual(['session.jsonl'])
+    await expect(prepared.publish()).rejects.toBe(failure)
+    expect(await readdir(root)).toEqual(['session.jsonl.zstd'])
   })
 
-  it('publishes through the Windows no-overwrite path', async () => {
+  it('publishes a prepared stage through the Windows no-overwrite path', async () => {
     const winRoot = await tempRoot()
     const win = options(winRoot, 'none', streamingAdapter())
     await writeFile(win.sourcePath, line(header(0)))
-    await createJsonlGenerationTestRuntime({
+    const winPrepared = await createJsonlGenerationTestRuntime({
       platform: 'win32',
       publishNewWin32: rename,
-    }).ensure({
+    }).prepare({
       ...win,
       verifyCurrentFile: verifier(),
     })
+    await winPrepared.publish()
     expect(await readFile(win.currentPath, 'utf8')).toContain('"version":2')
   })
 
@@ -588,56 +658,6 @@ describe('JSONL immutable generation publication', () => {
     expect((await readdir(root)).sort()).toEqual(['session.jsonl', 'session.v2.jsonl'])
   })
 
-  it('takes the current fast path with one read and no format callback', async () => {
-    const root = await tempRoot()
-    const base = adapter()
-    const createRestore = vi.fn((value: Record<string, unknown>) => base.createRestore(value))
-    const encodeHeader = vi.fn((value: SessionFormatArtifact['header'], cut: number) =>
-      base.encodeHeader(value, cut))
-    const encodeEvent = vi.fn((value: SessionFormatArtifact['events'][number]) => base.encodeEvent(value))
-    const validateHistoricalHeader = vi.fn()
-    const request = {
-      ...options(root, 'none', { ...base, createRestore, encodeHeader, encodeEvent }, 2),
-      validateHistoricalHeader,
-    }
-    const contents = line({ ...header(2), isSeeded: false }) + line(event0)
-    await writeFile(request.sourcePath, contents)
-    const readStableFile = vi.fn(async (path: string, signal?: AbortSignal) =>
-      readFile(path, signal === undefined ? undefined : { signal }))
-
-    const result = await ensureWithOverrides(request, { fs: { readFile: readStableFile } })
-
-    expect(result).toMatchObject({ status: 'current', version: 2, path: request.sourcePath })
-    expect(readStableFile).toHaveBeenCalledOnce()
-    expect(createRestore).not.toHaveBeenCalled()
-    expect(encodeHeader).not.toHaveBeenCalled()
-    expect(encodeEvent).not.toHaveBeenCalled()
-    expect(validateHistoricalHeader).not.toHaveBeenCalled()
-    expect(await readFile(request.sourcePath, 'utf8')).toBe(contents)
-  })
-
-  it('bounds current snapshot retries under continuous revision churn', async () => {
-    const root = await tempRoot()
-    const request = options(root, 'none', adapter(), 2)
-    const contents = line(header(2)) + line(event0)
-    await writeFile(request.sourcePath, contents)
-    let revision = 0n
-    const statFile = vi.fn(async (path: string) => {
-      const value = await stat(path, { bigint: true })
-      revision += 1n
-      return { ...value, mtimeNs: value.mtimeNs + revision }
-    })
-    const readChangingFile = vi.fn(async () => Buffer.from(contents + line(event1)))
-
-    const result = await ensureWithOverrides(request, {
-      fs: { stat: statFile, readFile: readChangingFile },
-    })
-
-    expect(result.snapshot.bytes.toString('utf8')).toBe(contents)
-    expect(readChangingFile).toHaveBeenCalledTimes(2)
-    expect(statFile).toHaveBeenCalledTimes(3)
-  })
-
   it.each(['none', 'zstd'] as const)(
     'validates the selected %s historical header before invoking migration',
     async (compression) => {
@@ -700,24 +720,15 @@ describe('JSONL immutable generation publication', () => {
     expect(await readdir(root)).toEqual(['session.jsonl'])
   })
 
-  it('rejects malformed and future version discriminators before migration', async () => {
+  it('rejects a malformed version discriminator before migration', async () => {
     const root = await tempRoot()
     const malformed = options(join(root, 'malformed'))
-    const future = options(join(root, 'future'), 'none', adapter(), 3)
     await mkdir(join(root, 'malformed'))
-    await mkdir(join(root, 'future'))
     await writeFile(malformed.sourcePath, line(header(-1)))
-    await writeFile(future.sourcePath, line(header(3, 'future-id')))
 
     await expect(ensureJsonlGenerationCurrent(malformed)).rejects.toThrow(
       'header version is not a non-negative safe integer',
     )
-    await expect(ensureJsonlGenerationCurrent(future)).rejects.toMatchObject({
-      name: 'JsonlGenerationNewerVersionError',
-      storedVersion: 3,
-      currentVersion: 2,
-      storedId: 'future-id',
-    })
   })
 
   it.each([
@@ -976,7 +987,7 @@ describe('JSONL immutable generation publication', () => {
     }
   })
 
-  it('retries a bracketed physical read and a source changed before publication', async () => {
+  it('bounds a bracketed physical read and does not rerun migration after a publication race', async () => {
     const root = await tempRoot()
     const request = options(root)
     const first = Buffer.from(line(header(0)) + line(event0))
@@ -995,17 +1006,15 @@ describe('JSONL immutable generation publication', () => {
       if (phase === 'before-source-check' && attempt === 1) await writeFile(request.sourcePath, second)
     })
 
-    await ensureWithOverrides(
+    await expect(ensureWithOverrides(
       { ...request, format: { ...base, createRestore } },
       { fs: { stat: statFile }, barrier },
-    )
+    )).rejects.toBeInstanceOf(JsonlGenerationSourceChangedError)
 
     expect(stats).toBeGreaterThan(2)
-    expect(createRestore).toHaveBeenCalledTimes(2)
+    expect(createRestore).toHaveBeenCalledOnce()
     expect(await readFile(request.sourcePath)).toEqual(second)
-    expect(await readFile(request.currentPath, 'utf8')).toBe(
-      line(header(2)) + line(event0) + line(event1),
-    )
+    await expect(readFile(request.currentPath)).rejects.toMatchObject({ code: 'ENOENT' })
     expect((await readdir(root)).every(name => !name.includes('.tmp'))).toBe(true)
   })
 
@@ -1020,7 +1029,7 @@ describe('JSONL immutable generation publication', () => {
       if (phase === 'before-source-check' && attempt === 1) await writeFile(request.sourcePath, second)
     }
 
-    await expect(ensureWithOverrides(request, {
+    const failure = await ensureWithOverrides(request, {
       barrier,
       fs: {
         rm: async (path: string) => {
@@ -1028,7 +1037,10 @@ describe('JSONL immutable generation publication', () => {
           await rm(path, { force: true })
         },
       },
-    })).rejects.toBe(cleanup)
+    }).then(() => undefined, (error: unknown) => error)
+    if (!(failure instanceof AggregateError)) throw new Error('expected source and cleanup failures')
+    expect(failure.errors[0]).toBeInstanceOf(JsonlGenerationSourceChangedError)
+    expect(failure.errors[1]).toBe(cleanup)
     expect(await readFile(request.sourcePath)).toEqual(second)
     await expect(readFile(request.currentPath)).rejects.toMatchObject({ code: 'ENOENT' })
   })
@@ -1133,7 +1145,7 @@ describe('JSONL immutable generation publication', () => {
     const format = adapter()
     const request = {
       ...options(root, 'none', format),
-      verifyCurrentFile: async (...args: Parameters<EnsureJsonlGenerationOptions['verifyCurrentFile']>) => {
+      verifyCurrentFile: async (...args: Parameters<PrepareJsonlMigrationOptions['verifyCurrentFile']>) => {
         validations += 1
         if (validations === 2) throw 'non-error rejection'
         return verifier()(...args)
@@ -1175,7 +1187,26 @@ describe('JSONL immutable generation publication', () => {
     expect(await readFile(request.currentPath, 'utf8')).toBe(line(header(2)) + line(event0))
   })
 
-  it('reports cancellation during committed reopen and leaves the target', async () => {
+  it('retains a committed generation when its post-publication stat fails', async () => {
+    const root = await tempRoot()
+    const request = options(root)
+    const statFailure = new Error('published target stat failed')
+    await writeFile(request.sourcePath, line(header(0)) + line(event0))
+    let targetStats = 0
+
+    await expect(ensureWithOverrides(request, {
+      fs: {
+        stat: async (path) => {
+          if (path === request.currentPath && ++targetStats === 1) throw statFailure
+          return stat(path, { bigint: true })
+        },
+      },
+    })).rejects.toBe(statFailure)
+    expect(await readFile(request.currentPath, 'utf8')).toBe(line(header(2)) + line(event0))
+    expect((await readdir(root)).every(name => !name.includes('.tmp'))).toBe(true)
+  })
+
+  it('finishes a committed publication despite later caller cancellation', async () => {
     const root = await tempRoot()
     const controller = new AbortController()
     const reason = new Error('stop after publication')
@@ -1186,7 +1217,7 @@ describe('JSONL immutable generation publication', () => {
       barrier: (phase) => {
         if (phase === 'after-publication') controller.abort(reason)
       },
-    })).rejects.toBe(reason)
+    })).resolves.toMatchObject({ status: 'migrated', path: request.currentPath })
     expect(await readFile(request.currentPath, 'utf8')).toBe(line(header(2)) + line(event0))
   })
 
@@ -1219,7 +1250,7 @@ describe('JSONL immutable generation publication', () => {
     })).rejects.toMatchObject({ code: 'ENOENT', path: request.currentPath })
   })
 
-  it('reopens a target after exclusive publication', async () => {
+  it('does not reopen a target after exclusive publication', async () => {
     const root = await tempRoot()
     const request = options(root)
     await writeFile(request.sourcePath, line(header(0)) + line(event0))
@@ -1233,7 +1264,7 @@ describe('JSONL immutable generation publication', () => {
         },
       },
     })).resolves.toMatchObject({ status: 'migrated', path: request.currentPath })
-    expect(reads).toContain(request.currentPath)
+    expect(reads).not.toContain(request.currentPath)
     expect(await readFile(request.currentPath, 'utf8')).toBe(line(header(2)) + line(event0))
   })
 

packages/session/session-persistence-jsonl/tests/jsonl.spec.ts

@@ -4,6 +4,7 @@ import { Context } from '@deepseek-ai/cordis'
 import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat, symlink } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { dirname, join, relative, resolve } from 'node:path'
+import { scheduler } from 'node:timers/promises'
 import { SESSION_FORMAT_VERSION, SessionLogOffset, SessionSeq, SessionId } from '@deepseek-ai/dsh-session'
 import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
 import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
@@ -18,6 +19,7 @@ import {
 } from '../../session-persistence/tests/contract.ts'
 import { runLiveWritePathContract } from '../../session-persistence/tests/live-write-contract.ts'
 import { LIVE_WRITE_BATCH_MAX_DELAY_MS, type JsonlSessionHandle } from '../src/storage.ts'
+import { JsonlGenerationSourceChangedError } from '../src/generation.ts'
 import SessionStore from '@deepseek-ai/dsh-session'
 
 const statRace = vi.hoisted(() => ({
@@ -43,6 +45,21 @@ const readTally = vi.hoisted(() => ({
   enabled: false,
 }))
 
+const readFailure = vi.hoisted(() => ({
+  path: undefined as string | undefined,
+  error: undefined as Error | undefined,
+}))
+
+const pausedRead = vi.hoisted(() => ({
+  path: undefined as string | undefined,
+  active: false,
+  entered: undefined as (() => void) | undefined,
+  resume: undefined as Promise<void> | undefined,
+  release: undefined as (() => void) | undefined,
+  done: undefined as Promise<void> | undefined,
+  finished: undefined as (() => void) | undefined,
+}))
+
 vi.mock('node:fs/promises', async (importOriginal) => {
   const actual = await importOriginal<typeof import('node:fs/promises')>()
   return {
@@ -57,10 +74,25 @@ vi.mock('node:fs/promises', async (importOriginal) => {
       return { ...identity, mtimeNs: identity.mtimeNs + 1n }
     }) as typeof actual.stat,
     readFile: (async (...args: Parameters<typeof actual.readFile>) => {
-      if (readTally.enabled && typeof args[0] === 'string') {
-        readTally.bySuffix.set(args[0], (readTally.bySuffix.get(args[0]) ?? 0) + 1)
+      const path = typeof args[0] === 'string' ? args[0] : undefined
+      if (path === readFailure.path && readFailure.error !== undefined) throw readFailure.error
+      if (readTally.enabled && path !== undefined) {
+        readTally.bySuffix.set(path, (readTally.bySuffix.get(path) ?? 0) + 1)
+      }
+      if (path !== pausedRead.path || pausedRead.resume === undefined) {
+        return actual.readFile(...args)
+      }
+      const resume = pausedRead.resume
+      const finished = pausedRead.finished
+      pausedRead.active = true
+      pausedRead.entered?.()
+      await resume
+      try {
+        return await actual.readFile(...args)
+      } finally {
+        pausedRead.active = false
+        finished?.()
       }
-      return actual.readFile(...args)
     }) as typeof actual.readFile,
     readdir: (async (...args: Parameters<typeof actual.readdir>) => {
       if (String(args[0]) === readdirFailure.path && readdirFailure.error !== undefined) {
@@ -107,6 +139,27 @@ async function freshRoot(): Promise<string> {
   return dir
 }
 
+function pausePhysicalRead(path: string): {
+  readonly entered: Promise<void>
+  readonly finished: Promise<void>
+  release(): void
+} {
+  const entered = Promise.withResolvers<undefined>()
+  const resume = Promise.withResolvers<undefined>()
+  const finished = Promise.withResolvers<undefined>()
+  pausedRead.path = path
+  pausedRead.entered = () => { entered.resolve(undefined) }
+  pausedRead.resume = resume.promise
+  pausedRead.release = () => { resume.resolve(undefined) }
+  pausedRead.done = finished.promise
+  pausedRead.finished = () => { finished.resolve(undefined) }
+  return {
+    entered: entered.promise,
+    finished: finished.promise,
+    release: () => { resume.resolve(undefined) },
+  }
+}
+
 function rawLogPath(root: string, cwd: string | undefined, id: SessionId): string {
   return logPath(root, cwd, id, 'none')
 }
@@ -178,7 +231,7 @@ async function writeLog(persistence: SessionPersistence, m: SessionHeader, event
 async function readAll(persistence: SessionPersistence, id: SessionId): Promise<{ meta: SessionHeader; events: readonly SessionEvent[] }> {
   const handle = await persistence.open(id, 'read')
   try {
-    return { meta: handle.header, events: await handle.read() }
+    return { meta: handle.header, events: (await handle.read()).events }
   } finally {
     await handle.close()
   }
@@ -200,6 +253,18 @@ afterEach(async () => {
   statRace.mode = 'settle'
   readTally.bySuffix.clear()
   readTally.enabled = false
+  const pausedReadDone = pausedRead.active ? pausedRead.done : undefined
+  readFailure.path = undefined
+  readFailure.error = undefined
+  pausedRead.release?.()
+  await pausedReadDone
+  pausedRead.path = undefined
+  pausedRead.active = false
+  pausedRead.entered = undefined
+  pausedRead.resume = undefined
+  pausedRead.release = undefined
+  pausedRead.done = undefined
+  pausedRead.finished = undefined
   statFailure.path = undefined
   statFailure.error = undefined
   readdirFailure.path = undefined
@@ -451,6 +516,17 @@ describe('JsonlSessionPersistence: stored-format refusals', () => {
     expect(await ctx.sessionPersistence.list()).toEqual([])
   })
 
+  it('classifies an unparsable future generation header as corruption', async () => {
+    const id = SessionId('future-malformed')
+    const path = generationLogPath(root, '/work', id, 42, 'none')
+    await mkdir(dirname(path), { recursive: true })
+    await writeFile(path, '{not-json}\n')
+
+    await expect(ctx.sessionPersistence.open(id, 'read')).rejects.toMatchObject({
+      name: 'SessionPersistenceCorruptionError',
+    })
+  })
+
   it('refuses a well-shaped newer-version header at read open with the upgrade direction', async () => {
     // A header that satisfies the current shape but carries a future version:
     // stat can parse it, and the open still refuses before handing out a
@@ -524,7 +600,14 @@ describe('JsonlSessionPersistence: stored-format refusals', () => {
     const handle = await ctx.sessionPersistence.open(m.id, 'read', { signal: new AbortController().signal })
     try {
       expect(handle.header).toMatchObject({ id: m.id, cwd: '/work' })
-      expect(await handle.read()).toEqual(oneTurnLog())
+      const read = await handle.read()
+      expect(read.eventState).toBe('shared-frozen')
+      expect(read.events).toEqual(oneTurnLog())
+      expect(read.events.every(event => Object.isFrozen(event) && Object.isFrozen(event.data))).toBe(true)
+      const reread = await handle.read()
+      expect(reread.events).not.toBe(read.events)
+      expect(reread.events[0]).toBe(read.events[0])
+      expect((await handle.read(read.events.length)).eventState).toBe('shared-frozen')
     } finally {
       await handle.close()
     }
@@ -597,7 +680,7 @@ describe('JsonlSessionPersistence: immutable format generations', () => {
     await expect(stat(currentPath)).rejects.toMatchObject({ code: 'ENOENT' })
   })
 
-  it('publishes v2 beside an unchanged v0 source before returning a read handle', async () => {
+  it('serves a migrated v0 read without publishing at the service durability barrier', async () => {
     const header = meta('released-v0-read', '/work')
     const sourcePath = historicalLogPath(root, header.cwd, header.id)
     const currentPath = rawLogPath(root, header.cwd, header.id)
@@ -607,18 +690,147 @@ describe('JsonlSessionPersistence: immutable format generations', () => {
     await mkdir(dirname(sourcePath), { recursive: true })
     await writeFile(sourcePath, source)
 
-    await expect(readAll(ctx.sessionPersistence, header.id)).resolves.toEqual({
+    const restored = await readAll(ctx.sessionPersistence, header.id)
+    expect(restored).toEqual({
       meta: { ...header, delegationDepth: 0 },
       events: oneTurnLog(),
     })
+    const userMessage = restored.events.find(event => event.type === 'user/message')
+    expect(userMessage).toBeDefined()
+    expect(Object.isFrozen(userMessage?.data)).toBe(true)
     expect(await readFile(sourcePath)).toEqual(source)
-    const current = (await readFile(currentPath, 'utf8')).trimEnd().split('\n')
-    expect(JSON.parse(current[0] as string)).toMatchObject({
-      id: header.id,
-      version: SESSION_FORMAT_VERSION,
-    })
+    await expect(readFile(currentPath)).rejects.toMatchObject({ code: 'ENOENT' })
     expect((await readdir(dirname(sourcePath))).filter(name => name.startsWith('session')).sort())
-      .toEqual(['session.jsonl', 'session.v2.jsonl'])
+      .toEqual(['session.jsonl'])
+  })
+
+  it('resolves absent, current, and historical current-generation paths', async () => {
+    const persistence = ctx.sessionPersistence as JsonlSessionPersistence
+    expect(await persistence.resolveCurrentLog(SessionId('missing-generation'))).toBeUndefined()
+
+    const current = meta('resolved-current', '/work')
+    const currentHandle = await ctx.sessionPersistence.create(current)
+    await currentHandle.flush()
+    await currentHandle.close()
+    await expect(persistence.resolveCurrentLog(current.id)).resolves.toBe(rawLogPath(root, current.cwd, current.id))
+
+    const historical = meta('resolved-historical', '/work')
+    const sourcePath = historicalLogPath(root, historical.cwd, historical.id)
+    await mkdir(dirname(sourcePath), { recursive: true })
+    await writeFile(sourcePath, `${JSON.stringify(releasedV0Header(historical))}\n`)
+    await expect(persistence.resolveCurrentLog(historical.id)).resolves.toBeUndefined()
+
+    const future = meta('resolved-future', '/work')
+    const futurePath = generationLogPath(root, future.cwd, future.id, SESSION_FORMAT_VERSION + 1, 'none')
+    await mkdir(dirname(futurePath), { recursive: true })
+    await writeFile(futurePath, `${JSON.stringify({ ...toHeaderLine(future), version: SESSION_FORMAT_VERSION + 1 })}\n`)
+    await expect(persistence.resolveCurrentLog(future.id)).rejects.toMatchObject({
+      name: 'SessionFormatUnsupportedError',
+    })
+  })
+
+  it('singleflights concurrent historical reads and keeps service flush read-only', async () => {
+    const header = meta('released-v0-source-drift', '/work')
+    const sourcePath = historicalLogPath(root, header.cwd, header.id)
+    await mkdir(dirname(sourcePath), { recursive: true })
+    await writeFile(sourcePath, `${JSON.stringify(releasedV0Header(header))}\n`)
+    readTally.enabled = true
+
+    const [first, second] = await Promise.all([
+      ctx.sessionPersistence.open(header.id, 'read'),
+      ctx.sessionPersistence.open(header.id, 'read'),
+    ])
+    expect((await first.read()).events).toEqual([])
+    expect((await second.read()).events).toEqual([])
+    expect(readTally.bySuffix.get(sourcePath)).toBe(1)
+    await appendFile(sourcePath, '\n')
+
+    await expect(ctx.sessionPersistence.flush()).resolves.toBeUndefined()
+    await expect(stat(rawLogPath(root, header.cwd, header.id))).rejects.toMatchObject({ code: 'ENOENT' })
+    await Promise.all([first.close(), second.close()])
+    await ctx.fiber.dispose()
+    ctx = new Context()
+  })
+
+  it('does not join an in-flight historical preparation for an older source revision', async () => {
+    const header = meta('released-v0-revision-singleflight', '/work')
+    const sourcePath = historicalLogPath(root, header.cwd, header.id)
+    await mkdir(dirname(sourcePath), { recursive: true })
+    await writeFile(sourcePath, `${JSON.stringify(releasedV0Header(header))}\n`)
+    const pause = pausePhysicalRead(sourcePath)
+    readTally.enabled = true
+
+    const firstOpening = ctx.sessionPersistence.open(header.id, 'read')
+    await pause.entered
+    await appendFile(sourcePath, `${eventLines(releasedV1OneTurnLog())}\n`)
+    const secondOpening = ctx.sessionPersistence.open(header.id, 'read')
+    let tallyFailure: unknown
+    try {
+      await vi.waitFor(() => { expect(readTally.bySuffix.get(sourcePath)).toBe(2) })
+    } catch (error: unknown) {
+      tallyFailure = error
+    } finally {
+      pause.release()
+    }
+
+    const [first, second] = await Promise.all([firstOpening, secondOpening])
+    try {
+      if (tallyFailure !== undefined) throw tallyFailure
+      expect((await first.read()).events).toEqual(oneTurnLog())
+      expect((await second.read()).events).toEqual(oneTurnLog())
+    } finally {
+      await Promise.all([first.close(), second.close()])
+    }
+  })
+
+  it('lets one historical-open caller abort without cancelling another waiter', async () => {
+    const header = meta('released-v0-shared-cancellation', '/work')
+    const sourcePath = historicalLogPath(root, header.cwd, header.id)
+    await mkdir(dirname(sourcePath), { recursive: true })
+    await writeFile(sourcePath, `${JSON.stringify(releasedV0Header(header))}\n`)
+    const pause = pausePhysicalRead(sourcePath)
+    readTally.enabled = true
+    const controller = new AbortController()
+    const reason = new Error('first historical waiter cancelled')
+
+    const first = ctx.sessionPersistence.open(header.id, 'read', { signal: controller.signal })
+    const second = ctx.sessionPersistence.open(header.id, 'read')
+    await pause.entered
+    await scheduler.yield()
+    controller.abort(reason)
+    await expect(first).rejects.toBe(reason)
+    pause.release()
+    const handle = await second
+    expect((await handle.read()).events).toEqual([])
+    expect(readTally.bySuffix.get(sourcePath)).toBe(1)
+    await handle.close()
+  })
+
+  it('cancels shared historical preparation after its last waiter leaves', async () => {
+    const header = meta('released-v0-last-waiter-cancellation', '/work')
+    const sourcePath = historicalLogPath(root, header.cwd, header.id)
+    await mkdir(dirname(sourcePath), { recursive: true })
+    await writeFile(sourcePath, `${JSON.stringify(releasedV0Header(header))}\n`)
+    const pause = pausePhysicalRead(sourcePath)
+    readTally.enabled = true
+    const controller = new AbortController()
+    const reason = 'last historical waiter cancelled'
+
+    const opening = ctx.sessionPersistence.open(header.id, 'read', { signal: controller.signal })
+    await pause.entered
+    controller.abort(reason)
+    await expect(opening).rejects.toMatchObject({
+      message: 'session migration preparation aborted',
+      cause: reason,
+    })
+    pause.release()
+    await pause.finished
+    await scheduler.yield()
+
+    const retried = await ctx.sessionPersistence.open(header.id, 'read')
+    expect((await retried.read()).events).toEqual([])
+    expect(readTally.bySuffix.get(sourcePath)).toBe(2)
+    await retried.close()
   })
 
   it('migrates released-v0 retry, repeated-compaction, provenance, and late-title shapes', async () => {
@@ -648,16 +860,15 @@ describe('JsonlSessionPersistence: immutable format generations', () => {
     expect(titleBlock).toMatchObject({ type: 'text' })
     if (titleBlock?.type !== 'text') throw new Error('fixture title request lacks its text block')
     expect(titleBlock.text).toContain('{"seq":21,"text":"late"}')
-    const currentRows = (await readFile(currentPath, 'utf8')).trimEnd().split('\n')
-      .map(line => JSON.parse(line) as Record<string, unknown>)
-    expect(currentRows.find(row => row['type'] === 'user/message'
-      && (row['data'] as { source?: { plugin?: string } }).source?.plugin === 'compact'))
+    expect(restored.events.find(event => event.type === 'user/message'
+      && (event.data as { source?: { plugin?: string } }).source?.plugin === 'compact'))
       .toMatchObject({
         seq: 14,
         sourceEventSeqs: [12, 13, 11, 2, 3, 4],
         surfaceOp: { op: 'replace', start: 11, end: 4 },
       })
     expect(await readFile(sourcePath)).toEqual(source)
+    await expect(readFile(currentPath)).rejects.toMatchObject({ code: 'ENOENT' })
   })
 
   it('publishes v2 beside an unchanged physical v1 source with packed chunk rows', async () => {
@@ -677,10 +888,9 @@ describe('JsonlSessionPersistence: immutable format generations', () => {
       .toMatchObject({ data: { message: { content: [{ type: 'text', text: 'hello' }] } } })
 
     expect(await readFile(sourcePath)).toEqual(source)
-    expect(JSON.parse((await readFile(currentPath, 'utf8')).split('\n')[0] as string))
-      .toMatchObject({ id: header.id, version: SESSION_FORMAT_VERSION })
+    await expect(readFile(currentPath)).rejects.toMatchObject({ code: 'ENOENT' })
     expect((await readdir(dirname(sourcePath))).filter(name => name.startsWith('session')).sort())
-      .toEqual(['session.v1.jsonl', 'session.v2.jsonl'])
+      .toEqual(['session.v1.jsonl'])
   })
 
   it('selects v1 from a v0/v1 directory, then v2 from the retained three-generation set', async () => {
@@ -695,8 +905,12 @@ describe('JsonlSessionPersistence: immutable format generations', () => {
 
     const migrated = await readAll(ctx.sessionPersistence, header.id)
     expect(migrated.events.map(event => event.type)).toContain('assistant/message')
+    const writer = await ctx.sessionPersistence.open(header.id, 'write')
+    await writer.close()
     expect((await readdir(directory)).filter(name => name.startsWith('session')).sort())
-      .toEqual(['session.jsonl', 'session.v1.jsonl', 'session.v2.jsonl'])
+      .toEqual(process.platform === 'win32'
+        ? ['session.jsonl', 'session.v1.jsonl', 'session.v2.jsonl']
+        : ['session.jsonl', 'session.lock', 'session.v1.jsonl', 'session.v2.jsonl'])
 
     await writeFile(v0Path, 'corrupt lower v0\n')
     await writeFile(v1Path, 'corrupt lower v1\n')
@@ -704,18 +918,17 @@ describe('JsonlSessionPersistence: immutable format generations', () => {
     expect(await readFile(v2Path, 'utf8')).toContain('"version":2')
   })
 
-  it('uses the same migration path for a handle storage resolution', async () => {
+  it('does not publish a historical generation through handle storage resolution', async () => {
     const header = meta('released-v0-handle-read', '/work')
     const sourcePath = historicalLogPath(root, header.cwd, header.id)
     const currentPath = rawLogPath(root, header.cwd, header.id)
     await mkdir(dirname(sourcePath), { recursive: true })
     await writeFile(sourcePath, `${JSON.stringify(releasedV0Header(header))}\n`)
-    const storage = ctx.sessionPersistence as unknown as {
-      resolveLog(id: SessionId, signal?: AbortSignal): Promise<string | undefined>
-    }
+    const persistence = ctx.sessionPersistence as JsonlSessionPersistence
 
-    await expect(storage.resolveLog(header.id, new AbortController().signal)).resolves.toBe(currentPath)
+    await expect(persistence.resolveCurrentLog(header.id, new AbortController().signal)).resolves.toBeUndefined()
     expect(await readFile(sourcePath, 'utf8')).toBe(`${JSON.stringify(releasedV0Header(header))}\n`)
+    await expect(readFile(currentPath)).rejects.toMatchObject({ code: 'ENOENT' })
   })
 
   it('opens the migrated successor for append while retaining the historical source', async () => {
@@ -740,6 +953,68 @@ describe('JsonlSessionPersistence: immutable format generations', () => {
     ])
   })
 
+  it('switches an existing prepared read handle to the published append tail', async () => {
+    const header = meta('released-v0-read-handoff', '/work')
+    const sourcePath = historicalLogPath(root, header.cwd, header.id)
+    await mkdir(dirname(sourcePath), { recursive: true })
+    await writeFile(
+      sourcePath,
+      `${JSON.stringify(releasedV0Header(header))}\n${eventLines(releasedV1OneTurnLog())}\n`,
+    )
+    const reader = await ctx.sessionPersistence.open(header.id, 'read')
+    const suffix: SessionEvent[] = [
+      { type: 'turn/start', seq: SessionSeq(6), time: 9, data: { turn: 2 } },
+      { type: 'turn/end', seq: SessionSeq(7), time: 10, data: { turn: 2, reason: { kind: 'completed' } } },
+    ]
+    try {
+      expect((await reader.read()).events).toEqual(oneTurnLog())
+      await appendBatch(ctx.sessionPersistence, header.id, suffix)
+      expect((await reader.read()).events).toEqual([...oneTurnLog(), ...suffix])
+    } finally {
+      await reader.close()
+    }
+  })
+
+  it('fails a stale prepared publication once and re-prepares on the next write open', async () => {
+    const header = meta('released-v0-write-source-drift', '/work')
+    const sourcePath = historicalLogPath(root, header.cwd, header.id)
+    const currentPath = rawLogPath(root, header.cwd, header.id)
+    const source = `${JSON.stringify(releasedV0Header(header))}\n`
+    await mkdir(dirname(sourcePath), { recursive: true })
+    await writeFile(sourcePath, source)
+    await readAll(ctx.sessionPersistence, header.id)
+    vi.spyOn(scheduler, 'yield').mockImplementationOnce(async () => {
+      await appendFile(sourcePath, '\n')
+    })
+
+    await expect(ctx.sessionPersistence.open(header.id, 'write'))
+      .rejects.toBeInstanceOf(JsonlGenerationSourceChangedError)
+    await expect(readFile(currentPath)).rejects.toMatchObject({ code: 'ENOENT' })
+
+    const writer = await ctx.sessionPersistence.open(header.id, 'write')
+    await writer.close()
+    expect(await readFile(sourcePath, 'utf8')).toBe(`${source}\n`)
+    expect(await readFile(currentPath, 'utf8')).toContain('"version":2')
+  })
+
+  it('finishes publication before rejecting a write open cancelled during publication', async () => {
+    const header = meta('released-v0-publication-cancellation', '/work')
+    const sourcePath = historicalLogPath(root, header.cwd, header.id)
+    const currentPath = rawLogPath(root, header.cwd, header.id)
+    await mkdir(dirname(sourcePath), { recursive: true })
+    await writeFile(sourcePath, `${JSON.stringify(releasedV0Header(header))}\n`)
+    await readAll(ctx.sessionPersistence, header.id)
+    const controller = new AbortController()
+    const reason = new Error('write open cancelled during publication')
+    vi.spyOn(scheduler, 'yield').mockImplementationOnce(async () => { controller.abort(reason) })
+
+    await expect(ctx.sessionPersistence.open(header.id, 'write', { signal: controller.signal }))
+      .rejects.toBe(reason)
+    expect(await readFile(currentPath, 'utf8')).toContain('"version":2')
+    const writer = await ctx.sessionPersistence.open(header.id, 'write')
+    await writer.close()
+  })
+
   it('treats a historical generation as an existing id at create', async () => {
     const header = meta('released-v0-collision', '/work')
     const sourcePath = historicalLogPath(root, header.cwd, header.id)
@@ -822,6 +1097,12 @@ describe('JsonlSessionPersistence: immutable format generations', () => {
     await expect(ctx.sessionPersistence.open(header.id, 'read')).rejects.toMatchObject({ code: 'EACCES' })
     statFailure.error = new DOMException('source read aborted', 'AbortError')
     await expect(ctx.sessionPersistence.open(header.id, 'read')).rejects.toMatchObject({ name: 'AbortError' })
+    statFailure.error = undefined
+    readFailure.path = sourcePath
+    readFailure.error = new DOMException('source read failed', 'InvalidStateError')
+    await expect(ctx.sessionPersistence.open(header.id, 'read')).rejects.toMatchObject({
+      name: 'SessionPersistenceCorruptionError',
+    })
   })
 
   it('selects the highest opposite-encoding generation for its refusal', async () => {
@@ -930,6 +1211,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => {
   it('lazy materialization: create() writes no file until the first append', async () => {
     const m = meta('lazy', '/work')
     const handle = await ctx.sessionPersistence.create(m)
+    expect((await handle.read()).eventState).toBe('detached')
     // create() materializes no file before the first append — while the
     // created session is already visible to this process.
     const dir = sessionDir(root, '/work', m.id)
@@ -951,7 +1233,10 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => {
     await handle.close()
 
     expect(await readFile(rawLogPath(root, '/work', m.id), 'utf8')).toBe(`${JSON.stringify(toHeaderLine(m))}\n`)
-    await expect(readAll(ctx.sessionPersistence, m.id)).resolves.toMatchObject({ events: [] })
+    const reader = await ctx.sessionPersistence.open(m.id, 'read')
+    const read = await reader.read()
+    expect(read).toEqual({ eventState: 'shared-frozen', events: [] })
+    await reader.close()
   })
 
   it('close drains a routed event that arrives while it waits for an in-flight append', async () => {
@@ -1062,7 +1347,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => {
     // ...and the immediate write-open (resume) reuses the parsed log through
     // the revision guard instead of re-reading the file.
     const writer = await ctx.sessionPersistence.open(m.id, 'write')
-    expect((await writer.read()).length).toBe(oneTurnLog().length)
+    expect((await writer.read()).events.length).toBe(oneTurnLog().length)
     expect(readTally.bySuffix.get(path)).toBe(1)
 
     // A local append invalidates the memo: the next cold read re-parses and
@@ -1122,7 +1407,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => {
       const internals = ctx.sessionPersistence as unknown as { coldLogMemo: Map<SessionId, unknown> }
       internals.coldLogMemo.clear()
       statRace.path = rawLogPath(root, '/work', m.id)
-      expect(await handle.read()).toEqual(oneTurnLog())
+      expect((await handle.read()).events).toEqual(oneTurnLog())
       // The memo probe, the initial identity, the mismatching post-read stat
       // (reused as the retry's pre-read identity), and the retry's matching
       // post-read stat.
@@ -1146,7 +1431,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => {
       // serves the retry's pre-read committed prefix — here the whole log.
       // Four stats: the memo probe, the initial identity, and one mismatching
       // post-read stat per bounded attempt.
-      expect(await handle.read()).toEqual(oneTurnLog())
+      expect((await handle.read()).events).toEqual(oneTurnLog())
       expect(statRace.reads).toBe(4)
     } finally {
       statRace.path = undefined
@@ -1300,7 +1585,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => {
 
       // The retry now succeeds with NO seq gap — the log is contiguous 0..7.
       await handle.append(turn2)
-      expect((await handle.read()).map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
+      expect((await handle.read()).events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
     } finally {
       await handle.close()
     }
@@ -1412,7 +1697,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => {
       ], { signal })).rejects.toBe(reason)
       await expect(handle.flush({ signal })).rejects.toBe(reason)
       // The aborted mutations left the log untouched.
-      expect((await handle.read()).map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5])
+      expect((await handle.read()).events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5])
     } finally {
       await handle.close()
     }
@@ -1473,7 +1758,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => {
     await handle.append([])
     await expect(stat(rawLogPath(root, '/work', m.id))).rejects.toThrow()
     await handle.append(oneTurnLog())
-    expect((await handle.read()).map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5])
+    expect((await handle.read()).events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5])
     await handle.close()
   })
 
@@ -1498,7 +1783,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => {
     const m = meta('erased-pending')
     const creator = await ctx.sessionPersistence.create(m)
     const reader = await ctx.sessionPersistence.open(m.id, 'read')
-    expect(await reader.read()).toEqual([])
+    expect((await reader.read()).events).toEqual([])
     // The creator closes without ever appending: the session never existed.
     await creator.close()
     await expect(reader.read()).rejects.toThrow(/not found/)
@@ -1510,7 +1795,7 @@ describe('JsonlSessionPersistence: durability and crash semantics', () => {
     await writeLog(ctx.sessionPersistence, m, oneTurnLog())
     const reader = await ctx.sessionPersistence.open(m.id, 'read')
     try {
-      expect(await reader.read()).toHaveLength(6)
+      expect((await reader.read()).events).toHaveLength(6)
       // Committed events are never rewritten; a shorter file is damage, not a
       // legal state, and a handle must not silently backtrack.
       await writeFile(rawLogPath(root, '/work', m.id), [

packages/session/session-persistence-jsonl/tests/lease.spec.ts

@@ -181,7 +181,7 @@ describe('cross-process write lock', () => {
     await pendingWinner.close()
     // Reads never touch the lock.
     const reader = await second.open(SessionId('excluded'), 'read')
-    expect((await reader.read()).map(event => event.seq)).toEqual([0, 1])
+    expect((await reader.read()).events.map(event => event.seq)).toEqual([0, 1])
     await reader.close()
 
     await holder.close()

packages/session/session-persistence-jsonl/tests/lease.two-process.e2e.ts

@@ -51,7 +51,7 @@ describe('two-process write lock (built lib)', () => {
       await expect(mine.open(SessionId(SESSION), 'write')).rejects.toBeInstanceOf(SessionAlreadyOwnedError)
       // Reads are unaffected across processes.
       const reader = await mine.open(SessionId(SESSION), 'read')
-      expect((await reader.read()).map(event => event.seq)).toEqual([0, 1])
+      expect((await reader.read()).events.map(event => event.seq)).toEqual([0, 1])
       await reader.close()
 
       // Crash the holder: no release runs, but the kernel drops the lock with
@@ -60,7 +60,7 @@ describe('two-process write lock (built lib)', () => {
       await exited
       const taken = await mine.open(SessionId(SESSION), 'write')
       await taken.append([{ type: 'turn/start', seq: SessionSeq(2), time: 3, data: { turn: 2 } }])
-      expect((await taken.read()).map(event => event.seq)).toEqual([0, 1, 2])
+      expect((await taken.read()).events.map(event => event.seq)).toEqual([0, 1, 2])
       await taken.close()
     } finally {
       if (holder.exitCode === null) holder.kill('SIGKILL')

packages/session/session-persistence-jsonl/tests/migration-verifier.spec.ts

@@ -48,10 +48,12 @@ afterEach(() => {
 
 describe('migration verifier Worker lifecycle', () => {
   it('resolves only after terminating a successful Worker', async () => {
-    const verification = verifyCurrentGenerationInWorker('/stage', 'none', 'session', 2)
+    const expectedPrefix = { bytes: 3, digest: 'a'.repeat(64) }
+    const verification = verifyCurrentGenerationInWorker('/stage', 'none', 'session', 2, expectedPrefix)
     const instance = worker()
     expect(instance.options.workerData).toEqual({
       path: '/stage', compression: 'none', expectedId: 'session', expectedEventCount: 2,
+      expectedPrefix,
     })
     instance.emit('message', { ok: true, result })
 

packages/session/session-persistence-jsonl/tests/zstd.spec.ts

@@ -5,7 +5,7 @@ import type { FileHandle } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import { performance } from 'node:perf_hooks'
-import { SESSION_FORMAT_VERSION, SessionSeq, SessionId } from '@deepseek-ai/dsh-session'
+import { SessionSeq, SessionId } from '@deepseek-ai/dsh-session'
 import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
 import type { SessionPersistence } from '@deepseek-ai/dsh-session-persistence'
 import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl'
@@ -68,7 +68,7 @@ async function writeLog(persistence: SessionPersistence, m: SessionHeader, event
 async function readAll(persistence: SessionPersistence, id: SessionId): Promise<{ meta: SessionHeader; events: readonly SessionEvent[] }> {
   const handle = await persistence.open(id, 'read')
   try {
-    return { meta: handle.header, events: await handle.read() }
+    return { meta: handle.header, events: (await handle.read()).events }
   } finally {
     await handle.close()
   }
@@ -411,7 +411,7 @@ describe('JsonlSessionPersistence: default Zstandard encoding', () => {
     expect((await readAll(ctx.sessionPersistence, header.id)).events).toEqual(oneTurnLog())
   })
 
-  it('publishes v2 beside an unchanged compressed v0 source before returning a read handle', async () => {
+  it('serves a migrated compressed v0 read without publishing a successor', async () => {
     const root = await freshRoot()
     const ctx = await mount(root)
     const header = meta('zstd-v0-read', '/work')
@@ -429,11 +429,7 @@ describe('JsonlSessionPersistence: default Zstandard encoding', () => {
       events: oneTurnLog(),
     })
     expect(await readFile(sourcePath)).toEqual(source)
-    const current = (await decodeCompleteFrames(await readFile(currentPath))).toString().split('\n')
-    expect(JSON.parse(current[0] as string)).toMatchObject({
-      id: header.id,
-      version: SESSION_FORMAT_VERSION,
-    })
+    await expect(readFile(currentPath)).rejects.toMatchObject({ code: 'ENOENT' })
   })