|
|
@@ -0,0 +1,1101 @@
|
|
|
+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([])
|
|
|
+ })
|
|
|
+})
|