fix(session): refuse foreign format versions before parsing current structure

Review round: the JSONL backend now refuses a foreign header version straight
from the raw header line, before validating today's header shape or decoding
any event row, so a structurally different future format reports the upgrade
direction instead of corruption (shared message builder
sessionFormatVersionRefusal). HMR live-prefix adoption runs the unknown-type
guard like the other read paths. The appendCore comment now states why the
unknown-type guard is read-side only, the loadStoredFrom JSDoc and README pin
the seek-vs-sequential refusal-scope divergence, and the generated catalog
preamble lists the ignorable envelope field.

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

@@ -9,8 +9,9 @@
  */
 
 import { join } from 'node:path'
-import { decodeStorageRecord, packChunkRuns } from '@deepseek-ai/dsh-session'
+import { decodeStorageRecord, packChunkRuns, SESSION_FORMAT_VERSION } from '@deepseek-ai/dsh-session'
 import type { SessionEvent, SessionHeader, SessionId, StorageRecord } from '@deepseek-ai/dsh-session'
+import { SessionFormatUnsupportedError, sessionFormatVersionRefusal } from '@deepseek-ai/dsh-session-persistence'
 
 /** Physical encoding selected for JSONL session artifacts. */
 export type JsonlCompression = 'zstd' | 'none'
@@ -229,6 +230,22 @@ interface SessionLogScan {
 }
 
 /** Parse one complete header record supplied independently from event rows. */
+/**
+ * Refuse a header carrying a format version this build does not read BEFORE
+ * validating the current header shape or decoding any event row: a future
+ * format need not satisfy today's structural checks at all, and its user must
+ * see "upgrade the harness", never "corrupt session log".
+ * @param parsed - the JSON-parsed first line of a session artifact.
+ */
+function refuseForeignFormatVersion(parsed: unknown): void {
+  if (typeof parsed !== 'object' || parsed === null) return
+  const { version, id } = parsed as { version?: unknown; id?: unknown }
+  if (typeof version !== 'number' || version === SESSION_FORMAT_VERSION) return
+  throw new SessionFormatUnsupportedError(
+    sessionFormatVersionRefusal(typeof id === 'string' ? id : String(id), version),
+  )
+}
+
 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')
@@ -239,6 +256,7 @@ function parseHeaderRecord(record: Buffer): SessionHeader {
   } catch {
     throw new Error('corrupt session log: header line is not valid JSON')
   }
+  refuseForeignFormatVersion(parsed)
   if (!isHeaderLine(parsed)) {
     throw new Error('corrupt session log: first line is not a session header')
   }

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

@@ -16,7 +16,7 @@ import { scheduler } from 'node:timers/promises'
 import { randomBytes } from 'node:crypto'
 import {
   DEFAULT_PREPARED_SESSION_CACHE_SIZE, DEFAULT_WRITE_BATCH_MAX_DELAY_MS, MAX_WRITE_BATCH_DELAY_MS,
-  SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
+  SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator, SessionFormatUnsupportedError,
   type PersistenceBackend, type SessionLocation, type SessionPersistenceSnapshot,
   type SessionInspection, type SessionPersistenceRevision as PersistenceRevision, type StoredPrefix,
 } from '@deepseek-ai/dsh-session-persistence'
@@ -256,19 +256,29 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
       }
     }
     let prefix: Omit<StoredPrefix<JsonlTornMarker>, 'revision'>
-    if (this.compression === 'zstd') {
-      prefix = await this.readZstdPrefix(buffer, signal)
-    } else {
-      signal?.throwIfAborted()
-      const { meta, events, committedBytes } = scanLog(buffer)
-      signal?.throwIfAborted()
-      prefix = {
-        meta,
-        events,
-        ...committedBytes < buffer.byteLength
-          ? { tornMarker: { truncateTo: committedBytes, recoveredEvents: [] } }
-          : {},
+    try {
+      if (this.compression === 'zstd') {
+        prefix = await this.readZstdPrefix(buffer, signal)
+      } else {
+        signal?.throwIfAborted()
+        const { meta, events, committedBytes } = scanLog(buffer)
+        signal?.throwIfAborted()
+        prefix = {
+          meta,
+          events,
+          ...committedBytes < buffer.byteLength
+            ? { tornMarker: { truncateTo: committedBytes, recoveredEvents: [] } }
+            : {},
+        }
       }
+    } catch (error: unknown) {
+      // A parse-time format refusal predates any SessionHeader, so the
+      // coordinator's locate-based enrichment cannot run; attach the artifact
+      // this read actually refused.
+      if (error instanceof SessionFormatUnsupportedError && error.location === undefined) {
+        throw new SessionFormatUnsupportedError(`${error.message} (raw log: ${path})`, { kind: 'jsonl', path })
+      }
+      throw error
     }
     signal?.throwIfAborted()
     await this.assertStoredIdentity(path, prefix.meta, expectedId, signal)

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

@@ -3,7 +3,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
 import { Context } from '@deepseek-ai/cordis'
 import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat, symlink } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
-import { isAbsolute, join, relative, resolve } from 'node:path'
+import { dirname, isAbsolute, join, relative, resolve } from 'node:path'
 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'
@@ -187,6 +187,25 @@ describe('SessionPersistenceJsonl: format helpers', () => {
     await fiber.dispose()
   })
 
+  it('refuses a structurally foreign future header as unsupported, not corrupt', async () => {
+    const absoluteRoot = await freshRoot()
+    const ctx = new Context()
+    await ctx.plugin(SessionStore)
+    const fiber = await ctx.plugin(SessionPersistenceJsonl, { root: absoluteRoot, compression: 'none' })
+    // A future format need not satisfy today's header shape at all (no
+    // createdAt, unknown fields): the version must be refused before shape
+    // validation, so the user sees the upgrade direction.
+    const id = SessionId('future-shape')
+    const path = rawLogPath(resolve(absoluteRoot), '/work', id)
+    await mkdir(dirname(path), { recursive: true })
+    await writeFile(path, `${JSON.stringify({ type: 'session', version: 42, id, futureOnly: true })}\n{"future":"row"}\n`)
+    const failure = await ctx.sessionPersistence.load(id).then(() => undefined, (error: unknown) => error as Error)
+    expect(failure?.name).toBe('SessionFormatUnsupportedError')
+    expect(failure?.message).toMatch(/written by a newer harness.*upgrade the harness/)
+    expect(failure?.message).toContain(`(raw log: ${path})`)
+    await fiber.dispose()
+  })
+
   it('points a format refusal at the raw log path', async () => {
     const absoluteRoot = await freshRoot()
     const ctx = new Context()