fix(session): cover V3 admission and keep compressed fixture migratable

Keep the synthetic compressed historical fixture step-first without changing frozen generation data or shared fixtures. Pin its migrated empty system head and unchanged source bytes. Exercise malformed native system rows before and after recoverable damage through scanner and read/write handles. Remove the unreachable duplicate unsupported-format catch after current-row admission; preserve its owned rejection and generic error path.

Validation: original zstd regression red; admission-bypass negative control 3/3 red; five focused specs 314/314 pass with format.ts 100% statements/branches/functions/lines. Final admission cleanup 15/15 pass. Full build-backed lint found one matcher typing error; exact identifier oracle replaces it and focused type-aware lint passes.

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

@@ -502,9 +502,7 @@ export class SessionLogScanner {
     try {
       this.restore.decodeRow(decoded)
     } catch (error: unknown) {
-      if (error instanceof SessionFormatUnsupportedMigrationError) {
-        throw new SessionFormatUnsupportedError(error.message)
-      }
+      // Unsupported V3 rows have already been refused before recovery.
       /* v8 ignore next -- every production Session format decoder rejects with Error. */
       const detail = error instanceof Error ? error.message : String(error)
       const issue = new Error(`corrupt session log: invalid committed event at line ${this.eventLine}: ${detail}`, {

packages/session/session-persistence-jsonl/tests/v3-event-admission.spec.ts

@@ -71,6 +71,22 @@ describe('native V3 event admission at EOF', () => {
     },
   )
 
+  it.each(['', '{not json\n', 'null\n'])('refuses malformed system payloads after %j without modifying storage', async (corruption) => {
+    const malformed = { type: 'system/message', seq: 1, time: 2, data: null, surfaceOp: 'append' }
+    const bytes = Buffer.from(prefix + corruption + JSON.stringify(malformed) + '\n')
+    expect(() => scanLog(bytes)).toThrow('system/message data must be an object')
+    const path = await store(bytes)
+    const sourceStat = await stat(path)
+    for (const access of ['read', 'write'] as const) {
+      const opened = ctx.sessionPersistence.open(id, access).then(async (handle) => {
+        await handle.close()
+      })
+      await expect(opened).rejects.toThrow('system/message data must be an object')
+      expect(await readFile(path)).toEqual(bytes)
+      expect(await stat(path)).toMatchObject({ dev: sourceStat.dev, ino: sourceStat.ino })
+    }
+  })
+
   it.each(obsoleteTypes)('retains ignorable %s through scanning and a provider append', async (type) => {
     const event = obsoleteEvent(type, true)
     const bytes = Buffer.from(prefix + JSON.stringify(event) + '\n')

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

@@ -417,16 +417,36 @@ describe('JsonlSessionPersistence: default Zstandard encoding', () => {
     const header = meta('zstd-v0-read', '/work')
     const sourcePath = generationLogPath(root, header.cwd, header.id, 0, 'zstd')
     const currentPath = logPath(root, header.cwd, header.id, 'zstd')
+    const [turn, user, step, ...tail] = releasedV1OneTurnLog()
+    // Synthetic historical input opens its step before any surface so V3 can reserve the system head.
+    const historical = [turn!, step!, user!, ...tail].map((event, seq) => ({
+      ...event, seq: SessionSeq(seq), time: seq < 3 ? seq + 1 : event.time,
+    }))
     const source = Buffer.concat([
       await compressZstdFrame(`${JSON.stringify(releasedV0Header(header))}\n`),
-      await compressZstdFrame(`${releasedV1OneTurnLog().map(event => JSON.stringify(event)).join('\n')}\n`),
+      await compressZstdFrame(`${historical.map(event => JSON.stringify(event)).join('\n')}\n`),
     ])
     await mkdir(sessionDir(root, header.cwd, header.id), { recursive: true })
     await writeFile(sourcePath, source)
 
     await expect(readAll(ctx.sessionPersistence, header.id)).resolves.toEqual({
       meta: { ...header, delegationDepth: 0 },
-      events: oneTurnLog(),
+      events: [
+        historical[0],
+        historical[1],
+        {
+          type: 'system/message', seq: 2, time: 2, surfaceOp: 'append',
+          data: {
+            turn: 1, step: 1,
+            message: {
+              id: 'v2-to-v3-system-fc06c3f7720f3bc94ea7a2b7fadde6a5b100c6ab6ca342d2222bd017184a0b67',
+              role: 'system', source: { kind: 'plugin', plugin: '@deepseek-ai/dsh-system-prompt' }, content: [],
+            },
+          },
+        },
+        { ...historical[2], seq: 3 },
+        ...oneTurnLog().slice(3).map(event => ({ ...event, seq: event.seq + 1 })),
+      ],
     })
     expect(await readFile(sourcePath)).toEqual(source)
     await expect(readFile(currentPath)).rejects.toMatchObject({ code: 'ENOENT' })