From 13dee20db6f6d9ef4f451901ec45866f6ad6809d Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Thu, 3 Sep 2026 17:39:12 +0800 Subject: [PATCH] refactor(session): drop dead migration plumbing `ensureJsonlGenerationCurrent` returned a live Zstandard body iterator that every production caller disposed unread before decoding the same bytes again; the header read now decodes only the first frame and owns no decoder past its return. The JSONL backend reuses the generation module's stable-revision read instead of keeping a second copy, `encodeCurrent` encodes the validated current artifact it is documented to receive instead of re-running the migration chain on it, and the test-only `inspectVersion` catalog method and the `migrateSessionSnapshotFixture` duplicate of `prepareSessionSnapshotFixtureForComparison` are removed. --- .../session/session-format/README.i18n.yaml | 4 +- packages/session/session-format/README.md | 2 +- packages/session/session-format/README.zh.md | 2 +- .../session/session-format/src/catalog.ts | 12 +--- packages/session/session-format/src/types.ts | 4 +- .../session-format/tests/catalog.spec.ts | 1 - .../src/generation.ts | 70 ++++++------------- .../session-persistence-jsonl/src/index.ts | 51 ++------------ .../tests/generation.spec.ts | 13 ---- packages/test-support/llm-replay/src/index.ts | 12 +--- .../llm-replay/tests/llm-replay.spec.ts | 3 +- 11 files changed, 37 insertions(+), 137 deletions(-) diff --git a/packages/session/session-format/README.i18n.yaml b/packages/session/session-format/README.i18n.yaml index c2d0610142..7c5d6478ee 100644 --- a/packages/session/session-format/README.i18n.yaml +++ b/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: cfc1f7aafba463b454c669aa732cb4c9b049a5ef -README.zh.md: a02f3a6fa2863787207286bb25d92b44cadf6fe5 +README.md: 8fdb0282b3f9a9ac183dfc1703a4dab94004b95a +README.zh.md: 234df3794175241b2be1f0ab007cd7e8ecdc13d4 diff --git a/packages/session/session-format/README.md b/packages/session/session-format/README.md index cfc1f7aafb..8fdb0282b3 100644 --- a/packages/session/session-format/README.md +++ b/packages/session/session-format/README.md @@ -36,7 +36,7 @@ const catalog = createSessionFormatCatalog({ currentVersion, codecs, migrations, const descriptor = catalog.readHeader(physicalHeader) ``` -`createSessionFormatCatalog()` accepts one frozen codec per supported version, 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. +`createSessionFormatCatalog()` accepts one frozen codec per supported version, 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. 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. diff --git a/packages/session/session-format/README.zh.md b/packages/session/session-format/README.zh.md index a02f3a6fa2..234df37941 100644 --- a/packages/session/session-format/README.zh.md +++ b/packages/session/session-format/README.zh.md @@ -36,7 +36,7 @@ const catalog = createSessionFormatCatalog({ currentVersion, codecs, migrations, const descriptor = catalog.readHeader(physicalHeader) ``` -`createSessionFormatCatalog()` 接收每个受支持版本的一个冻结编解码器、每组相邻版本的一个迁移，以及当前产物与标头还原器。`inspectVersion()` 只读取物理版本以执行方向分派。`readHeader()` 在不读取事件的情况下返回 `current`、`migration-required`、`unsupported` 或 `malformed` 描述符。每个迁移边会先校验自己的目标标头，然后再运行最终的当前标头还原器。正文读取方调用 `decodeArtifact()` 或 `decodeRecoverableArtifact()`，然后调用 `migrate()`；写入方只使用经过校验的当前产物调用 `encodeCurrent()`。 +`createSessionFormatCatalog()` 接收每个受支持版本的一个冻结编解码器、每组相邻版本的一个迁移，以及当前产物与标头还原器。`readHeader()` 在不读取事件的情况下返回 `current`、`migration-required`、`unsupported` 或 `malformed` 描述符。每个迁移边会先校验自己的目标标头，然后再运行最终的当前标头还原器。正文读取方调用 `decodeArtifact()` 或 `decodeRecoverableArtifact()`，然后调用 `migrate()`；写入方只使用经过校验的当前产物调用 `encodeCurrent()`。 可恢复解码器返回已接受的逻辑前缀。编解码器可以丢弃一个格式错误或序号不连续的行及其未提交后缀，但后续成功解码的 `turn/end` 会使原始问题成为致命错误。 diff --git a/packages/session/session-format/src/catalog.ts b/packages/session/session-format/src/catalog.ts index 677b4f979a..e46530a4ce 100644 --- a/packages/session/session-format/src/catalog.ts +++ b/packages/session/session-format/src/catalog.ts @@ -38,14 +38,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) } @@ -93,7 +89,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}`, @@ -130,9 +126,8 @@ 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 codec = codecs.get(chain.currentVersion) as SessionFormatCodec - const encoded = codec.encodeArtifact(current, encodeOptions) + const encoded = codec.encodeArtifact(artifact, encodeOptions) 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)) @@ -144,7 +139,6 @@ export function createSessionFormatCatalog(options: SessionFormatCatalogOptions) return Object.freeze({ currentVersion: chain.currentVersion, - inspectVersion, readHeader, decodeArtifact, decodeRecoverableArtifact, diff --git a/packages/session/session-format/src/types.ts b/packages/session/session-format/src/types.ts index bf90c6e5c8..f3d6ebc41f 100644 --- a/packages/session/session-format/src/types.ts +++ b/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, options: SessionFormatEncodeOptions): EncodedSessionFormatArtifact } diff --git a/packages/session/session-format/tests/catalog.spec.ts b/packages/session/session-format/tests/catalog.spec.ts index 1077c37cc7..9b96d1910c 100644 --- a/packages/session/session-format/tests/catalog.spec.ts +++ b/packages/session/session-format/tests/catalog.spec.ts @@ -58,7 +58,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, diff --git a/packages/session/session-persistence-jsonl/src/generation.ts b/packages/session/session-persistence-jsonl/src/generation.ts index 255272e9d0..492c74cd15 100644 --- a/packages/session/session-persistence-jsonl/src/generation.ts +++ b/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 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 - 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 readonly record: Buffer - readonly zstdBody?: JsonlZstdBodyFrames -} - -class OwnedZstdBodyFrames implements JsonlZstdBodyFrames { - constructor( - readonly frames: Generator, - 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 { + 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, - record, - zstdBody: new OwnedZstdBodyFrames(decodedFrames, scan), - } - } catch (error: unknown) { + return { value: value as Record, 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 diff --git a/packages/session/session-persistence-jsonl/src/index.ts b/packages/session/session-persistence-jsonl/src/index.ts index c3fe79d8f0..80afe5f3da 100644 --- a/packages/session/session-persistence-jsonl/src/index.ts +++ b/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' @@ -115,14 +117,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 @@ -131,7 +125,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, @@ -393,7 +387,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, @@ -469,8 +462,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. */ @@ -545,7 +538,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 } @@ -613,37 +605,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, diff --git a/packages/session/session-persistence-jsonl/tests/generation.spec.ts b/packages/session/session-persistence-jsonl/tests/generation.spec.ts index 721dab44af..16e2f526a6 100644 --- a/packages/session/session-persistence-jsonl/tests/generation.spec.ts +++ b/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) => { diff --git a/packages/test-support/llm-replay/src/index.ts b/packages/test-support/llm-replay/src/index.ts index 8fb14405d8..4b159de614 100644 --- a/packages/test-support/llm-replay/src/index.ts +++ b/packages/test-support/llm-replay/src/index.ts @@ -292,21 +292,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) diff --git a/packages/test-support/llm-replay/tests/llm-replay.spec.ts b/packages/test-support/llm-replay/tests/llm-replay.spec.ts index 7be5882e75..6e89ae6eaf 100644 --- a/packages/test-support/llm-replay/tests/llm-replay.spec.ts +++ b/packages/test-support/llm-replay/tests/llm-replay.spec.ts @@ -18,7 +18,6 @@ import { installLlmReplay, loadReplayScript, loadSessionScripts, - migrateSessionSnapshotFixture, name, parseSessionHeader, parseSessionLog, @@ -878,7 +877,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: 1 }) + expect(JSON.parse(prepareSessionSnapshotFixtureForComparison(source).split('\n')[0] as string)).toMatchObject({ version: 1 }) expect(readFileSync(file, 'utf8')).toBe(source) })