|
|
@@ -0,0 +1,889 @@
|
|
|
+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 {
|
|
|
+ SessionFormatContext,
|
|
|
+ SessionFormatStep,
|
|
|
+ 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')
|
|
|
+
|
|
|
+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 migration(
|
|
|
+ from: number,
|
|
|
+ calls: string[],
|
|
|
+): SessionFormatStep {
|
|
|
+ return {
|
|
|
+ from,
|
|
|
+ to: from + 1,
|
|
|
+ migrateHeader(meta) {
|
|
|
+ calls.push(`header:${from}`)
|
|
|
+ return { ...(meta as Record<string, unknown>), version: from + 1 }
|
|
|
+ },
|
|
|
+ migrateEvents(events) {
|
|
|
+ return (async function* (): AsyncIterable<unknown> {
|
|
|
+ let observedInput = false
|
|
|
+ for await (const value of events) {
|
|
|
+ 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[]
|
|
|
+ : []
|
|
|
+ yield {
|
|
|
+ ...event,
|
|
|
+ data: {
|
|
|
+ ...data,
|
|
|
+ [`migratedFrom${from}`]: true,
|
|
|
+ migrationPath: [...migrationPath, from],
|
|
|
+ },
|
|
|
+ }
|
|
|
+ }
|
|
|
+ })()
|
|
|
+ },
|
|
|
+ }
|
|
|
+}
|
|
|
+
|
|
|
+async function configuredDecoder(
|
|
|
+ currentVersion: number,
|
|
|
+ migrations: readonly SessionFormatStep[],
|
|
|
+ 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_STEPS: 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: SessionFormatStep = {
|
|
|
+ from: 0,
|
|
|
+ to: 1,
|
|
|
+ migrateHeader: meta => ({ ...(meta as Record<string, unknown>), version: 1 }),
|
|
|
+ migrateEvents: events => (async function* (): AsyncIterable<unknown> {
|
|
|
+ let previousSeq: number | undefined
|
|
|
+ for await (const value of events) {
|
|
|
+ const event = value as SessionEvent
|
|
|
+ yield previousSeq === undefined
|
|
|
+ ? event
|
|
|
+ : { ...event, data: { ...event.data, previousSeq } }
|
|
|
+ previousSeq = event.seq
|
|
|
+ }
|
|
|
+ })(),
|
|
|
+ }
|
|
|
+ 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('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('passes the stable stored identity to every header and event step', async () => {
|
|
|
+ const contexts: SessionFormatContext[] = []
|
|
|
+ const step: SessionFormatStep = {
|
|
|
+ from: 0,
|
|
|
+ to: 1,
|
|
|
+ migrateHeader(meta, context) {
|
|
|
+ contexts.push(context)
|
|
|
+ return { ...(meta as Record<string, unknown>), version: 1 }
|
|
|
+ },
|
|
|
+ migrateEvents(events, context) {
|
|
|
+ contexts.push(context)
|
|
|
+ return events
|
|
|
+ },
|
|
|
+ }
|
|
|
+ const { decodeStoredSession } = await configuredDecoder(1, [step])
|
|
|
+
|
|
|
+ const decoded = decodeStoredSession(storedSource(0, eventLog()).source, id)
|
|
|
+ await collectEvents(decoded.events)
|
|
|
+ await decoded.completed
|
|
|
+
|
|
|
+ expect(contexts).toEqual([{ sessionId: id }, { sessionId: id }])
|
|
|
+ })
|
|
|
+
|
|
|
+ it('migrates and validates a header without requiring an event source', async () => {
|
|
|
+ const calls: string[] = []
|
|
|
+ const { decodeStoredSessionHeader } = await configuredDecoder(
|
|
|
+ 2,
|
|
|
+ [migration(0, calls), 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('applies event migration before the current event vocabulary check', async () => {
|
|
|
+ const step: SessionFormatStep = {
|
|
|
+ from: 0,
|
|
|
+ to: 1,
|
|
|
+ migrateHeader: meta => ({ ...(meta as Record<string, unknown>), version: 1 }),
|
|
|
+ migrateEvents: events => (async function* (): AsyncIterable<unknown> {
|
|
|
+ for await (const value of events) {
|
|
|
+ const event = value as Record<string, unknown>
|
|
|
+ yield { ...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('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('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 step that returns the wrong header version', async () => {
|
|
|
+ const bad: SessionFormatStep = {
|
|
|
+ from: 0,
|
|
|
+ to: 1,
|
|
|
+ migrateHeader: meta => ({ ...(meta as Record<string, unknown>), version: 0 }),
|
|
|
+ migrateEvents: events => events,
|
|
|
+ }
|
|
|
+ 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: SessionFormatStep = {
|
|
|
+ from: 1,
|
|
|
+ to: 2,
|
|
|
+ migrateHeader: meta => ({ ...(meta as Record<string, unknown>), version: 1 }),
|
|
|
+ migrateEvents: events => events,
|
|
|
+ }
|
|
|
+ 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 step that changes the session id or cwd storage identity', async () => {
|
|
|
+ const changedId: SessionFormatStep = {
|
|
|
+ from: 0,
|
|
|
+ to: 1,
|
|
|
+ migrateHeader: meta => ({ ...(meta as Record<string, unknown>), version: 1, id: 'other' }),
|
|
|
+ migrateEvents: events => events,
|
|
|
+ }
|
|
|
+ const first = await configuredDecoder(1, [changedId])
|
|
|
+ expect(() => first.decodeStoredSession(storedSource(0, []).source, id))
|
|
|
+ .toThrow(/changed session storage identity/)
|
|
|
+
|
|
|
+ const changedCwd: SessionFormatStep = {
|
|
|
+ from: 0,
|
|
|
+ to: 1,
|
|
|
+ migrateHeader: meta => ({ ...(meta as Record<string, unknown>), version: 1, cwd: '/other' }),
|
|
|
+ migrateEvents: events => events,
|
|
|
+ }
|
|
|
+ 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: SessionFormatStep = {
|
|
|
+ from: 0,
|
|
|
+ to: 1,
|
|
|
+ migrateHeader: () => { throw cause },
|
|
|
+ migrateEvents: events => events,
|
|
|
+ }
|
|
|
+ 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: SessionFormatStep = {
|
|
|
+ from: 0,
|
|
|
+ to: 1,
|
|
|
+ migrateHeader: meta => ({ ...(meta as Record<string, unknown>), version: 1 }),
|
|
|
+ migrateEvents: () => (async function* (): AsyncIterable<unknown> {
|
|
|
+ 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`,
|
|
|
+ cause,
|
|
|
+ })
|
|
|
+ })
|
|
|
+
|
|
|
+ it('runs current event validation on the migrated output', async () => {
|
|
|
+ const step: SessionFormatStep = {
|
|
|
+ from: 0,
|
|
|
+ to: 1,
|
|
|
+ migrateHeader: meta => ({ ...(meta as Record<string, unknown>), version: 1 }),
|
|
|
+ migrateEvents: events => (async function* (): AsyncIterable<unknown> {
|
|
|
+ for await (const value of events) {
|
|
|
+ const event = value as SessionEvent
|
|
|
+ yield { ...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(/expected 0, got 1/)
|
|
|
+ })
|
|
|
+
|
|
|
+ it('runs current header validation only after the final header step', async () => {
|
|
|
+ const calls: string[] = []
|
|
|
+ const finalStep: SessionFormatStep = {
|
|
|
+ from: 1,
|
|
|
+ to: 2,
|
|
|
+ migrateHeader(meta) {
|
|
|
+ calls.push('header:1')
|
|
|
+ const { createdAt: _createdAt, ...rest } = meta as Record<string, unknown>
|
|
|
+ return { ...rest, version: 2 }
|
|
|
+ },
|
|
|
+ migrateEvents: events => events,
|
|
|
+ }
|
|
|
+ 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: SessionFormatStep = {
|
|
|
+ from: 0,
|
|
|
+ to: 1,
|
|
|
+ migrateHeader(meta) {
|
|
|
+ const record = meta as Record<string, unknown>
|
|
|
+ record['version'] = 1
|
|
|
+ return record
|
|
|
+ },
|
|
|
+ migrateEvents: events => (async function* (): AsyncIterable<unknown> {
|
|
|
+ for await (const value of events) {
|
|
|
+ const event = value as SessionEvent
|
|
|
+ const data = event.data as Record<string, unknown>
|
|
|
+ data['mutated'] = true
|
|
|
+ yield 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, future-targeting, and incomplete static registries at initialization', async () => {
|
|
|
+ const calls: string[] = []
|
|
|
+ await expect(configuredDecoder(1, [migration(0, calls), migration(0, calls)]))
|
|
|
+ .rejects.toThrow(/duplicate Session format step/)
|
|
|
+
|
|
|
+ await expect(configuredDecoder(1, [migration(-1, calls)]))
|
|
|
+ .rejects.toThrow(/adjacent non-negative version/)
|
|
|
+
|
|
|
+ const nonAdjacent: SessionFormatStep = {
|
|
|
+ ...migration(0, calls),
|
|
|
+ to: 2,
|
|
|
+ }
|
|
|
+ await expect(configuredDecoder(2, [nonAdjacent]))
|
|
|
+ .rejects.toThrow(/adjacent non-negative version/)
|
|
|
+
|
|
|
+ const fractional: SessionFormatStep = {
|
|
|
+ ...migration(0, calls),
|
|
|
+ from: 0.5,
|
|
|
+ to: 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/)
|
|
|
+
|
|
|
+ await expect(configuredDecoder(2, [migration(0, calls)]))
|
|
|
+ .rejects.toThrow(/incomplete path/)
|
|
|
+ })
|
|
|
+})
|