test(jsonl): cover reusable decoder branches

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

@@ -280,6 +280,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
       signal?.throwIfAborted()
       const headerFrame = decodedFrames.next()
       signal?.throwIfAborted()
+      /* v8 ignore next -- a non-empty structural frame list makes the decoder yield its first frame or throw. */
       if (headerFrame.done) throw new Error('empty or header-less Zstandard session log')
       assertZstdHeaderFrame(headerFrame.value)
       const scanner = new SessionLogScanner(headerFrame.value)

packages/session-persistence/session-persistence-jsonl/src/zstd-private-decoder.ts

@@ -41,6 +41,7 @@ function privateZstdStream(
   const errorKey = Reflect.ownKeys(stream).find((key): key is symbol => (
     typeof key === 'symbol' && key.description === 'kError'
   ))
+  /* v8 ignore next -- one test runtime exposes one Node-private shape; the Node 22/24/26 matrix checks compatibility. */
   if (
     typeof handle !== 'object' || handle === null
     || typeof (handle as { writeSync?: unknown }).writeSync !== 'function'
@@ -48,6 +49,7 @@ function privateZstdStream(
     || candidate._writeState.length < 2
     || typeof candidate._defaultFlushFlag !== 'number'
     || errorKey === undefined
+    || candidate[errorKey] !== null
   ) return undefined
   return { stream: stream as NodeZstdPrivateStream, errorKey }
 }
@@ -81,10 +83,13 @@ export class NodePrivateZstdFrameDecoder implements ZstdFrameDecoder {
   static create(): NodePrivateZstdFrameDecoder | undefined {
     const stream = createZstdDecompress({ chunkSize: DECODE_CHUNK_SIZE })
     const privateAccess = privateZstdStream(stream)
+    /* v8 ignore next -- reached only when a supported Node release changes its private stream shape. */
     if (privateAccess !== undefined) {
       return new NodePrivateZstdFrameDecoder(privateAccess.stream, privateAccess.errorKey)
     }
+    /* v8 ignore next -- the active Node runtime passed the private-shape probe above. */
     stream.close()
+    /* v8 ignore next -- the active Node runtime passed the private-shape probe above. */
     return undefined
   }
 
@@ -111,6 +116,7 @@ export class NodePrivateZstdFrameDecoder implements ZstdFrameDecoder {
   /** Decode one frame; its returned scratch view remains valid until the next call. */
   private decodeFrame(input: Buffer): Buffer {
     const handle = this.stream._handle
+    /* v8 ignore next -- decode() rejects closed instances before entering this private frame operation. */
     if (this.closed || handle === null) throw new Error('cannot decode with a closed Zstandard frame decoder')
 
     let inputOffset = 0
@@ -125,11 +131,11 @@ export class NodePrivateZstdFrameDecoder implements ZstdFrameDecoder {
         inputRemaining,
         this.output,
         0,
-        DECODE_CHUNK_SIZE,
+        this.output.length,
       )
       if (this.decoderError !== undefined) throw this.decoderError
       const internalError = this.stream[this.errorKey]
-      if (internalError !== undefined && internalError !== null) {
+      if (internalError !== null) {
         if (internalError instanceof Error) throw internalError
         throw new Error('Zstandard decoder exposed a non-Error internal failure')
       }
@@ -137,21 +143,23 @@ export class NodePrivateZstdFrameDecoder implements ZstdFrameDecoder {
       const outputAfter = this.stream._writeState[0]
       const inputAfter = this.stream._writeState[1]
       const consumed = inputRemaining - inputAfter
-      const produced = DECODE_CHUNK_SIZE - outputAfter
+      const produced = this.output.length - outputAfter
       if (produced > 0) {
         outputBytes += produced
+        /* v8 ignore next -- Buffer cannot materialize a frame beyond its own process-wide maximum length. */
         if (outputBytes > bufferConstants.MAX_LENGTH) {
           throw new Error(`Zstandard frame output exceeds ${bufferConstants.MAX_LENGTH} bytes`)
         }
       }
 
       if (outputAfter !== 0) {
+        /* v8 ignore next -- structurally scanned ranges contain exactly one complete frame and no trailing bytes. */
         if (inputAfter !== 0) throw new Error('Zstandard frame decoder left trailing input')
         const finalChunk = this.output.subarray(0, produced)
         if (fullChunks.length === 0) return finalChunk
         if (produced > 0) fullChunks.push(Buffer.from(finalChunk))
-        const [onlyChunk] = fullChunks
+        const onlyChunk = fullChunks[0] as Buffer
-        return fullChunks.length === 1 && onlyChunk !== undefined
+        return fullChunks.length === 1
           ? onlyChunk
           : Buffer.concat(fullChunks, outputBytes)
       }

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

@@ -6,7 +6,7 @@
  */
 
 import {
-  constants, zstdCompress, zstdDecompress, zstdDecompressSync, type ZstdOptions,
+  constants, zstdCompress, zstdDecompress, type ZstdOptions,
 } from 'node:zlib'
 import { promisify } from 'node:util'
 import { NodePrivateZstdFrameDecoder } from './zstd-private-decoder.ts'
@@ -121,17 +121,6 @@ export async function decompressZstdFrame(input: Buffer): Promise<Buffer> {
   return zstdDecompressAsync(input)
 }
 
-/**
- * Synchronously decompress one complete frame and validate its checksum.
- * Complete-log readers time-slice repeated calls so the event loop regains
- * control without paying one asynchronous native dispatch per frame.
- * @param input - one structurally complete Zstandard frame.
- * @returns the frame plaintext.
- */
-export function decompressZstdFrameSync(input: Buffer): Buffer {
-  return zstdDecompressSync(input)
-}
-
 /** Common lifecycle for interchangeable synchronous multi-frame decoders. */
 export interface ZstdFrameDecoder {
   /**

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

@@ -694,6 +694,46 @@ describe('SessionPersistenceJsonl: write path (session/event → flush)', () =>
 
 
 describe('SessionPersistenceJsonl: scanLog unit', () => {
+  it('requires exactly one newline-terminated header record', () => {
+    const header = JSON.stringify(toHeaderLine(meta('scanner-header')))
+    expect(() => new SessionLogScanner(Buffer.alloc(0))).toThrow(/header-less/)
+    expect(() => new SessionLogScanner(Buffer.from(header))).toThrow(/header-less/)
+    expect(() => new SessionLogScanner(Buffer.from(`${header}\n${header}\n`))).toThrow(/header-less/)
+  })
+
+  it('handles empty writes, boundary newlines, torn fragments, and scanner completion', () => {
+    const header = Buffer.from(`${JSON.stringify(toHeaderLine(meta('scanner-lifecycle')))}\n`)
+    const event = Buffer.from(JSON.stringify(oneTurnLog()[0]))
+    const scanner = new SessionLogScanner(header)
+
+    scanner.write(Buffer.alloc(0))
+    scanner.write(event)
+    scanner.write(Buffer.from('\nignored torn tail'))
+    const result = scanner.finish()
+
+    expect(result.events).toEqual([oneTurnLog()[0]])
+    expect(result.committedBytes).toBe(header.length + event.length + 1)
+    expect(() =>{  scanner.write(Buffer.from('\n')) }).toThrow(/finished/)
+  })
+
+  it('keeps scanning after a tolerable corrupt suffix until a committed turn end appears', () => {
+    const header = Buffer.from(`${JSON.stringify(toHeaderLine(meta('scanner-corrupt-suffix')))}\n`)
+    const scanner = new SessionLogScanner(header)
+    scanner.write(Buffer.from([
+      JSON.stringify(oneTurnLog()[0]),
+      '{not json',
+      JSON.stringify({ type: 'step/start', seq: 1, time: 2, data: { turn: 1, step: 1 } }),
+      '',
+    ].join('\n')))
+    expect(scanner.finish().events).toEqual([oneTurnLog()[0]])
+
+    const committed = new SessionLogScanner(header)
+    expect(() =>{  committed.write(Buffer.from([
+      JSON.stringify({ type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } }),
+      '',
+    ].join('\n'))) }).toThrow(/seq gap in committed region/)
+  })
+
   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`)

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

@@ -10,8 +10,8 @@ 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, createZstdFrameDecoder, decompressZstdFrame, decompressZstdFrameSync, decompressZstdPrefix,
+  compressZstdFrame, createZstdFrameDecoder, decompressZstdFrame, decompressZstdPrefix, scanZstdFrames,
-  scanZstdFrames,
+  type ZstdFrameDecoder,
 } from '../src/zstd.ts'
 import { NodePrivateZstdFrameDecoder } from '../src/zstd-private-decoder.ts'
 import { PublicZstdFrameDecoder } from '../src/zstd-public-decoder.ts'
@@ -157,7 +157,6 @@ 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))
@@ -185,6 +184,96 @@ describe('Zstandard frame structure', () => {
     }
   })
 
+  it('falls back to the public decoder when the private Node contract is unavailable', () => {
+    vi.spyOn(NodePrivateZstdFrameDecoder, 'create').mockReturnValue(undefined)
+    const decoder = createZstdFrameDecoder()
+    expect(decoder).toBeInstanceOf(PublicZstdFrameDecoder)
+    decoder.close()
+  })
+
+  it('enforces decoder lifecycle and checksum errors through both implementations', async () => {
+    const frame = await compressZstdFrame('frame\n')
+    const range = [{ start: 0, end: frame.length }]
+    const corrupt = Buffer.from(frame)
+    corrupt[corrupt.length - 1] = corrupt[corrupt.length - 1]! ^ 0xFF
+    const factories: Array<() => ZstdFrameDecoder> = [
+      () => new PublicZstdFrameDecoder(),
+      () => NodePrivateZstdFrameDecoder.create()!,
+    ]
+
+    for (const create of factories) {
+      const interrupted = create()
+      const iterator = interrupted.decode(frame, range)
+      expect(iterator.next().value?.toString()).toBe('frame\n')
+      iterator.return()
+      expect(() => Array.from(interrupted.decode(frame, range))).toThrow(/already started/)
+      interrupted.close()
+
+      const closed = create()
+      closed.close()
+      closed.close()
+      expect(() => Array.from(closed.decode(frame, range))).toThrow(/closed/)
+
+      const invalid = create()
+      expect(() => Array.from(invalid.decode(corrupt, range))).toThrow(/frame at byte 0 failed validation/)
+    }
+  })
+
+  it('assembles private-decoder output at and beyond its reusable chunk boundary', async () => {
+    for (const length of [8, 9]) {
+      const plaintext = Buffer.alloc(length, 0x61)
+      const frame = await compressZstdFrame(plaintext)
+      const decoder = NodePrivateZstdFrameDecoder.create()!
+      ;(decoder as unknown as { output: Buffer }).output = Buffer.allocUnsafe(8)
+      const [decoded] = Array.from(
+        decoder.decode(frame, [{ start: 0, end: frame.length }]),
+        chunk => Buffer.from(chunk),
+      )
+      expect(decoded).toEqual(plaintext)
+    }
+  })
+
+  it('normalizes private decoder stream failures', async () => {
+    interface PrivateDecoderInternals {
+      stream: {
+        [key: symbol]: unknown
+        emit(event: string, error: Error): boolean
+      }
+      errorKey: symbol
+    }
+    const frame = await compressZstdFrame('frame\n')
+    const range = [{ start: 0, end: frame.length }]
+
+    const emitted = NodePrivateZstdFrameDecoder.create()!
+    const emittedInternals = emitted as unknown as PrivateDecoderInternals
+    const first = new Error('first emitted decoder failure')
+    emittedInternals.stream.emit('error', first)
+    emittedInternals.stream.emit('error', new Error('later emitted decoder failure'))
+    try {
+      Array.from(emitted.decode(frame, range))
+      throw new Error('expected emitted decoder failure')
+    } catch (error) {
+      expect((error as Error).cause).toBe(first)
+    }
+
+    for (const internalFailure of [new Error('internal decoder failure'), 'not an Error']) {
+      const decoder = NodePrivateZstdFrameDecoder.create()!
+      const internals = decoder as unknown as PrivateDecoderInternals
+      internals.stream[internals.errorKey] = internalFailure
+      try {
+        Array.from(decoder.decode(frame, range))
+        throw new Error('expected internal decoder failure')
+      } catch (error) {
+        const cause = (error as Error).cause
+        if (internalFailure instanceof Error) {
+          expect(cause).toBe(internalFailure)
+        } else {
+          expect(cause).toMatchObject({ message: 'Zstandard decoder exposed a non-Error internal failure' })
+        }
+      }
+    }
+  })
+
   it('distinguishes incomplete frame regions from invalid complete structure', () => {
     expect(scanZstdFrames(MAGIC.subarray(0, 2))).toEqual({ frames: [], tornStart: 0 })
     expect(scanZstdFrames(MAGIC)).toEqual({ frames: [], tornStart: 0 })