fix(jsonl): recover truncated zstd frames on node 26

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()

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<Buffer>
 }
 
 /**
- * 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<Buffer> {
   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<Buffer> {
+  return zstdDecompressAsync(input, INCOMPLETE_FRAME_OPTIONS)
+}

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"')
   })
 })

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