diff --git a/.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.i18n.yaml b/.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.i18n.yaml index ba5bae421d..d63e6d4b88 100644 --- a/.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.i18n.yaml +++ b/.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.i18n.yaml @@ -1,6 +1,6 @@ # Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each # 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 -2026-07-19-zstandard-jsonl-session-logs.md: 74430624c771a265fb281e588e28733bc55d3eb6 -2026-07-19-zstandard-jsonl-session-logs.zh.md: b22275d1a7c54a743b11f4396318dd87e4f5b42a +# pnpm run verify-translation-pairing --write .agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.md +2026-07-19-zstandard-jsonl-session-logs.md: 287ec94a91101850e9343d36ffd27870daf1333b +2026-07-19-zstandard-jsonl-session-logs.zh.md: 4e578432640651de1eb1977229b7cdd462766c24 diff --git a/.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.md b/.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.md index 74430624c7..287ec94a91 100644 --- a/.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.md +++ b/.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.md @@ -28,11 +28,11 @@ First materialization compresses the two initial frames before opening the tempo ### Read, listing, and crash recovery -A frame-boundary scanner reads the standard magic, variable header fields, block headers and payload sizes, and optional checksum trailer. It does not interpret compressed blocks. Complete frames are decompressed independently and sequentially, which validates their checksums, and their plaintext is passed to the existing JSONL scanner. A checksum/decompression failure in any complete frame, a malformed complete-frame JSONL tail, or invalid frame structure is corruption and rejects. +A frame-boundary scanner reads the standard magic, variable header fields, block headers and payload sizes, and optional checksum trailer. It does not interpret compressed blocks. Complete frames are decompressed independently and sequentially with Node's default `ZSTD_e_end`, which requires frame completion and validates their checksums, and their plaintext is passed to the existing JSONL scanner. A checksum/decompression failure in any complete frame, a malformed complete-frame JSONL tail, or invalid frame structure is corruption and rejects. Listing reads in bounded chunks only until the first complete frame is available, validates and decompresses that header frame, and never reads an event frame. The dedicated header frame therefore preserves metadata-only listing even for very large session logs. -EOF inside the final frame is a recoverable torn tail. Node's decoder is given the available frame prefix; every complete newline-terminated event it emits is retained. Repair truncates from that frame's starting byte and appends one new checksummed frame containing the recovered complete events followed by the coordinator's synthetic tool, step, and turn closers. If the tear occurs before any complete event is decodable, repair drops the partial frame and retains all prior complete frames. +EOF inside the final frame is a recoverable torn tail. After the scanner establishes that boundary, a dedicated prefix decoder uses `finishFlush: ZSTD_e_flush` so Node emits available plaintext without requiring frame or checksum completion; every complete newline-terminated event it emits is retained. Repair truncates from that frame's starting byte and appends one new checksummed frame containing the recovered complete events followed by the coordinator's synthetic tool, step, and turn closers. If the tear occurs before any complete event is decodable, repair drops the partial frame and retains all prior complete frames. ### Consumers and verification diff --git a/.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.zh.md b/.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.zh.md index b22275d1a7..4e57843264 100644 --- a/.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.zh.md +++ b/.agents/notes/implemented/architecture/2026-07-19-zstandard-jsonl-session-logs.zh.md @@ -28,11 +28,11 @@ JSONL 持久化后端会逐字保留每个 `SessionEvent`，其中包括数量 ### 读取、列举与崩溃恢复 -帧边界扫描器会读取标准魔数、可变头字段、块头与负载长度，以及可选校验和尾部，但不会解释压缩块。后端独立且按顺序解压完整帧，由此验证各帧校验和，再把明文交给既有 JSONL 扫描器。任何完整帧的校验和或解压失败、完整帧中畸形的 JSONL 尾部，或者无效帧结构都属于损坏并拒绝加载。 +帧边界扫描器会读取标准魔数、可变头字段、块头与负载长度，以及可选校验和尾部，但不会解释压缩块。后端使用 Node 默认的 `ZSTD_e_end` 独立且按顺序解压完整帧；该模式要求帧完整并验证各帧校验和，再把明文交给既有 JSONL 扫描器。任何完整帧的校验和或解压失败、完整帧中畸形的 JSONL 尾部，或者无效帧结构都属于损坏并拒绝加载。 列举只按有界分片读取到第一个完整帧可用为止，验证并解压该头部帧，绝不读取事件帧。因此，即使会话日志很大，专用头部帧仍能维持仅元数据列举。 -最终帧内部遇到 EOF 属于可恢复的撕裂尾部。后端把已有帧前缀交给 Node 解码器，并保留其产出的每个完整、以换行结束的事件。修复从该帧起始字节截断，再追加一个新的带校验和帧，其中依次包含恢复出的完整事件，以及协调器生成的工具、步骤与轮次闭合事件。如果撕裂位置尚不足以解码任何完整事件，修复会丢弃该不完整帧并保留此前全部完整帧。 +最终帧内部遇到 EOF 属于可恢复的撕裂尾部。扫描器确定该边界后，专用前缀解码器会使用 `finishFlush: ZSTD_e_flush`，使 Node 不必等到帧结束或读到完整校验和就能产出已有明文；其中每个完整且以换行结束的事件都会保留。修复从该帧起始字节截断，再追加一个新的带校验和帧，其中依次包含恢复出的完整事件，以及协调器生成的工具、步骤与轮次闭合事件。如果撕裂位置尚不足以解码任何完整事件，修复会丢弃该不完整帧并保留此前全部完整帧。 ### 消费方与验证 diff --git a/packages/session-persistence/session-persistence-jsonl/src/index.ts b/packages/session-persistence/session-persistence-jsonl/src/index.ts index 4ef8e30ebf..4967010902 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/index.ts @@ -22,7 +22,7 @@ import { encodeSegment, eventLines, logPath, logSuffix, parseHeaderMeta, projectDir, scanLog, sessionDir, toHeaderLine, type JsonlCompression, } from './format.ts' -import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from './zstd.ts' +import { compressZstdFrame, decompressZstdFrame, decompressZstdPrefix, scanZstdFrames } from './zstd.ts' import { ensureDurableDirectoryWin32, publishNewFileWin32 } from './win32.ts' export type { JsonlCompression } from './format.ts' @@ -232,7 +232,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi let recoveredPlaintext: Buffer = Buffer.alloc(0) try { signal?.throwIfAborted() - recoveredPlaintext = await decompressZstdFrame(buffer.subarray(tornStart)) + recoveredPlaintext = await decompressZstdPrefix(buffer.subarray(tornStart)) } catch { /* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent */ if (signal?.aborted) signal.throwIfAborted() diff --git a/packages/session-persistence/session-persistence-jsonl/src/zstd.ts b/packages/session-persistence/session-persistence-jsonl/src/zstd.ts index bba2ef6344..e29747e399 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/zstd.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/zstd.ts @@ -14,6 +14,9 @@ const zstdDecompressAsync = promisify(zstdDecompress) const CHECKSUM_OPTIONS: ZstdOptions = { params: { [constants.ZSTD_c_checksumFlag]: 1 }, } +const INCOMPLETE_FRAME_OPTIONS: ZstdOptions = { + finishFlush: constants.ZSTD_e_flush, +} /** Byte range occupied by one structurally complete Zstandard frame. */ export interface ZstdFrameRange { @@ -106,11 +109,21 @@ export async function compressZstdFrame(input: Buffer | string): Promise } /** - * Decompress one complete frame or the available prefix of a torn final frame. - * Complete-frame checksums are validated by Node's decoder. - * @param input - bytes beginning at a Zstandard frame boundary. - * @returns plaintext produced from the available input. + * Decompress one complete frame and validate its checksum. + * @param input - one structurally complete Zstandard frame. + * @returns the frame plaintext. */ export async function decompressZstdFrame(input: Buffer): Promise { return zstdDecompressAsync(input) } + +/** + * Recover available plaintext from a structurally incomplete final frame. + * `ZSTD_e_flush` deliberately suppresses final-frame and checksum completion; + * callers must establish the torn frame boundary before using this helper. + * @param input - available bytes from a known incomplete Zstandard frame. + * @returns plaintext produced from the available input. + */ +export async function decompressZstdPrefix(input: Buffer): Promise { + return zstdDecompressAsync(input, INCOMPLETE_FRAME_OPTIONS) +} diff --git a/packages/session-persistence/session-persistence-jsonl/tests/zstd.compat.spec.ts b/packages/session-persistence/session-persistence-jsonl/tests/zstd.compat.spec.ts index bd552e738e..697b95d58a 100644 --- a/packages/session-persistence/session-persistence-jsonl/tests/zstd.compat.spec.ts +++ b/packages/session-persistence/session-persistence-jsonl/tests/zstd.compat.spec.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from 'vitest' -import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from '../src/zstd.ts' +import { compressZstdFrame, decompressZstdFrame, decompressZstdPrefix, scanZstdFrames } from '../src/zstd.ts' describe('JSONL Zstandard compatibility', () => { it('round-trips concatenated checksummed frames through the built-in Node API', async () => { @@ -19,6 +19,6 @@ describe('JSONL Zstandard compatibility', () => { const eventFrame = encoded.subarray(frames[1]!.start, frames[1]!.end) const missingChecksumByte = eventFrame.subarray(0, -1) expect(scanZstdFrames(missingChecksumByte)).toEqual({ frames: [], tornStart: 0 }) - expect((await decompressZstdFrame(missingChecksumByte)).toString()).toContain('"type":"turn/start"') + expect((await decompressZstdPrefix(missingChecksumByte)).toString()).toContain('"type":"turn/start"') }) }) diff --git a/packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts b/packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts index cef1ff71e5..0baf0ca734 100644 --- a/packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts +++ b/packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts @@ -8,7 +8,7 @@ import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' import type { SessionEvent } from '@deepseek-ai/dsh-session' import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl' import { logPath, scanLog, sessionDir, toHeaderLine, type JsonlCompression } from '../src/format.ts' -import { compressZstdFrame, decompressZstdFrame, scanZstdFrames } from '../src/zstd.ts' +import { compressZstdFrame, decompressZstdFrame, decompressZstdPrefix, scanZstdFrames } from '../src/zstd.ts' import { runPersistenceContract, meta, oneTurnLog } from '../../session-persistence/tests/contract.ts' import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts' @@ -69,7 +69,7 @@ async function tornFrame( const candidate = frame.subarray(0, end) if (scanZstdFrames(candidate).tornStart !== 0) continue try { - const decoded = (await decompressZstdFrame(candidate)).toString('utf8') + const decoded = (await decompressZstdPrefix(candidate)).toString('utf8') if (accepts(decoded)) return candidate } catch { // Some early cuts precede the first decodable block; keep searching for