Merge commit '1f9d5f72e9' into worktree/session-format-05-v1-v2-chunk-migration

# Conflicts:
#	packages/session/session-format/README.i18n.yaml
#	packages/session/session-format/README.md
#	packages/session/session-format/README.zh.md
#	packages/session/session-format/src/catalog.ts
#	packages/session/session-format/src/types.ts
#	packages/test-support/llm-replay/tests/llm-replay.spec.ts

packages/session/session-format/README.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write packages/session/session-format/README.md
-README.md: 2146fa0aceac6baaa5061adee6f8d4502cef2d76
-README.zh.md: b8f048726aa8ad26420a0d8f0ffeee837475576b
+README.md: cd95942128531b2a8b16b7ffd29ac42753d9b889
+README.zh.md: 812874b8b8966c6a0f1297b024601c4dce36fa96

packages/session/session-format/README.md

@@ -36,7 +36,7 @@ const catalog = createSessionFormatCatalog({ currentVersion, codecs, encodeCurre
 const descriptor = catalog.readHeader(physicalHeader)
 ```
 
-`createSessionFormatCatalog()` accepts one frozen decoder per supported version, the current format's encoder, one migration per adjacent version pair, and current artifact and header restorers. `inspectVersion()` reads only the physical version for directional dispatch. `readHeader()` returns a `current`, `migration-required`, `unsupported`, or `malformed` descriptor without reading events. Each edge validates its target header before the final current-header restorer runs. Body readers call `decodeArtifact()` or `decodeRecoverableArtifact()`, then `migrate()`; writers call `encodeCurrent()` only with a validated current artifact. Frozen v0/v1 codec exports retain their format-specific `packChunks` option without adding that historical control to the current writer or common decoder interface.
+`createSessionFormatCatalog()` accepts one frozen decoder per supported version, the current format's encoder, one migration per adjacent version pair, and current artifact and header restorers. `readHeader()` returns a `current`, `migration-required`, `unsupported`, or `malformed` descriptor without reading events. Each edge validates its target header before the final current-header restorer runs. Body readers call `decodeArtifact()` or `decodeRecoverableArtifact()`, then `migrate()`; writers call `encodeCurrent()` only with a validated current artifact. Frozen v0/v1 codec exports retain their format-specific `packChunks` option without adding that historical control to the current writer or common decoder interface.
 
 The recoverable decoder returns the accepted logical prefix. A codec may drop one malformed or sequence-gapped row and its uncommitted suffix, but a later decoded `turn/end` makes the original issue fatal.
 

packages/session/session-format/README.zh.md

@@ -36,7 +36,7 @@ const catalog = createSessionFormatCatalog({ currentVersion, codecs, encodeCurre
 const descriptor = catalog.readHeader(physicalHeader)
 ```
 
-`createSessionFormatCatalog()` 接收每个受支持版本的一个冻结解码器、当前格式的编码器、每组相邻版本的一个迁移，以及当前产物与标头还原器。`inspectVersion()` 只读取物理版本以执行方向分派。`readHeader()` 在不读取事件的情况下返回 `current`、`migration-required`、`unsupported` 或 `malformed` 描述符。每个迁移边会先校验自己的目标标头，然后再运行最终的当前标头还原器。正文读取方调用 `decodeArtifact()` 或 `decodeRecoverableArtifact()`，然后调用 `migrate()`；写入方只使用经过校验的当前产物调用 `encodeCurrent()`。冻结的 v0/v1 编解码器导出会保留其格式专用的 `packChunks` 选项，但不会把这项历史控制加入当前 writer 或通用解码器接口。
+`createSessionFormatCatalog()` 接收每个受支持版本的一个冻结解码器、当前格式的编码器、每组相邻版本的一个迁移，以及当前产物与标头还原器。`readHeader()` 在不读取事件的情况下返回 `current`、`migration-required`、`unsupported` 或 `malformed` 描述符。每个迁移边会先校验自己的目标标头，然后再运行最终的当前标头还原器。正文读取方调用 `decodeArtifact()` 或 `decodeRecoverableArtifact()`，然后调用 `migrate()`；写入方只使用经过校验的当前产物调用 `encodeCurrent()`。冻结的 v0/v1 编解码器导出会保留其格式专用的 `packChunks` 选项，但不会把这项历史控制加入当前 writer 或通用解码器接口。
 
 可恢复解码器返回已接受的逻辑前缀。编解码器可以丢弃一个格式错误或序号不连续的行及其未提交后缀，但后续成功解码的 `turn/end` 会使原始问题成为致命错误。
 

packages/session/session-format/src/catalog.ts

@@ -37,14 +37,10 @@ export function createSessionFormatCatalog(options: SessionFormatCatalogOptions)
     throw new SessionFormatError(`Session format codec v${invalid} is newer than current v${chain.currentVersion}`)
   }
 
-  function inspectVersion(headerValue: unknown): number {
-    return inspectSessionFormatVersion(headerValue)
-  }
-
   function readHeader(headerValue: unknown): SessionFormatHeaderReadResult {
     let storedVersion: number | undefined
     try {
-      storedVersion = inspectVersion(headerValue)
+      storedVersion = inspectSessionFormatVersion(headerValue)
     } catch (error: unknown) {
       return malformed(chain.currentVersion, error)
     }
@@ -92,7 +88,7 @@ export function createSessionFormatCatalog(options: SessionFormatCatalogOptions)
     readonly storedVersion: number
     readonly codec: SessionFormatCodec
   } {
-    const storedVersion = inspectVersion(headerValue)
+    const storedVersion = inspectSessionFormatVersion(headerValue)
     if (storedVersion > chain.currentVersion) {
       throw new SessionFormatUnsupportedMigrationError(
         `stored Session uses newer format v${storedVersion}; this build writes v${chain.currentVersion}`,
@@ -128,8 +124,7 @@ export function createSessionFormatCatalog(options: SessionFormatCatalogOptions)
     if (inspectSessionFormatVersion(artifact.header) !== chain.currentVersion) {
       throw new SessionFormatError(`encodeCurrent requires Session format v${chain.currentVersion}`)
     }
-    const current = chain.migrate(artifact)
-    const encoded = options.encodeCurrentArtifact(current)
+    const encoded = options.encodeCurrentArtifact(artifact)
     const header = snapshotSessionFormatJson(encoded.header, 'encoded current Session header') as SessionFormatJsonObject
     const rows = Object.freeze(encoded.rows.map((row, index) =>
       snapshotSessionFormatJson(row, `encoded current Session row ${index}`) as SessionFormatJsonObject))
@@ -141,7 +136,6 @@ export function createSessionFormatCatalog(options: SessionFormatCatalogOptions)
 
   return Object.freeze({
     currentVersion: chain.currentVersion,
-    inspectVersion,
     readHeader,
     decodeArtifact,
     decodeRecoverableArtifact,

packages/session/session-format/src/types.ts

@@ -134,8 +134,6 @@ export interface SessionFormatCatalogOptions extends SessionFormatChainOptions {
 /** Build-static physical dispatch and adjacent migration catalog. */
 export interface SessionFormatCatalog {
   readonly currentVersion: number
-  /** Read only the minimally required physical version. */
-  inspectVersion(headerValue: unknown): number
   /** Classify and translate one header without reading event rows. */
   readHeader(headerValue: unknown): SessionFormatHeaderReadResult
   /** Dispatch a complete physical JSON artifact through its frozen version codec. */
@@ -147,6 +145,6 @@ export interface SessionFormatCatalog {
   ): SessionFormatArtifact
   /** Restore current input directly or run all required adjacent migrations in memory. */
   migrate(artifact: SessionFormatArtifact): SessionFormatArtifact
-  /** Validate and encode an exact current logical artifact. */
+  /** Encode one current artifact that `migrate` returned or a live Session produced; it is not re-validated here. */
   encodeCurrent(artifact: SessionFormatArtifact): EncodedSessionFormatArtifact
 }

packages/session/session-format/tests/catalog.spec.ts

@@ -59,7 +59,6 @@ describe('Session format catalog', () => {
       delegationDepth: 0,
     } as const
 
-    expect(catalog.inspectVersion(oldHeader)).toBe(0)
     expect(catalog.readHeader(oldHeader)).toEqual({
       status: 'migration-required',
       storedVersion: 0,

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

@@ -28,7 +28,6 @@ import {
   decompressZstdFrame,
   decompressZstdPrefix,
   scanZstdFrames,
-  type ZstdFrameScan,
 } from './zstd.ts'
 
 /** Parsed JSONL values supplied to the format catalog. */
@@ -68,11 +67,7 @@ export interface EnsureJsonlGenerationOptions {
   readonly signal?: AbortSignal
 }
 
-/**
- * Result of current classification or exclusive publication. A present
- * `snapshot.zstdBody` owns a live decoder: the fused consumer must exhaust it
- * or call its disposer on every exit.
- */
+/** Result of current classification or exclusive publication. */
 export type EnsureJsonlGenerationResult =
   | {
     readonly status: 'current'
@@ -154,17 +149,10 @@ export interface JsonlPhysicalSnapshot {
   readonly identity: JsonlPhysicalIdentity
   readonly headerValue: Record<string, unknown>
   readonly headerRecord: Buffer
-  /** Single-use decoder owner; the fused consumer must dispose it on every exit. */
-  readonly zstdBody?: JsonlZstdBodyFrames
 }
 
-/** Live body-frame iterator plus the structural scan and mandatory decoder disposal. */
-export interface JsonlZstdBodyFrames extends Disposable {
-  readonly frames: Generator<Buffer, void, void>
-  readonly scan: ZstdFrameScan
-}
-
-interface StablePhysicalFile {
+/** 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
 }
@@ -172,18 +160,6 @@ interface StablePhysicalFile {
 interface JsonlPhysicalHeader {
   readonly value: Record<string, unknown>
   readonly record: Buffer
-  readonly zstdBody?: JsonlZstdBodyFrames
-}
-
-class OwnedZstdBodyFrames implements JsonlZstdBodyFrames {
-  constructor(
-    readonly frames: Generator<Buffer, void, void>,
-    readonly scan: ZstdFrameScan,
-  ) {}
-
-  [Symbol.dispose](): void {
-    this.frames.return()
-  }
 }
 
 interface DecodedPhysicalJsonl {
@@ -253,10 +229,20 @@ function fingerprint(value: JsonlPhysicalIdentity, bytes: Buffer): string {
 }
 
 /**
- * Read one stable revision with a single retry. If an append overlaps both
- * reads, return the second read's committed pre-read prefix instead of
- * starving behind a continuous writer.
+ * 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
+ * instead of starving behind a continuous writer.
+ * @param path - the generation file to read.
+ * @param signal - optional cancellation for the stat/read work.
+ * @returns the stable bytes (or the committed prefix) and their stat identity.
  */
+export async function readStableJsonlFile(
+  path: string,
+  signal?: AbortSignal,
+): Promise<StablePhysicalFile> {
+  return readStableSnapshot(path, signal, defaultFileSystem)
+}
+
 async function readStableSnapshot(
   path: string,
   signal: AbortSignal | undefined,
@@ -430,11 +416,10 @@ function readRawHeader(bytes: Buffer): JsonlPhysicalHeader {
 
 function readZstdHeader(bytes: Buffer, signal?: AbortSignal): JsonlPhysicalHeader {
   signal?.throwIfAborted()
-  const scan = scanZstdFrames(bytes)
-  const first = scan.frames[0]
+  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, scan.frames)
+  const decodedFrames = decoder.decode(bytes, [first])
   try {
     const decoded = decodedFrames.next()
     /* v8 ignore next -- one complete frame yields once or the decoder throws. */
@@ -444,15 +429,10 @@ function readZstdHeader(bytes: Buffer, signal?: AbortSignal): JsonlPhysicalHeade
     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,
-      zstdBody: new OwnedZstdBodyFrames(decodedFrames, scan),
-    }
-  } catch (error: unknown) {
+    return { value: value as Record<string, unknown>, record }
+  } finally {
     decodedFrames.return()
     decoder.close()
-    throw error
   }
 }
 
@@ -700,14 +680,12 @@ async function ensureCurrent(
     const quickHeader = readPhysicalHeader(source.bytes, compression, signal)
     const quickVersion = storedVersion(quickHeader.value)
     if (quickVersion !== sourceVersion) {
-      quickHeader.zstdBody?.[Symbol.dispose]()
       throw new Error(
         `resolved JSONL source filename identifies v${sourceVersion}, but its header identifies v${quickVersion}: `
         + sourcePath,
       )
     }
     if (sourceVersion > format.currentVersion) {
-      quickHeader.zstdBody?.[Symbol.dispose]()
       throw new JsonlGenerationNewerVersionError(sourceVersion, format.currentVersion, storedId(quickHeader.value))
     }
     if (sourceVersion === format.currentVersion) {
@@ -715,15 +693,9 @@ async function ensureCurrent(
         status: 'current',
         version: quickVersion,
         path: sourcePath,
-        snapshot: {
-          ...source,
-          headerValue: quickHeader.value,
-          headerRecord: quickHeader.record,
-          ...quickHeader.zstdBody === undefined ? {} : { zstdBody: quickHeader.zstdBody },
-        },
+        snapshot: { ...source, headerValue: quickHeader.value, headerRecord: quickHeader.record },
       }
     }
-    quickHeader.zstdBody?.[Symbol.dispose]()
     const validation = options.validateHistoricalHeader?.(quickHeader.value)
     if (validation !== undefined) await validation
 

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

@@ -13,7 +13,7 @@ import {
   sessionFormatCatalog,
 } from '@deepseek-ai/dsh-session-format-catalog'
 import { readdirSync, type Dirent } from 'node:fs'
-import { open, mkdir, readFile, readdir, realpath, link, rm, stat, truncate } from 'node:fs/promises'
+import { open, mkdir, readdir, realpath, link, rm, stat, truncate } from 'node:fs/promises'
 import { dirname, join, resolve } from 'node:path'
 import { performance } from 'node:perf_hooks'
 import { scheduler } from 'node:timers/promises'
@@ -45,8 +45,10 @@ import {
   ensureJsonlGenerationCurrent,
   JsonlGenerationNewerVersionError,
   JsonlGenerationUnsupportedMigrationError,
+  readStableJsonlFile,
   type EnsureJsonlGenerationResult,
   type JsonlGenerationFormatAdapter,
+  type JsonlPhysicalIdentity,
 } from './generation.ts'
 
 export type { JsonlCompression } from './format.ts'
@@ -106,14 +108,6 @@ interface StoredLog {
   readonly revision: PersistenceRevision
 }
 
-interface FileRevisionIdentity {
-  readonly dev: bigint
-  readonly ino: bigint
-  readonly size: bigint
-  readonly mtimeNs: bigint
-  readonly ctimeNs: bigint
-}
-
 /** One authoritative immutable generation selected from a Session directory. */
 interface ResolvedJsonlGeneration {
   readonly sourcePath: string
@@ -122,7 +116,7 @@ interface ResolvedJsonlGeneration {
 }
 
 /** Build the stat-derived best-effort change token shared by full and lightweight reads. */
-function fileRevision(identity: FileRevisionIdentity): PersistenceRevision {
+function fileRevision(identity: JsonlPhysicalIdentity): PersistenceRevision {
   return SessionPersistenceRevision([
     identity.dev,
     identity.ino,
@@ -381,7 +375,6 @@ class JsonlSessionPersistence extends SessionPersistence {
     const current = await this.ensureCurrentLog(id, signal, selected)
     /* v8 ignore next -- supplying a resolved generation makes absence unreachable. */
     if (current === undefined) throw new SessionPersistenceNotFoundError(id)
-    current.snapshot.zstdBody?.[Symbol.dispose]()
     return this.decodeStoredLog(
       current.path,
       id,
@@ -457,8 +450,8 @@ class JsonlSessionPersistence extends SessionPersistence {
       this.coldLogMemo.set(expectedId, memoized)
       return memoized
     }
-    const { buffer, revision } = await this.readStableFile(path, signal)
-    return this.decodeStoredLog(path, expectedId, buffer, revision, signal)
+    const { bytes, identity } = await readStableJsonlFile(path, signal)
+    return this.decodeStoredLog(path, expectedId, bytes, fileRevision(identity), signal)
   }
 
   /** Decode and memoize one already-stable current physical snapshot. */
@@ -533,7 +526,6 @@ class JsonlSessionPersistence extends SessionPersistence {
     if (selected === undefined) return undefined
     if (selected.sourceVersion === SESSION_FORMAT_VERSION) return selected.sourcePath
     const current = await this.ensureCurrentLog(id, signal)
-    current?.snapshot.zstdBody?.[Symbol.dispose]()
     return current?.path
   }
 
@@ -601,37 +593,6 @@ class JsonlSessionPersistence extends SessionPersistence {
     this.tracker.release(handle, materialized)
   }
 
-  /**
-   * Read a file's bytes with one bounded stability retry: a writer appending
-   * between stat and readFile yields a torn read, so a changed revision
-   * triggers exactly one re-read. A second change does not loop — the log is
-   * append-only, so the bytes at the retry's own pre-read stat size are a
-   * committed prefix, and the decoders treat anything past a torn cut as
-   * unwritten. A continuous writer therefore delays a read by at most one
-   * extra whole-file read instead of starving it.
-   * @param path - the artifact file to read.
-   * @param signal - optional cancellation for the stat/read work.
-   * @returns the stable bytes (or the committed prefix) and their revision.
-   */
-  private async readStableFile(
-    path: string,
-    signal?: AbortSignal,
-  ): Promise<{ buffer: Buffer; revision: PersistenceRevision }> {
-    signal?.throwIfAborted()
-    let identity = await stat(path, { bigint: true })
-    for (let attempt = 0; ; attempt += 1) {
-      const before = fileRevision(identity)
-      const buffer = await readFile(path, { signal })
-      signal?.throwIfAborted()
-      const after = await stat(path, { bigint: true })
-      if (before === fileRevision(after)) return { buffer, revision: before }
-      if (attempt === 1) {
-        return { buffer: buffer.subarray(0, Number(identity.size)), revision: before }
-      }
-      identity = after
-    }
-  }
-
   /** Decode complete frames and retain complete JSONL records from a torn final frame. */
   private async readZstdPrefix(
     buffer: Buffer,

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

@@ -191,19 +191,6 @@ describe('JSONL immutable generation publication', () => {
     expect(statFile).toHaveBeenCalledTimes(3)
   })
 
-  it('returns a disposable Zstandard body owner on the current fast path', async () => {
-    const root = await tempRoot()
-    const request = options(root, 'zstd', adapter(), 1)
-    await writeFile(request.sourcePath, await encodeZstd(1, [event0]))
-
-    const result = await ensureJsonlGenerationCurrent(request)
-    const body = result.snapshot.zstdBody
-
-    expect(body).toBeDefined()
-    body?.[Symbol.dispose]()
-    expect(body?.frames.next().done).toBe(true)
-  })
-
   it.each(['none', 'zstd'] as const)(
     'validates the selected %s historical header before invoking migration',
     async (compression) => {

packages/test-support/llm-replay/src/index.ts

@@ -298,21 +298,11 @@ function parsedSessionFixture(
 }
 
 /**
- * Convert one persisted or projected snapshot fixture to the current physical format in memory.
+ * Convert one persisted or projected snapshot fixture to the current physical format in memory for expected-output comparison.
  * Projected cwd tokens remain tokens so the ordinary snapshot normalizer can compare them with a fresh run.
  * @param text - one complete Session fixture.
  * @returns current-format JSONL with complete event envelopes; the input string and source file remain unchanged.
  */
-export function migrateSessionSnapshotFixture(text: string): string {
-  const parsed = parseSessionFixture(text)
-  return encodeCurrentSessionSnapshotFixture(text, parsed)
-}
-
-/**
- * Prepare one fixture for expected-output comparison through strict format validation.
- * @param text - one complete Session fixture.
- * @returns current-generation comparison JSONL without modifying the source file.
- */
 export function prepareSessionSnapshotFixtureForComparison(text: string): string {
   const parsed = parseSessionFixture(text)
   return encodeCurrentSessionSnapshotFixture(text, parsed)

packages/test-support/llm-replay/tests/llm-replay.spec.ts

@@ -28,7 +28,6 @@ import {
   installLlmReplay,
   loadReplayScript,
   loadSessionScripts,
-  migrateSessionSnapshotFixture,
   name,
   parseSessionHeader,
   parseSessionLog,
@@ -1194,7 +1193,7 @@ describe('loadReplayScript', () => {
     writeFileSync(file, source, 'utf8')
 
     expect(loadReplayScript({ file })).toEqual([{ kind: 'chunks', chunks: TEXT_CHUNKS }])
-    expect(JSON.parse(migrateSessionSnapshotFixture(source).split('\n')[0] as string)).toMatchObject({ version: 2 })
+    expect(JSON.parse(prepareSessionSnapshotFixtureForComparison(source).split('\n')[0] as string)).toMatchObject({ version: 2 })
     expect(readFileSync(file, 'utf8')).toBe(source)
   })