deepseek-harness/packages/session/session-persistence/tests/format-decoder.spec.ts

import { afterEach, describe, expect, it, vi } from 'vitest'
import { SessionId } from '@deepseek-ai/dsh-session'
import type { SessionEvent } from '@deepseek-ai/dsh-session'
import {
  SessionPersistenceRevision,
  SessionPersistenceRevisionConflictError,
} from '../src/revision.ts'
import type {
  SessionFormatMigration,
  StoredEventReadCompletion,
  StoredSessionSource,
} from '../src/format-decoder.ts'
import { sessionFormatVersionRefusal } from '../src/format-decoder.ts'
import { unversionedFormatCompatibility } from '../src/format-v0-compat.ts'

const id = SessionId('format-migration')
type SessionFormatMigrationInstance = InstanceType<SessionFormatMigration>

function eventLog(): SessionEvent[] {
  return [
    { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } },
    { type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } },
  ]
}

async function collectEvents(events: AsyncIterable<SessionEvent>): Promise<SessionEvent[]> {
  const collected: SessionEvent[] = []
  for await (const event of events) collected.push(event)
  return collected
}

async function decodedFailure(
  decoded: ReturnType<typeof import('../src/format-decoder.ts')['decodeStoredSession']>,
): Promise<Error> {
  const completion = decoded.completed.catch((error: unknown) => error)
  const consumption = collectEvents(decoded.events).catch((error: unknown) => error)
  const [streamFailure, completionFailure] = await Promise.all([consumption, completion])
  expect(streamFailure).toBe(completionFailure)
  expect(streamFailure).toBeInstanceOf(Error)
  return streamFailure as Error
}

function storedSource(
  version: number,
  events: readonly unknown[],
): { source: StoredSessionSource<never>; reads: number[]; meta: Record<string, unknown> } {
  const reads: number[] = []
  const meta: Record<string, unknown> = { version, id, createdAt: 1 }
  return {
    meta,
    reads,
    source: {
      meta,
      revision: SessionPersistenceRevision(`format-v${version}`),
      readEvents({ fromSeq = 0 } = {}) {
        reads.push(fromSeq)
        return {
          events: (async function* (): AsyncIterable<unknown> {
            for (const event of events) {
              const seq = typeof event === 'object' && event !== null
                ? (event as { seq?: unknown }).seq
                : undefined
              if (!Number.isSafeInteger(seq) || (seq as number) < 0 || (seq as number) >= fromSeq) {
                yield structuredClone(event)
              }
            }
          })(),
          completed: Promise.resolve({}),
        }
      },
    },
  }
}

function defineMigration(
  from: number,
  create: () => SessionFormatMigrationInstance,
  to = from + 1,
): SessionFormatMigration {
  return class implements SessionFormatMigrationInstance {
    static readonly from = from
    static readonly to = to

    private readonly delegate = create()

    header(meta: unknown): unknown {
      return this.delegate.header(meta)
    }

    event(value: unknown): unknown {
      return this.delegate.event(value)
    }

    finish(): void {
      this.delegate.finish?.()
    }
  }
}

function migration(
  from: number,
  calls: string[],
  to = from + 1,
): SessionFormatMigration {
  return defineMigration(from, () => {
    let observedInput = false
    return {
      header(meta) {
        calls.push(`header:${from}`)
        return { ...(meta as Record<string, unknown>), version: to }
      },
      event(value) {
        if (!observedInput) {
          calls.push(`events:${from}`)
          observedInput = true
        }
        const event = value as SessionEvent
        const data = event.data as Record<string, unknown>
        const migrationPath = Array.isArray(data['migrationPath'])
          ? data['migrationPath'] as unknown[]
          : []
        return {
          ...event,
          data: {
            ...data,
            [`migratedFrom${from}`]: true,
            migrationPath: [...migrationPath, from],
          },
        }
      },
    }
  }, to)
}

async function configuredDecoder(
  currentVersion: number,
  migrations: readonly SessionFormatMigration[],
  calls: string[] = [],
): Promise<{
  decodeStoredSession: typeof import('../src/format-decoder.ts')['decodeStoredSession']
  decodeStoredSessionHeader: typeof import('../src/format-decoder.ts')['decodeStoredSessionHeader']
  validateHeader: ReturnType<typeof vi.fn>
}> {
  vi.resetModules()
  const validateHeader = vi.fn((sessionId: SessionId, _seed: unknown, meta: unknown) => {
    calls.push('validate-header')
    const record = meta as Record<string, unknown>
    if (record['version'] !== currentVersion) {
      throw new Error(`current header validator received v${String(record['version'])}`)
    }
    if (record['id'] !== sessionId) throw new Error('current header validator received the wrong id')
    if (!Number.isSafeInteger(record['createdAt'])) {
      throw new Error('current header validator received invalid createdAt')
    }
    return { header: Object.freeze(structuredClone(record)) }
  })
  vi.doMock('@deepseek-ai/dsh-session', async () => {
    const actual = await vi.importActual<typeof import('@deepseek-ai/dsh-session')>(
      '@deepseek-ai/dsh-session',
    )
    return {
      ...actual,
      SESSION_FORMAT_VERSION: currentVersion,
      Session: { create: validateHeader },
    }
  })
  vi.doMock('../src/format-migrations/index.ts', () => ({
    SESSION_FORMAT_MIGRATIONS: migrations,
  }))
  const decoder = await import('../src/format-decoder.ts')
  return {
    decodeStoredSession: decoder.decodeStoredSession,
    decodeStoredSessionHeader: decoder.decodeStoredSessionHeader,
    validateHeader,
  }
}

afterEach(() => {
  vi.doUnmock('@deepseek-ai/dsh-session')
  vi.doUnmock('../src/format-migrations/index.ts')
  vi.resetModules()
})

describe('versioned Session format decoder', { concurrent: false }, () => {
  it('describes both unsupported format directions', () => {
    expect(sessionFormatVersionRefusal(id, 1)).toContain('newer harness')
    expect(sessionFormatVersionRefusal(id, -1)).toContain('older than the supported')
  })

  it('runs a single migration lazily and reads the complete old log before slicing', async () => {
    const calls: string[] = []
    const step = migration(0, calls)
    const { decodeStoredSession, validateHeader } = await configuredDecoder(1, [step], calls)
    const originalEvents = eventLog()
    const originalSnapshot = structuredClone(originalEvents)
    const stored = storedSource(0, originalEvents)

    const decoded = decodeStoredSession(stored.source, id, 1)
    expect(decoded.sourceVersion).toBe(0)
    expect(decoded.meta.version).toBe(1)
    expect(calls).toEqual(['header:0', 'validate-header'])
    expect(stored.reads).toEqual([])

    const migrated = await collectEvents(decoded.events)
    await decoded.completed

    expect(stored.reads).toEqual([0])
    expect(calls).toEqual(['header:0', 'validate-header', 'events:0'])
    expect(migrated).toEqual([
      {
        ...originalEvents[1],
        data: { ...originalEvents[1]?.data, migratedFrom0: true, migrationPath: [0] },
      },
    ])
    expect(originalEvents).toEqual(originalSnapshot)
    expect(stored.meta).toEqual({ version: 0, id, createdAt: 1 })
    expect(validateHeader).toHaveBeenCalledOnce()
  })

  it('lets an old-format suffix migration use facts from events before fromSeq', async () => {
    const step = defineMigration(0, () => {
      let previousSeq: number | undefined
      return {
        header: meta => ({ ...(meta as Record<string, unknown>), version: 1 }),
        event(value) {
          const event = value as SessionEvent
          const migrated = previousSeq === undefined
            ? event
            : { ...event, data: { ...event.data, previousSeq } }
          previousSeq = event.seq
          return migrated
        },
      }
    })
    const { decodeStoredSession } = await configuredDecoder(1, [step])
    const stored = storedSource(0, eventLog())

    const decoded = decodeStoredSession(stored.source, id, 1)
    const events = await collectEvents(decoded.events)
    await decoded.completed

    expect(stored.reads).toEqual([0])
    expect(events).toEqual([{
      ...eventLog()[1],
      data: { ...eventLog()[1]?.data, previousSeq: 0 },
    }])
  })

  it('streams migrated events with backpressure instead of buffering the complete log', async () => {
    const releaseTail = Promise.withResolvers<undefined>()
    const physicalCompletion = Promise.withResolvers<StoredEventReadCompletion<never>>()
    const reads: number[] = []
    const source: StoredSessionSource<never> = {
      meta: { version: 0, id, createdAt: 1 },
      revision: SessionPersistenceRevision('streaming-source'),
      readEvents({ fromSeq = 0 } = {}) {
        reads.push(fromSeq)
        return {
          events: (async function* (): AsyncIterable<unknown> {
            try {
              yield structuredClone(eventLog()[0])
              await releaseTail.promise
              yield structuredClone(eventLog()[1])
              physicalCompletion.resolve({})
            } catch (error: unknown) {
              physicalCompletion.reject(error)
              throw error
            }
          })(),
          completed: physicalCompletion.promise,
        }
      },
    }
    const { decodeStoredSession } = await configuredDecoder(1, [migration(0, [])])
    const decoded = decodeStoredSession(source, id)
    const iterator = decoded.events[Symbol.asyncIterator]()

    const first = await iterator.next()
    expect(first).toMatchObject({ done: false, value: { seq: 0 } })
    expect(reads).toEqual([0])
    let completed = false
    void decoded.completed.then(() => { completed = true })
    await Promise.resolve()
    expect(completed).toBe(false)

    releaseTail.resolve(undefined)
    await expect(iterator.next()).resolves.toMatchObject({ done: false, value: { seq: 1 } })
    await expect(iterator.next()).resolves.toEqual({ done: true, value: undefined })
    await expect(decoded.completed).resolves.toEqual({})
  })

  it('runs a complete multi-step chain before current header and event validation', async () => {
    const calls: string[] = []
    const { decodeStoredSession } = await configuredDecoder(
      2,
      [migration(0, calls), migration(1, calls)],
      calls,
    )
    const stored = storedSource(0, eventLog())

    const decoded = decodeStoredSession(stored.source, id)
    expect(decoded.meta.version).toBe(2)
    expect(calls).toEqual(['header:0', 'header:1', 'validate-header'])

    const events = await collectEvents(decoded.events)
    await decoded.completed

    expect(calls).toEqual([
      'header:0',
      'header:1',
      'validate-header',
      'events:0',
      'events:1',
    ])
    expect(events[0]?.data).toMatchObject({ migratedFrom0: true, migratedFrom1: true })
    expect(events[0]?.data).toMatchObject({ migrationPath: [0, 1] })
  })

  it('detaches each migration output before the next migration mutates its input', async () => {
    const retained: Array<Record<string, unknown>> = []
    const first = defineMigration(0, () => ({
      header: meta => ({ ...(meta as Record<string, unknown>), version: 1 }),
      event(value) {
        const event = value as SessionEvent
        const output = {
          ...event,
          data: { ...(event.data as Record<string, unknown>), first: true },
        }
        retained.push(output.data)
        return output
      },
    }))
    const second = defineMigration(1, () => ({
      header: meta => ({ ...(meta as Record<string, unknown>), version: 2 }),
      event(value) {
        const event = value as SessionEvent
        const data = event.data as Record<string, unknown>
        data['second'] = true
        return event
      },
    }))
    const { decodeStoredSession } = await configuredDecoder(2, [first, second])

    const decoded = decodeStoredSession(storedSource(0, eventLog()).source, id)
    const events = await collectEvents(decoded.events)
    await decoded.completed

    expect(events.every(event => (event.data as Record<string, unknown>)['second'] === true)).toBe(true)
    expect(retained.every(data => data['second'] === undefined)).toBe(true)
  })

  it('rejects a non-JSON event output before a later migration can repair it', async () => {
    const first = defineMigration(0, () => ({
      header: meta => ({ ...(meta as Record<string, unknown>), version: 1 }),
      event(value) {
        const event = value as SessionEvent
        return {
          ...event,
          data: { ...(event.data as Record<string, unknown>), transient: undefined },
        }
      },
    }))
    const second = defineMigration(1, () => ({
      header: meta => ({ ...(meta as Record<string, unknown>), version: 2 }),
      event(value) {
        const event = value as SessionEvent
        const data = event.data as Record<string, unknown>
        delete data['transient']
        return event
      },
    }))
    const { decodeStoredSession } = await configuredDecoder(2, [first, second])

    const failure = await decodedFailure(
      decodeStoredSession(storedSource(0, eventLog()).source, id),
    )

    expect(failure.message).toMatch(/event migration v0 -> v1 failed at seq 0/)
    expect((failure.cause as Error).message).toMatch(/not losslessly JSON-serializable/)
  })

  it('plans by version even when registry entries are declared out of order', async () => {
    const calls: string[] = []
    const { decodeStoredSession } = await configuredDecoder(
      2,
      [migration(1, calls), migration(0, calls)],
      calls,
    )

    const decoded = decodeStoredSession(storedSource(0, eventLog()).source, id)
    const events = await collectEvents(decoded.events)
    await decoded.completed

    expect(calls.slice(0, 3)).toEqual(['header:0', 'header:1', 'validate-header'])
    expect(events[0]?.data).toMatchObject({ migrationPath: [0, 1] })
  })

  it('starts a multi-version registry at the source version', async () => {
    const calls: string[] = []
    const { decodeStoredSession } = await configuredDecoder(
      2,
      [migration(0, calls), migration(1, calls)],
      calls,
    )
    const stored = storedSource(1, eventLog())

    const decoded = decodeStoredSession(stored.source, id, 1)
    const events = await collectEvents(decoded.events)
    await decoded.completed

    expect(calls).toEqual(['header:1', 'validate-header', 'events:1'])
    expect(stored.reads).toEqual([0])
    expect(events[0]?.data).toMatchObject({ migrationPath: [1] })
  })

  it('retains instance state from the header through events and finishes at EOF', async () => {
    const calls: string[] = []
    const Migration = defineMigration(0, () => {
      let headerId: SessionId | undefined
      let migratedEvents = 0
      return {
        header(meta) {
          calls.push('header')
          headerId = SessionId((meta as Record<string, unknown>)['id'] as string)
          return { ...(meta as Record<string, unknown>), version: 1 }
        },
        event(value) {
          calls.push(`event:${migratedEvents}`)
          migratedEvents += 1
          return {
            ...(value as SessionEvent),
            data: { ...(value as SessionEvent).data, headerId, migratedEvents },
          }
        },
        finish() {
          calls.push(`finish:${migratedEvents}`)
        },
      }
    })
    const { decodeStoredSession } = await configuredDecoder(1, [Migration])

    const decoded = decodeStoredSession(storedSource(0, eventLog()).source, id)
    expect(calls).toEqual(['header'])
    const events = await collectEvents(decoded.events)
    await decoded.completed

    expect(calls).toEqual(['header', 'event:0', 'event:1', 'finish:2'])
    expect(events.map(event => event.data)).toMatchObject([
      { headerId: id, migratedEvents: 1 },
      { headerId: id, migratedEvents: 2 },
    ])
  })

  it('migrates and validates a header without requiring an event source', async () => {
    const calls: string[] = []
    const first = defineMigration(0, () => ({
      header(meta) {
        calls.push('header:0')
        return { ...(meta as Record<string, unknown>), version: 1 }
      },
      event: value => value,
      finish() {
        calls.push('finish:0')
      },
    }))
    const { decodeStoredSessionHeader } = await configuredDecoder(
      2,
      [first, migration(1, calls)],
      calls,
    )

    const header = decodeStoredSessionHeader({ version: 0, id, createdAt: 1 }, id)

    expect(header.version).toBe(2)
    expect(calls).toEqual(['header:0', 'header:1', 'validate-header'])
  })

  it('allows a migration instance without finish', async () => {
    class MigrationWithoutFinish implements SessionFormatMigrationInstance {
      static readonly from = 0
      static readonly to = 1

      header(meta: unknown): unknown {
        return { ...(meta as Record<string, unknown>), version: 1 }
      }

      event(value: unknown): unknown {
        return value
      }
    }
    const { decodeStoredSession } = await configuredDecoder(1, [MigrationWithoutFinish])

    const decoded = decodeStoredSession(storedSource(0, eventLog()).source, id)

    await expect(collectEvents(decoded.events)).resolves.toEqual(eventLog())
    await expect(decoded.completed).resolves.toEqual({})
  })

  it('applies event migration before the current event vocabulary check', async () => {
    const step = defineMigration(0, () => ({
      header: meta => ({ ...(meta as Record<string, unknown>), version: 1 }),
      event(value) {
        const event = value as Record<string, unknown>
        return { ...event, type: 'turn/start', data: { turn: 1 } }
      },
    }))
    const { decodeStoredSession } = await configuredDecoder(1, [step])
    const stored = storedSource(0, [
      { type: 'legacy/turn-begin', seq: 0, time: 1, data: { legacyTurn: 1 } },
    ])

    const decoded = decodeStoredSession(stored.source, id)
    const events = await collectEvents(decoded.events)
    await decoded.completed

    expect(events).toEqual([
      { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } },
    ])
  })

  it('uses suffix access directly for the current format', async () => {
    const { decodeStoredSession } = await configuredDecoder(2, [])
    const stored = storedSource(2, eventLog())

    const decoded = decodeStoredSession(stored.source, id, 1)
    const events = await collectEvents(decoded.events)
    await decoded.completed

    expect(stored.reads).toEqual([1])
    expect(events).toEqual(eventLog().slice(1))
  })

  it('buffers a safe current-v0 suffix once without reopening the prefix', async () => {
    const { decodeStoredSession } = await configuredDecoder(0, [])
    const stored = storedSource(0, eventLog())

    const decoded = decodeStoredSession(stored.source, id, 1)
    expect(await collectEvents(decoded.events)).toEqual(eventLog().slice(1))
    await expect(decoded.completed).resolves.toEqual({})

    expect(stored.reads).toEqual([1])
  })

  it('reopens the complete current-v0 log when a legacy suffix record needs its prefix', async () => {
    const { decodeStoredSession } = await configuredDecoder(0, [])
    const legacy = [
      { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } },
      {
        type: 'steering/message',
        seq: 1,
        time: 2,
        data: { turn: 1, content: [{ type: 'text', text: 'continue' }], source: { kind: 'user' } },
      },
    ]
    const stored = storedSource(0, legacy)

    const decoded = decodeStoredSession(stored.source, id, 1)
    const events = await collectEvents(decoded.events)
    await decoded.completed

    expect(stored.reads).toEqual([1, 0])
    expect(events).toMatchObject([{ type: 'user/message', seq: 1 }])
  })

  it('observes a failed physical completion after reopening a required v0 prefix', async () => {
    const failure = new SessionPersistenceRevisionConflictError('reopened prefix changed')
    const fullCompletion = Promise.withResolvers<StoredEventReadCompletion<never>>()
    const source: StoredSessionSource<never> = {
      meta: { version: 0, id, createdAt: 1 },
      revision: SessionPersistenceRevision('prefix-conflict'),
      readEvents({ fromSeq = 0 } = {}) {
        if (fromSeq > 0) {
          return {
            events: (async function* (): AsyncIterable<unknown> {
              yield {
                type: 'steering/message', seq: 1, time: 2,
                data: { turn: 1, content: [], source: { kind: 'user' } },
              }
            })(),
            completed: Promise.resolve({}),
          }
        }
        return {
          events: (async function* (): AsyncIterable<unknown> {
            fullCompletion.reject(failure)
            throw failure
          })(),
          completed: fullCompletion.promise,
        }
      },
    }
    const { decodeStoredSession } = await configuredDecoder(0, [])
    const decoded = decodeStoredSession(source, id, 1)

    await expect(decodedFailure(decoded)).resolves.toBe(failure)
  })

  it('classifies every v0 prefix-independent suffix value without assuming a record', () => {
    const compatibility = unversionedFormatCompatibility(0)
    if (compatibility === undefined) throw new Error('v0 compatibility must be registered')

    expect(compatibility.requiresPrefix(null)).toBe(false)
    expect(compatibility.requiresPrefix({ type: 'turn/end', data: null })).toBe(false)
    expect(compatibility.requiresPrefix({ type: 'user/message', data: { id: 'current', content: [] } })).toBe(false)
    expect(compatibility.requiresPrefix({ type: 'user/message', data: { content: [] } })).toBe(true)
    expect(compatibility.requiresPrefix({ type: 'assistant/message', data: { content: [] } })).toBe(true)
    expect(compatibility.requiresPrefix({ type: 'tool/result', data: { callId: 'call' } })).toBe(true)
  })

  it('preserves already-canonical v0 turn-end reasons', async () => {
    const compatibility = unversionedFormatCompatibility(0)
    if (compatibility === undefined) throw new Error('v0 compatibility must be registered')
    const events = [
      {
        type: 'turn/end', seq: 0, time: 1,
        data: { turn: 1, reason: { kind: 'aborted', reason: { kind: 'disposed' } } },
      },
      {
        type: 'turn/end', seq: 1, time: 2,
        data: { turn: 2, reason: { kind: 'error', error: { message: 'failed', code: 'UNKNOWN' } } },
      },
    ]
    const input = (async function* (): AsyncIterable<unknown> {
      yield* events
    })()
    const canonical: unknown[] = []

    for await (const event of compatibility.canonicalizeEvents(input, id)) canonical.push(event)

    expect(canonical).toEqual(events)
  })

  it('canonicalizes every historical compact event name without changing its record', async () => {
    const { decodeStoredSession } = await configuredDecoder(0, [])
    const events = [
      {
        type: 'compact/start', seq: 0, time: 1,
        data: { compactionId: 'legacy', turn: 1 },
        surfaceOp: { op: 'retain' },
      },
      {
        type: 'compact/summary', seq: 1, time: 2,
        data: { summary: 'old summary', shadowedSeqs: [7, 8] },
        durableMetadata: { source: 'historical-v0' },
      },
      {
        type: 'compaction/end', seq: 2, time: 3,
        data: { compactionId: 'current', turn: 1 },
      },
      {
        type: 'compact/end', seq: 3, time: 4,
        data: { compactionId: 'legacy', turn: 1 },
      },
      {
        type: 'compact/prune', seq: 4, time: 5,
        data: {
          shadowedRange: { start: 7, end: 8 },
          shadowedSeqs: [7, 8],
          shadowedTokenCount: 456,
        },
      },
    ]
    const stored = storedSource(0, events)

    const decoded = decodeStoredSession(stored.source, id)
    const canonical = await collectEvents(decoded.events)
    await decoded.completed

    expect(canonical).toEqual(events.map(event => ({
      ...event,
      type: event.type.replace(/^compact\//, 'compaction/'),
    })))
    expect(stored.reads).toEqual([0])
  })

  it('still rejects other unknown v0 event names after compaction normalization', async () => {
    const { decodeStoredSession } = await configuredDecoder(0, [])
    const decoded = decodeStoredSession(storedSource(0, [{
      type: 'compact/future', seq: 0, time: 1, data: {},
    }]).source, id)

    const failure = await decodedFailure(decoded)
    expect(failure.message).toMatch(/event type "compact\/future".*not marked ignorable/)
  })

  it('does not run older registered steps for an already-current source', async () => {
    const calls: string[] = []
    const { decodeStoredSession } = await configuredDecoder(
      2,
      [migration(0, calls), migration(1, calls)],
      calls,
    )
    const stored = storedSource(2, eventLog())

    const decoded = decodeStoredSession(stored.source, id, 1)
    await collectEvents(decoded.events)
    await decoded.completed

    expect(calls).toEqual(['validate-header'])
    expect(stored.reads).toEqual([1])
  })

  it('opens a fresh revision-bound reader for each decode of the same source', async () => {
    const { decodeStoredSession } = await configuredDecoder(0, [])
    const stored = storedSource(0, eventLog())

    const first = decodeStoredSession(stored.source, id)
    expect(await collectEvents(first.events)).toEqual(eventLog())
    await first.completed
    const second = decodeStoredSession(stored.source, id)
    expect(await collectEvents(second.events)).toEqual(eventLog())
    await second.completed

    expect(first.revision).toBe(second.revision)
    expect(stored.reads).toEqual([0, 0])
  })

  it('propagates a physical revision conflict unchanged through events and completion', async () => {
    const failure = new SessionPersistenceRevisionConflictError('source changed')
    const physicalCompletion = Promise.withResolvers<StoredEventReadCompletion<never>>()
    const source: StoredSessionSource<never> = {
      meta: { version: 0, id, createdAt: 1 },
      revision: SessionPersistenceRevision('conflicting-source'),
      readEvents: () => ({
        events: (async function* (): AsyncIterable<unknown> {
          physicalCompletion.reject(failure)
          throw failure
        })(),
        completed: physicalCompletion.promise,
      }),
    }
    const { decodeStoredSession } = await configuredDecoder(0, [])
    const decoded = decodeStoredSession(source, id)
    const completion = decoded.completed.catch((error: unknown) => error)
    const consumption = collectEvents(decoded.events).catch((error: unknown) => error)

    const [streamFailure, completionFailure] = await Promise.all([consumption, completion])
    expect(streamFailure).toBe(failure)
    expect(completionFailure).toBe(failure)
  })

  it('propagates an upstream revision conflict unchanged through a migration step', async () => {
    const { decodeStoredSession } = await configuredDecoder(1, [migration(0, [])])
    const { SessionPersistenceRevisionConflictError: DecoderRevisionConflictError } = await import('../src/revision.ts')
    const failure = new DecoderRevisionConflictError('migrating source changed')
    const source: StoredSessionSource<never> = {
      meta: { version: 0, id, createdAt: 1 },
      revision: SessionPersistenceRevision('conflicting-migration-source'),
      readEvents: () => ({
        events: (async function* (): AsyncIterable<unknown> {
          throw failure
        })(),
        completed: Promise.reject(failure),
      }),
    }
    await expect(decodedFailure(decodeStoredSession(source, id))).resolves.toBe(failure)
  })

  it('rejects a missing path and a future source in the correct direction', async () => {
    const { decodeStoredSession, validateHeader } = await configuredDecoder(2, [])
    const old = storedSource(0, [])
    const future = storedSource(3, [])

    expect(() => decodeStoredSession(old.source, id))
      .toThrow(/missing v0 -> v1/)
    expect(() => decodeStoredSession(future.source, id))
      .toThrow(/newer harness/)
    expect(old.reads).toEqual([])
    expect(future.reads).toEqual([])
    expect(validateHeader).not.toHaveBeenCalled()
  })

  it('preserves the raw location in unsupported-format diagnostics', async () => {
    const { decodeStoredSession } = await configuredDecoder(0, [])
    const stored = storedSource(1, [])
    const location = { kind: 'jsonl', path: '/tmp/session.jsonl' }
    const source: StoredSessionSource<never> = { ...stored.source, location }

    let failure: unknown
    try {
      decodeStoredSession(source, id)
    } catch (error: unknown) {
      failure = error
    }
    expect(failure).toMatchObject({
      name: 'SessionFormatUnsupportedError',
      location,
    })
    expect((failure as Error).message).toContain('(raw log: /tmp/session.jsonl)')
  })

  it('rejects an invalid suffix before validating the header or opening events', async () => {
    const { decodeStoredSession, validateHeader } = await configuredDecoder(0, [])
    const stored = storedSource(0, eventLog())

    for (const fromSeq of [-1, 1.5, Number.MAX_SAFE_INTEGER + 1]) {
      expect(() => decodeStoredSession(stored.source, id, fromSeq))
        .toThrow(/fromSeq must be a non-negative safe integer/)
    }
    expect(validateHeader).not.toHaveBeenCalled()
    expect(stored.reads).toEqual([])
  })

  it('validates unknown durable header fields before path selection', async () => {
    const { decodeStoredSession, validateHeader } = await configuredDecoder(0, [])
    const cases: Array<{ meta: unknown; message: RegExp }> = [
      { meta: null, message: /header is not a lossless JSON record/ },
      { meta: { version: '0', id }, message: /invalid format version/ },
      { meta: { version: 0, id: 42 }, message: /has no string id/ },
    ]
    let reads = 0

    for (const entry of cases) {
      const source: StoredSessionSource<never> = {
        meta: entry.meta,
        revision: SessionPersistenceRevision('invalid-header'),
        readEvents: () => {
          reads += 1
          return { events: (async function* () {})(), completed: Promise.resolve({}) }
        },
      }
      expect(() => decodeStoredSession(source, id)).toThrow(entry.message)
    }
    expect(reads).toBe(0)
    expect(validateHeader).not.toHaveBeenCalled()
  })

  it('rejects every malformed current event envelope through the stream and completion', async () => {
    const { decodeStoredSession } = await configuredDecoder(1, [])
    const cases: Array<{ value: unknown; message: RegExp }> = [
      { value: null, message: /non-record event/ },
      { value: { seq: 0, time: 1, data: {} }, message: /without a string type/ },
      { value: { type: 'turn/start', seq: -1, time: 1, data: {} }, message: /invalid seq -1/ },
      { value: { type: 'turn/start', seq: 0, time: 'now', data: {} }, message: /invalid time/ },
      { value: { type: 'turn/start', seq: 0, time: 1 }, message: /without data/ },
    ]

    for (const entry of cases) {
      const decoded = decodeStoredSession(storedSource(1, [entry.value]).source, id)
      expect((await decodedFailure(decoded)).message).toMatch(entry.message)
    }
  })

  it('rejects a stored event that cannot be represented as JSON', async () => {
    const { decodeStoredSession } = await configuredDecoder(1, [])
    const decoded = decodeStoredSession(storedSource(1, [undefined]).source, id)

    expect((await decodedFailure(decoded)).message).toMatch(/not losslessly JSON-serializable/)
  })

  it('rejects every malformed v0 event before same-version canonicalization', async () => {
    const { decodeStoredSession } = await configuredDecoder(0, [])
    const cases: Array<{ value: unknown; message: RegExp }> = [
      { value: null, message: /non-record event/ },
      { value: { seq: 0, time: 1, data: {} }, message: /without a string type/ },
      { value: { type: 'turn/start', seq: -1, time: 1, data: {} }, message: /invalid seq -1/ },
      { value: { type: 'turn/start', seq: 0, time: 'now', data: {} }, message: /invalid time/ },
      { value: { type: 'turn/start', seq: 0, time: 1 }, message: /without data/ },
    ]

    for (const entry of cases) {
      const decoded = decodeStoredSession(storedSource(0, [entry.value]).source, id)
      expect((await decodedFailure(decoded)).message).toMatch(entry.message)
    }
  })

  it('lets a v0 turn/end with opaque data reach current validation unchanged', async () => {
    const { decodeStoredSession } = await configuredDecoder(0, [])
    const decoded = decodeStoredSession(storedSource(0, [
      { type: 'turn/end', seq: 0, time: 1, data: null },
    ]).source, id)

    await expect(collectEvents(decoded.events)).resolves.toEqual([
      { type: 'turn/end', seq: 0, time: 1, data: null },
    ])
    await expect(decoded.completed).resolves.toEqual({})
  })

  it('rejects a migration that returns the wrong header version', async () => {
    const bad = defineMigration(0, () => ({
      header: meta => ({ ...(meta as Record<string, unknown>), version: 0 }),
      event: value => value,
    }))
    const first = await configuredDecoder(1, [bad])

    expect(() => first.decodeStoredSession(storedSource(0, []).source, id))
      .toThrow(/returned header version 0/)
    expect(first.validateHeader).not.toHaveBeenCalled()

    const calls: string[] = []
    const badSecond = defineMigration(1, () => ({
      header: meta => ({ ...(meta as Record<string, unknown>), version: 1 }),
      event: value => value,
    }))
    const second = await configuredDecoder(2, [migration(0, calls), badSecond], calls)
    const stored = storedSource(0, [])
    expect(() => second.decodeStoredSession(stored.source, id))
      .toThrow(/v1 -> v2 returned header version 1/)
    expect(calls).toEqual(['header:0'])
    expect(second.validateHeader).not.toHaveBeenCalled()
    expect(stored.reads).toEqual([])
  })

  it('rejects a migration that changes the session id or cwd storage identity', async () => {
    const changedId = defineMigration(0, () => ({
      header: meta => ({ ...(meta as Record<string, unknown>), version: 1, id: 'other' }),
      event: value => value,
    }))
    const first = await configuredDecoder(1, [changedId])
    expect(() => first.decodeStoredSession(storedSource(0, []).source, id))
      .toThrow(/changed session storage identity/)

    const changedCwd = defineMigration(0, () => ({
      header: meta => ({ ...(meta as Record<string, unknown>), version: 1, cwd: '/other' }),
      event: value => value,
    }))
    const second = await configuredDecoder(1, [changedCwd])
    const stored = storedSource(0, [])
    stored.meta['cwd'] = '/work'
    expect(() => second.decodeStoredSession(stored.source, id))
      .toThrow(/changed session storage identity/)
  })

  it('wraps a header migration failure with the failing version step', async () => {
    const cause = new Error('bad legacy header')
    const step = defineMigration(0, () => ({
      header: () => { throw cause },
      event: value => value,
    }))
    const { decodeStoredSession } = await configuredDecoder(1, [step])

    let failure: unknown
    try {
      decodeStoredSession(storedSource(0, []).source, id)
    } catch (error: unknown) {
      failure = error
    }
    expect(failure).toMatchObject({
      message: `session "${id}" header migration v0 -> v1 failed`,
      cause,
    })
  })

  it('mirrors an event migration failure through the stream and completion promise', async () => {
    const cause = new Error('bad legacy event')
    const step = defineMigration(0, () => ({
      header: meta => ({ ...(meta as Record<string, unknown>), version: 1 }),
      event: () => { throw cause },
    }))
    const { decodeStoredSession } = await configuredDecoder(1, [step])
    const decoded = decodeStoredSession(storedSource(0, eventLog()).source, id)
    const completion = decoded.completed.catch((error: unknown) => error)
    const consumption = collectEvents(decoded.events).catch((error: unknown) => error)

    const [streamFailure, completionFailure] = await Promise.all([consumption, completion])
    expect(streamFailure).toBe(completionFailure)
    expect(streamFailure).toMatchObject({
      message: `session "${id}" event migration v0 -> v1 failed at seq 0`,
      cause,
    })
  })

  it('mirrors a finish failure through the stream and completion promise', async () => {
    const cause = new Error('unclosed legacy state')
    const Migration = defineMigration(0, () => ({
      header: meta => ({ ...(meta as Record<string, unknown>), version: 1 }),
      event: value => value,
      finish: () => { throw cause },
    }))
    const { decodeStoredSession } = await configuredDecoder(1, [Migration])

    const failure = await decodedFailure(decodeStoredSession(storedSource(0, eventLog()).source, id))
    expect(failure).toMatchObject({
      message: `session "${id}" event migration v0 -> v1 failed at EOF`,
      cause,
    })
  })

  it('rejects a migration that changes an event sequence number', async () => {
    const step = defineMigration(0, () => ({
      header: meta => ({ ...(meta as Record<string, unknown>), version: 1 }),
      event(value) {
        const event = value as SessionEvent
        return { ...event, seq: event.seq + 1 }
      },
    }))
    const { decodeStoredSession } = await configuredDecoder(1, [step])
    const decoded = decodeStoredSession(storedSource(0, eventLog()).source, id)
    const completion = decoded.completed.catch((error: unknown) => error)
    const consumption = collectEvents(decoded.events).catch((error: unknown) => error)

    const [streamFailure, completionFailure] = await Promise.all([consumption, completion])
    expect(streamFailure).toBe(completionFailure)
    expect((streamFailure as Error).message).toMatch(/changed event seq 0 to 1/)
  })

  it('rejects a non-contiguous current-format event sequence', async () => {
    const { decodeStoredSession } = await configuredDecoder(0, [])
    const stored = storedSource(0, [
      { type: 'turn/start', seq: 1, time: 1, data: { turn: 1 } },
    ])

    const failure = await decodedFailure(decodeStoredSession(stored.source, id))

    expect(failure.message).toContain(`session "${id}" event seq mismatch: expected 0, got 1`)
  })

  it('runs current header validation only after the final header step', async () => {
    const calls: string[] = []
    const finalStep = defineMigration(1, () => ({
      header(meta) {
        calls.push('header:1')
        const { createdAt: _createdAt, ...rest } = meta as Record<string, unknown>
        return { ...rest, version: 2 }
      },
      event: value => value,
    }))
    const { decodeStoredSession } = await configuredDecoder(
      2,
      [migration(0, calls), finalStep],
      calls,
    )
    const stored = storedSource(0, eventLog())

    expect(() => decodeStoredSession(stored.source, id))
      .toThrow(/current header validator received invalid createdAt/)
    expect(calls).toEqual(['header:0', 'header:1', 'validate-header'])
    expect(stored.reads).toEqual([])
  })

  it('detaches stored header and event objects before a mutating migration runs', async () => {
    const originalEvents = eventLog()
    const eventSnapshot = structuredClone(originalEvents)
    const step = defineMigration(0, () => ({
      header(meta) {
        const record = meta as Record<string, unknown>
        record['version'] = 1
        return record
      },
      event(value) {
        const event = value as SessionEvent
        const data = event.data as Record<string, unknown>
        data['mutated'] = true
        return event
      },
    }))
    const { decodeStoredSession } = await configuredDecoder(1, [step])
    const stored = storedSource(0, originalEvents)

    const decoded = decodeStoredSession(stored.source, id)
    const migrated = await collectEvents(decoded.events)
    await decoded.completed

    expect(migrated.every(event => (event.data as Record<string, unknown>)['mutated'] === true)).toBe(true)
    expect(stored.meta).toEqual({ version: 0, id, createdAt: 1 })
    expect(originalEvents).toEqual(eventSnapshot)
  })

  it('rejects duplicate, invalid, and future-targeting static registries at initialization', async () => {
    const calls: string[] = []
    await expect(configuredDecoder(1, [migration(0, calls), migration(0, calls)]))
      .rejects.toThrow(/duplicate Session format migration/)

    await expect(configuredDecoder(1, [migration(-1, calls)]))
      .rejects.toThrow(/adjacent non-negative version/)

    const nonAdjacent = migration(0, calls, 2)
    await expect(configuredDecoder(2, [nonAdjacent]))
      .rejects.toThrow(/adjacent non-negative version/)

    const fractional = migration(0.5, calls, 1.5)
    await expect(configuredDecoder(2, [fractional]))
      .rejects.toThrow(/adjacent non-negative version/)

    await expect(configuredDecoder(1, [migration(1, calls)]))
      .rejects.toThrow(/targets a version newer than this build/)
  })

  it('initializes with a gapped registry and refuses only sessions at or below the gap', async () => {
    const calls: string[] = []
    const { decodeStoredSession } = await configuredDecoder(
      3,
      [migration(0, calls), migration(2, calls)],
      calls,
    )
    const current = storedSource(3, [])
    const pastGap = storedSource(2, eventLog())
    const atGap = storedSource(1, eventLog())
    const belowGap = storedSource(0, eventLog())

    expect(decodeStoredSession(current.source, id).meta.version).toBe(3)
    const decoded = decodeStoredSession(pastGap.source, id)
    expect(decoded.sourceVersion).toBe(2)
    expect(decoded.meta.version).toBe(3)
    expect(() => decodeStoredSession(atGap.source, id))
      .toThrow(/missing v1 -> v2/)
    expect(() => decodeStoredSession(belowGap.source, id))
      .toThrow(/missing v1 -> v2/)
    expect(pastGap.reads).toEqual([])
  })
})