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 d63e6d4b88..acb9d72ef1 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 @@ -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 .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 +2026-07-19-zstandard-jsonl-session-logs.md: 93fc20f931c75552352834b9340e7d38680d4254 +2026-07-19-zstandard-jsonl-session-logs.zh.md: 061d7fcb55c775eed10e99bae47777d32cc8eee1 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 287ec94a91..93fc20f931 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,7 +28,7 @@ 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 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. +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 independently checksum-validated and passed through the [large-session restore pipeline](2026-08-05-large-session-jsonl-restore-pipeline.md), which owns decoder reuse, cooperative yielding, and incremental JSONL scanning. 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. @@ -53,5 +53,5 @@ The shared persistence and coordinator contracts run against both encodings. Bac - Ordinary session roots store `.jsonl.zstd` and retain append-only, fsync, rollback, and interrupted-turn recovery semantics. - Raw JSONL remains a deliberate configuration, but changing encoding requires a fresh/separate root or selecting the mode that matches existing artifacts. - One frame per durable batch adds bounded framing/checksum overhead and allows header-only listing plus repair from an exact append boundary. -- External tools must understand concatenated Zstandard frames or consume raw-mode artifacts; generic one-shot Node decompression reads only the first independent frame, so backend reads walk frames explicitly. +- External tools must understand concatenated Zstandard frames or consume raw-mode artifacts; generic one-shot Node decompression reads only the first independent frame, so backend reads walk frames through the [restore pipeline](2026-08-05-large-session-jsonl-restore-pipeline.md). - The implementation depends on Node's experimental built-in Zstandard API without an npm dependency; the supported-version compatibility gate makes drift visible. 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 4e57843264..061d7fcb55 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,7 +28,7 @@ JSONL 持久化后端会逐字保留每个 `SessionEvent`，其中包括数量 ### 读取、列举与崩溃恢复 -帧边界扫描器会读取标准魔数、可变头字段、块头与负载长度，以及可选校验和尾部，但不会解释压缩块。后端使用 Node 默认的 `ZSTD_e_end` 独立且按顺序解压完整帧；该模式要求帧完整并验证各帧校验和，再把明文交给既有 JSONL 扫描器。任何完整帧的校验和或解压失败、完整帧中畸形的 JSONL 尾部，或者无效帧结构都属于损坏并拒绝加载。 +帧边界扫描器会读取标准魔数、可变头字段、块头与负载长度，以及可选校验和尾部，但不会解释压缩块。完整帧会独立验证校验和，再进入[大型会话恢复流水线](2026-08-05-large-session-jsonl-restore-pipeline.md)；该流水线负责复用解码器、协作式让出事件循环和增量扫描 JSONL。任何完整帧的校验和或解压失败、完整帧中畸形的 JSONL 尾部，或者无效帧结构都属于损坏并拒绝加载。 列举只按有界分片读取到第一个完整帧可用为止，验证并解压该头部帧，绝不读取事件帧。因此，即使会话日志很大，专用头部帧仍能维持仅元数据列举。 @@ -53,5 +53,5 @@ CLI、ACP 与 stdio 应用包公开对称的 `persistenceCompression` 透传配 - 普通会话根目录存储 `.jsonl.zstd`，并保留仅追加、fsync、回滚与中断轮次恢复语义。 - 原始 JSONL 仍是显式配置，但切换编码需要使用全新或单独根目录，或者选择与既有产物匹配的模式。 - 每个持久批次一个帧会增加有界的帧与校验和开销，同时支持仅头部列举和从精确追加边界开始修复。 -- 外部工具必须理解串联的 Zstandard 帧，或者消费原始模式产物；Node 通用的一次性解压只读取第一个独立帧，因此后端读取会显式遍历各帧。 +- 外部工具必须理解串联的 Zstandard 帧，或者消费原始模式产物；Node 通用的一次性解压只读取第一个独立帧，因此后端读取会通过[恢复流水线](2026-08-05-large-session-jsonl-restore-pipeline.md)遍历各帧。 - 实现依赖 Node 的实验性内置 Zstandard API，但不增加 NPM 依赖；受支持版本兼容性门禁会暴露 API 漂移。 diff --git a/docs/config-catalog.md b/docs/config-catalog.md index 55b2c0d9d3..7b4391de5a 100644 --- a/docs/config-catalog.md +++ b/docs/config-catalog.md @@ -1143,7 +1143,7 @@ export interface Config { export type JsonlCompression = 'zstd' | 'none' ``` -Source: [`packages/session-persistence/session-persistence-jsonl/src/index.ts:40`](../packages/session-persistence/session-persistence-jsonl/src/index.ts) +Source: [`packages/session-persistence/session-persistence-jsonl/src/index.ts:45`](../packages/session-persistence/session-persistence-jsonl/src/index.ts) ## `@deepseek-ai/dsh-session-persistence-sqlite` diff --git a/packages/session-persistence/session-persistence-jsonl/src/format.ts b/packages/session-persistence/session-persistence-jsonl/src/format.ts index d37b5708a8..96e8221c65 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/format.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/format.ts @@ -217,106 +217,157 @@ export function eventLines(events: readonly SessionEvent[], packChunks: boolean) return records.map(record => JSON.stringify(record)).join('\n') } -/** - * Parse a JSONL log buffer into its preserved event prefix (the header is line - * 0). Event lines pass through verbatim; packed chunk rows expand back into - * their events, so callers see one contiguous event list regardless of layout. - * Fully written events in an interrupted final turn remain part of the - * prefix. The first unparsable record or seq gap after the last `turn/end` - * marks a tolerated torn tail; the same hole in the committed region rejects. - * - * @param buffer - the raw bytes of the log file (header line first). - * @returns the header, the preserved event prefix, and `committedBytes` — the - * byte offset the next append truncates any torn tail to. - */ -export function scanLog(buffer: Buffer): { meta: SessionHeader; events: SessionEvent[]; committedBytes: number } { - const text = buffer.toString('utf8') - // Track complete lines by byte offset: a non-newline tail is torn and ignored, - // and a running counter avoids rescanning a long multi-byte log. - const lines: { text: string; endByte: number }[] = [] - let start = 0 - let byteOffset = 0 - for (let i = 0; i < text.length; i++) { - if (text[i] === '\n') { - const lineText = text.slice(start, i) - byteOffset += Buffer.byteLength(lineText, 'utf8') + 1 // +1 for the '\n' (a 1-byte char) - lines.push({ text: lineText, endByte: byteOffset }) - start = i + 1 - } +interface SessionLogScan { + meta: SessionHeader + events: SessionEvent[] + committedBytes: number +} + +/** Parse one complete header record supplied independently from event rows. */ +function parseHeaderRecord(record: Buffer): SessionHeader { + if (record.length === 0 || record.at(-1) !== 0x0A || record.indexOf(0x0A) !== record.length - 1) { + throw new Error('empty or header-less session log') } - - const [headerEntry, ...eventEntries] = lines - if (headerEntry === undefined) throw new Error('empty or header-less session log') - - // Line 0 is the header. - let parsedHeader: unknown + let parsed: unknown try { - parsedHeader = JSON.parse(headerEntry.text) + parsed = JSON.parse(record.subarray(0, -1).toString('utf8')) } catch { throw new Error('corrupt session log: header line is not valid JSON') } - if (!isHeaderLine(parsedHeader)) { + if (!isHeaderLine(parsed)) { throw new Error('corrupt session log: first line is not a session header') } - const headerLine = parsedHeader + return fromHeaderLine(parsed) +} - // Parse and decode every complete line first so the last valid `turn/end` - // determines whether an earlier hole interrupts an otherwise closed - // execution or belongs to a tolerable final suffix. One line yields one - // event, or a whole run for a packed chunk row; a row-tagged line that fails - // row validation is a hole, exactly like unparsable JSON. - interface Parsed { ok: boolean; events?: SessionEvent[]; endByte: number } - const parsed: Parsed[] = eventEntries.map((entry) => { - try { - return { ok: true, events: decodeStorageRecord(JSON.parse(entry.text)), endByte: entry.endByte } - } catch { - return { ok: false, endByte: entry.endByte } - } - }) +/** + * Incrementally scan complete JSONL event records after an independently + * supplied header record. Newline search and byte offsets stay on raw buffers; + * only complete records are decoded to UTF-8. A fragment crossing writes is + * copied because a decoder may reuse its output buffer after `write()` returns. + */ +export class SessionLogScanner { + private readonly meta: SessionHeader + private readonly events: SessionEvent[] = [] + private fragments: Buffer[] = [] + private fragmentBytes = 0 + private inputBytes: number + private committedBytes: number + private eventLine = 0 + private issue: Error | undefined + private finished = false - // The last index (into eventEntries) that ends in a valid `turn/end`. A hole - // before this boundary cannot be a torn final suffix because later execution - // already closed. Standalone events after it remain part of the preserved - // contiguous prefix. A packed row never stores a turn/end, so only - // single-event lines can match. - let lastTurnEnd = -1 - for (let i = parsed.length - 1; i >= 0; i--) { - const p = parsed[i] - if (p?.ok && p.events?.some(e => e.type === 'turn/end')) { lastTurnEnd = i; break } + /** + * Create an event scanner from exactly one newline-terminated header record. + * @param headerRecord - the complete first JSONL record, including its newline. + */ + constructor(headerRecord: Buffer) { + this.meta = parseHeaderRecord(headerRecord) + this.inputBytes = headerRecord.length + this.committedBytes = headerRecord.length } - // Preserve the contiguous prefix, including a complete interrupted turn; - // holes through the last committed boundary throw, while later holes stop. - // Contiguity is a cursor over seqs (not the line index): a packed row - // advances the cursor by its whole run. - const preserved: SessionEvent[] = [] - let lastPreservedLine = -1 - scan: for (let i = 0; i < parsed.length; i++) { - const p = parsed[i] - if (!p?.ok || p.events === undefined) { - if (i <= lastTurnEnd) throw new Error(`corrupt session log: unparsable committed event at line ${i + 1}`) - break // torn tail fragment after the last turn/end — stop, tolerate - } - for (const event of p.events) { - if (event.seq !== preserved.length) { - if (i <= lastTurnEnd) { - throw new Error(`corrupt session log: seq gap in committed region at line ${i + 1} (expected ${preserved.length}, got ${event.seq})`) - } - break scan // gap after the last turn/end — torn tail, stop + /** + * Consume the next raw plaintext chunk, retaining only an incomplete final record. + * @param chunk - bytes immediately following all previously supplied bytes. + */ + write(chunk: Buffer): void { + if (this.finished) throw new Error('cannot write to a finished session log scanner') + const chunkStart = this.inputBytes + this.inputBytes += chunk.length + let lineStart = 0 + for ( + let newline = chunk.indexOf(0x0A); + newline !== -1; + newline = chunk.indexOf(0x0A, lineStart) + ) { + const fragment = chunk.subarray(lineStart, newline) + let line = fragment + if (this.fragments.length > 0) { + if (fragment.length > 0) this.fragments.push(fragment) + line = Buffer.concat(this.fragments, this.fragmentBytes + fragment.length) + this.fragments = [] + this.fragmentBytes = 0 } - preserved.push(event) + this.consumeEventLine(line, chunkStart + newline + 1) + lineStart = newline + 1 + } + if (lineStart < chunk.length) { + const fragment = Buffer.from(chunk.subarray(lineStart)) + this.fragments.push(fragment) + this.fragmentBytes += fragment.length } - lastPreservedLine = i } - // committedBytes = end of the last FULLY preserved line (header if none): the - // next append truncates any torn bytes past this point before writing the - // synthetic closers + new events. A line is preserved whole or not at all — - // a mid-row seq gap discards the whole row, keeping the truncation offset on - // a line boundary. - const lastPreserved = parsed[lastPreservedLine] - const committedBytes = lastPreserved !== undefined ? lastPreserved.endByte : headerEntry.endByte - return { meta: fromHeaderLine(headerLine), events: preserved, committedBytes } + /** + * Snapshot progress before appending a recoverable torn-frame prefix. + * @returns byte, committed-prefix, and expanded-event cursors. + */ + checkpoint(): { inputBytes: number; committedBytes: number; eventCount: number } { + return { + inputBytes: this.inputBytes, + committedBytes: this.committedBytes, + eventCount: this.events.length, + } + } + + /** + * Finish scanning, ignoring a final record without a newline as a torn tail. + * @returns the header, contiguous event prefix, and safe truncation offset. + */ + finish(): SessionLogScan { + this.finished = true + return { meta: this.meta, events: this.events, committedBytes: this.committedBytes } + } + + /** Decode one complete event row and update the contiguous prefix. */ + private consumeEventLine(line: Buffer, endByte: number): void { + this.eventLine += 1 + let decoded: SessionEvent[] + try { + decoded = decodeStorageRecord(JSON.parse(line.toString('utf8'))) + } catch { + this.issue ??= new Error(`corrupt session log: unparsable committed event at line ${this.eventLine}`) + return + } + + if (this.issue !== undefined) { + if (decoded.some(event => event.type === 'turn/end')) throw this.issue + return + } + + const rowStart = this.events.length + for (const event of decoded) { + if (event.seq !== this.events.length) { + const expected = this.events.length + this.events.length = rowStart + this.issue = new Error( + `corrupt session log: seq gap in committed region at line ${this.eventLine} ` + + `(expected ${expected}, got ${event.seq})`, + ) + if (decoded.some(candidate => candidate.type === 'turn/end')) throw this.issue + return + } + this.events.push(event) + } + this.committedBytes = endByte + } +} + +/** + * Parse a complete or torn JSONL buffer into its preserved event prefix. This + * compatibility wrapper supplies the first record separately, then delegates + * event rows to {@link SessionLogScanner}. + * + * @param buffer - the raw bytes of the log file (header line first). + * @returns the header, preserved event prefix, and byte offset safe to append at. + */ +export function scanLog(buffer: Buffer): SessionLogScan { + const headerEnd = buffer.indexOf(0x0A) + if (headerEnd === -1) throw new Error('empty or header-less session log') + const scanner = new SessionLogScanner(buffer.subarray(0, headerEnd + 1)) + scanner.write(buffer.subarray(headerEnd + 1)) + return scanner.finish() } /** diff --git a/packages/session-persistence/session-persistence-jsonl/src/index.ts b/packages/session-persistence/session-persistence-jsonl/src/index.ts index 8523a5ffdd..332a6338cb 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/index.ts @@ -11,6 +11,8 @@ import z from 'schemastery' import { readdirSync } from 'node:fs' import { open, mkdir, readFile, 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' import { randomBytes } from 'node:crypto' import { DEFAULT_PREPARED_SESSION_CACHE_SIZE, SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator, @@ -19,16 +21,27 @@ import { } from '@deepseek-ai/dsh-session-persistence' import type { SessionEvent, SessionId, SessionHeader, SessionPreparation } from '@deepseek-ai/dsh-session' import { - encodeSegment, eventLines, logPath, logSuffix, parseHeaderMeta, projectDir, scanLog, sessionDir, toHeaderLine, + encodeSegment, eventLines, logPath, logSuffix, parseHeaderMeta, projectDir, scanLog, sessionDir, + SessionLogScanner, toHeaderLine, type JsonlCompression, } from './format.ts' -import { compressZstdFrame, decompressZstdFrame, decompressZstdPrefix, scanZstdFrames } from './zstd.ts' +import { + compressZstdFrame, createZstdFrameDecoder, decompressZstdFrame, decompressZstdPrefix, scanZstdFrames, +} from './zstd.ts' import { ensureDurableDirectoryWin32, publishNewFileWin32 } from './win32.ts' export type { JsonlCompression } from './format.ts' const DEFAULT_PACK_CHUNKS = true const DEFAULT_COMPRESSION: JsonlCompression = 'zstd' +const ZSTD_DECODE_YIELD_INTERVAL_MS = 1000 + +/** Assert that the independently decodable first frame contains only the header record. */ +function assertZstdHeaderFrame(plaintext: Buffer): void { + if (plaintext.length === 0 || plaintext.indexOf(0x0A) !== plaintext.length - 1) { + throw new Error('corrupt Zstandard session log: first frame is not exactly one header line') + } +} /** Loader schema for the JSONL artifact's physical encoding. */ export const JsonlCompressionSchema: z = z.union([ @@ -260,61 +273,66 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi signal?.throwIfAborted() if (frames.length === 0) throw new Error('empty or header-less Zstandard session log') - const plaintextFrames: Buffer[] = [] - for (const frame of frames) { - let plaintext: Buffer - try { + const decoder = createZstdFrameDecoder() + let yieldDeadline = performance.now() + ZSTD_DECODE_YIELD_INTERVAL_MS + try { + const decodedFrames = decoder.decode(buffer, frames) + signal?.throwIfAborted() + const headerFrame = decodedFrames.next() + signal?.throwIfAborted() + if (headerFrame.done) throw new Error('empty or header-less Zstandard session log') + assertZstdHeaderFrame(headerFrame.value) + const scanner = new SessionLogScanner(headerFrame.value) + + let remainingFrames = frames.length - 1 + for (const plaintext of decodedFrames) { signal?.throwIfAborted() - plaintext = await decompressZstdFrame(buffer.subarray(frame.start, frame.end)) - } catch (error) { - /* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent */ - if (signal?.aborted) signal.throwIfAborted() - throw new Error(`corrupt Zstandard session log: frame at byte ${frame.start} failed validation`, { cause: error }) + scanner.write(plaintext) + remainingFrames -= 1 + if (remainingFrames > 0 && performance.now() >= yieldDeadline) { + await scheduler.yield() + signal?.throwIfAborted() + yieldDeadline = performance.now() + ZSTD_DECODE_YIELD_INTERVAL_MS + } } signal?.throwIfAborted() - plaintextFrames.push(plaintext) - } + const complete = scanner.checkpoint() + if (complete.committedBytes !== complete.inputBytes) { + throw new Error('corrupt Zstandard session log: complete frame contains a torn JSONL record') + } + if (tornStart === undefined) { + const prefix = scanner.finish() + return { meta: prefix.meta, events: prefix.events } + } - const headerFrame = plaintextFrames[0] - if (headerFrame === undefined || headerFrame.length === 0 || headerFrame.indexOf(0x0A) !== headerFrame.length - 1) { - throw new Error('corrupt Zstandard session log: first frame is not exactly one header line') - } - signal?.throwIfAborted() - const completePlaintext = Buffer.concat(plaintextFrames) - signal?.throwIfAborted() - const completePrefix = scanLog(completePlaintext) - signal?.throwIfAborted() - if (completePrefix.committedBytes !== completePlaintext.length) { - throw new Error('corrupt Zstandard session log: complete frame contains a torn JSONL record') - } - if (tornStart === undefined) { - return { meta: completePrefix.meta, events: completePrefix.events } - } - - let recoveredPlaintext: Buffer = Buffer.alloc(0) - try { + let recoveredPlaintext: Buffer = Buffer.alloc(0) + try { + signal?.throwIfAborted() + recoveredPlaintext = await decompressZstdPrefix(buffer.subarray(tornStart)) + } catch { + /* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent */ + if (signal?.aborted) signal.throwIfAborted() + // A structurally incomplete final frame may end before Node's decoder can + // emit any plaintext; the complete prior frames remain recoverable. + } signal?.throwIfAborted() - recoveredPlaintext = await decompressZstdPrefix(buffer.subarray(tornStart)) - } catch { + scanner.write(recoveredPlaintext) + const recoveredPrefix = scanner.finish() + signal?.throwIfAborted() + return { + meta: recoveredPrefix.meta, + events: recoveredPrefix.events, + tornMarker: { + truncateTo: tornStart, + recoveredEvents: recoveredPrefix.events.slice(complete.eventCount), + }, + } + } catch (error) { /* v8 ignore next -- decoder failure plus concurrent abort is timing-dependent */ if (signal?.aborted) signal.throwIfAborted() - // A structurally incomplete final frame may end before Node's decoder can - // emit any plaintext; the complete prior frames remain recoverable. - } - signal?.throwIfAborted() - const recoveredPrefix = scanLog(Buffer.concat([completePlaintext, recoveredPlaintext])) - signal?.throwIfAborted() - /* v8 ignore next 3 -- appending plaintext cannot shorten the already-scanned complete prefix */ - if (recoveredPrefix.events.length < completePrefix.events.length) { - throw new Error('corrupt Zstandard session log: recovered prefix does not extend complete frames') - } - return { - meta: recoveredPrefix.meta, - events: recoveredPrefix.events, - tornMarker: { - truncateTo: tornStart, - recoveredEvents: recoveredPrefix.events.slice(completePrefix.events.length), - }, + throw error + } finally { + decoder.close() } } @@ -662,9 +680,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi throw new Error('corrupt Zstandard session log: header frame failed validation', { cause: error }) } signal?.throwIfAborted() - if (plaintext.length === 0 || plaintext.indexOf(0x0A) !== plaintext.length - 1) { - throw new Error('corrupt Zstandard session log: first frame is not exactly one header line') - } + assertZstdHeaderFrame(plaintext) return plaintext.subarray(0, -1).toString('utf8') } } finally { diff --git a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts index 4c191aa0cd..db1624cb8f 100644 --- a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts +++ b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts @@ -8,7 +8,7 @@ import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' import type { Session, SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session' import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl' import { - encodeSegment, eventLines, logPath, projectDir, projectKey, scanLog, sessionDir, toHeaderLine, + encodeSegment, eventLines, logPath, projectDir, projectKey, scanLog, sessionDir, SessionLogScanner, toHeaderLine, } from '../src/format.ts' import { runPersistenceContract, meta, oneTurnLog, appendLog } from '../../session-persistence/tests/contract.ts' import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts' @@ -694,6 +694,25 @@ describe('SessionPersistenceJsonl: write path (session/event → flush)', () => describe('SessionPersistenceJsonl: scanLog unit', () => { + it('incrementally scans records split across reusable decoder chunks', () => { + const header = Buffer.from(`${JSON.stringify(toHeaderLine(meta('incremental')))}\n`) + const body = Buffer.from(`${oneTurnLog().map(event => JSON.stringify(event)).join('\n').replace('"hi"', '"你好"')}\n`) + const split = body.indexOf(Buffer.from('你')) + 1 + const firstChunk = Buffer.from(body.subarray(0, split)) + const scanner = new SessionLogScanner(header) + + scanner.write(firstChunk) + const checkpoint = scanner.checkpoint() + firstChunk.fill(0) + scanner.write(body.subarray(split)) + + expect(checkpoint).toMatchObject({ + inputBytes: header.length + split, + eventCount: 1, + }) + expect(scanner.finish()).toEqual(scanLog(Buffer.concat([header, body]))) + }) + it('rejects a header-less / empty log', () => { expect(() => scanLog(Buffer.from(''))).toThrow() }) 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 ffb262eec1..13936f449f 100644 --- a/packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts +++ b/packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts @@ -4,11 +4,17 @@ import { appendFile, mkdir, mkdtemp, open, readFile, readdir, rm, stat, writeFil import type { FileHandle } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' +import { performance } from 'node:perf_hooks' 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, decompressZstdPrefix, scanZstdFrames } from '../src/zstd.ts' +import { + compressZstdFrame, createZstdFrameDecoder, decompressZstdFrame, decompressZstdFrameSync, decompressZstdPrefix, + scanZstdFrames, +} from '../src/zstd.ts' +import { NodePrivateZstdFrameDecoder } from '../src/zstd-private-decoder.ts' +import { PublicZstdFrameDecoder } from '../src/zstd-public-decoder.ts' import { runPersistenceContract, meta, oneTurnLog } from '../../session-persistence/tests/contract.ts' import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts' @@ -17,7 +23,7 @@ const roots: string[] = [] const contexts: Context[] = [] interface ZstdReaderInternals { - readZstdPrefix(buffer: Buffer, signal?: AbortSignal): Promise + readZstdPrefix(buffer: Buffer, signal?: AbortSignal): Promise<{ events: SessionEvent[] }> } type HeaderRead = ( @@ -151,6 +157,32 @@ describe('Zstandard frame structure', () => { expect(first[4]! & 0x04).toBe(0x04) expect(second[4]! & 0x04).toBe(0x04) expect((await decompressZstdFrame(first)).toString()).toBe('header\n') + expect(decompressZstdFrameSync(second).toString()).toBe('event\n') + const decoder = createZstdFrameDecoder() + try { + const plaintext = Array.from(decoder.decode(stream, scanZstdFrames(stream).frames), chunk => Buffer.from(chunk)) + expect(Buffer.concat(plaintext).toString()).toBe('header\nevent\n') + } finally { + decoder.close() + } + }) + + it('keeps the public and Node-private synchronous decoders interchangeable', async () => { + const frames = [await compressZstdFrame('first\n'), await compressZstdFrame('second\n')] + const stream = Buffer.concat(frames) + const ranges = scanZstdFrames(stream).frames + const privateDecoder = NodePrivateZstdFrameDecoder.create() + expect(privateDecoder).toBeDefined() + + for (const decoder of [new PublicZstdFrameDecoder(), privateDecoder!]) { + try { + const plaintext = Array.from(decoder.decode(stream, ranges), chunk => Buffer.from(chunk)) + expect(plaintext).toHaveLength(2) + expect(Buffer.concat(plaintext).toString()).toBe('first\nsecond\n') + } finally { + decoder.close() + } + } }) it('distinguishes incomplete frame regions from invalid complete structure', () => { @@ -312,7 +344,7 @@ describe('SessionPersistenceJsonl: default Zstandard encoding', () => { await expect(ctx.sessionPersistence.load(header.id)).rejects.toThrow(/frame at byte .* failed validation/) }) - it('stops multi-frame inspection after cancellation interrupts the active decode', async () => { + it('stops multi-frame inspection when cancellation arrives at a one-second slice boundary', async () => { const root = await freshRoot() const ctx = await mount(root) const header = meta('cancel-zstd-frames') @@ -320,22 +352,32 @@ describe('SessionPersistenceJsonl: default Zstandard encoding', () => { const eventFrame = await compressZstdFrame(`${JSON.stringify(oneTurnLog()[0])}\n`) const laterFrame = await compressZstdFrame(`${JSON.stringify(oneTurnLog()[1])}\n`) const stream = Buffer.concat([headerFrame, eventFrame, laterFrame]) - expect(scanZstdFrames(stream).frames).toHaveLength(3) const controller = new AbortController() const reason = new Error('cancel after Zstandard decode starts') const reader = ctx.sessionPersistence as unknown as ZstdReaderInternals - const zstdModule = await import('../src/zstd.ts') - const decode = vi.spyOn(zstdModule, 'decompressZstdFrame') - - // readZstdPrefix reaches its first asynchronous decompression before it - // returns this promise. The microtask abort therefore occurs after decode - // starts and must prevent every later frame from reaching the decoder. + vi.spyOn(performance, 'now').mockReturnValueOnce(0).mockReturnValue(1001) const pending = reader.readZstdPrefix(stream, controller.signal) queueMicrotask(() => { controller.abort(reason) }) await expect(pending).rejects.toBe(reason) - expect(decode).toHaveBeenCalledTimes(1) - expect(decode).toHaveBeenCalledWith(headerFrame) + }) + + it('continues decoding every frame after a one-second slice yields', async () => { + const root = await freshRoot() + const ctx = await mount(root) + const header = meta('yield-zstd-frames') + const events = oneTurnLog().slice(0, 2) + const headerFrame = await compressZstdFrame(`${JSON.stringify(toHeaderLine(header))}\n`) + const eventFrames = await Promise.all(events.map(async event => ( + compressZstdFrame(`${JSON.stringify(event)}\n`) + ))) + const stream = Buffer.concat([headerFrame, ...eventFrames]) + const reader = ctx.sessionPersistence as unknown as ZstdReaderInternals + vi.spyOn(performance, 'now').mockReturnValueOnce(0).mockReturnValue(1001) + + const prefix = await reader.readZstdPrefix(stream) + + expect(prefix.events).toEqual(events) }) it.each(['none', 'zstd'] as const)(