test(session): cover V2 system prompt persistence migration

packages/session/session-persistence-jsonl/tests/v2-system-migration.spec.ts

@@ -0,0 +1,306 @@
+import { freezeMessage, MessageId } from '@deepseek-ai/dsh-llm'
+import { Context } from '@deepseek-ai/cordis'
+import { Session, SessionId, SessionLogOffset, SessionSeq } from '@deepseek-ai/dsh-session'
+import type { SessionEvent } from '@deepseek-ai/dsh-session'
+import { SessionFormatUnsupportedError } from '@deepseek-ai/dsh-session-persistence'
+import type { SessionHandle } from '@deepseek-ai/dsh-session-persistence'
+import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl'
+import { mkdir, mkdtemp, readFile, readdir, rm, stat, writeFile } from 'node:fs/promises'
+import { tmpdir } from 'node:os'
+import { dirname, join } from 'node:path'
+import { afterEach, beforeEach, describe, expect, it } from 'vitest'
+import { generationLogPath } from '../src/format.ts'
+
+const id = SessionId('v2-system-migration')
+const config = { provider: 'mock', model: 'mock' }
+let root: string
+const contexts: Context[] = []
+
+beforeEach(async () => {
+  root = await mkdtemp(join(tmpdir(), 'dsh-v2-system-migration-'))
+})
+
+afterEach(async () => {
+  try {
+    for (const ctx of contexts.splice(0).reverse()) await ctx.fiber.dispose()
+  } finally {
+    await rm(root, { recursive: true, force: true })
+  }
+})
+
+async function mount(): Promise<Context> {
+  const ctx = new Context()
+  contexts.push(ctx)
+  await ctx.plugin(JsonlSessionPersistence, { root, compression: 'none' })
+  return ctx
+}
+
+function request(system?: string) {
+  return {
+    type: 'request/header',
+    data: { header: { config, ...(system === undefined ? {} : { system }) }, reason: 'change' },
+  }
+}
+
+function user(text: string) {
+  return {
+    type: 'user/message', surfaceOp: 'append',
+    data: { id: text, role: 'user', source: { kind: 'user' }, content: [{ type: 'text', text }] },
+  }
+}
+
+async function writeV2(rows: readonly object[], seeded = false): Promise<string> {
+  const path = generationLogPath(root, undefined, id, 2, 'none')
+  await mkdir(dirname(path), { recursive: true })
+  const header = {
+    type: 'session', version: 2, id, createdAt: 1, isSeeded: seeded, delegationDepth: 0,
+    ...(seeded ? { parentSession: 'parent' } : {}),
+  }
+  await writeFile(path, [header, ...rows.map((row, seq) => ({ ...row, seq, time: seq + 10 }))]
+    .map(row => JSON.stringify(row)).join('\n') + '\n')
+  return path
+}
+
+async function observeFile(path: string) {
+  const identity = await stat(path, { bigint: true })
+  return {
+    bytes: await readFile(path), dev: identity.dev, ino: identity.ino,
+    size: identity.size, mtimeNs: identity.mtimeNs, ctimeNs: identity.ctimeNs,
+  }
+}
+
+async function restore(handle: SessionHandle): Promise<Session> {
+  const read = await handle.read()
+  return Session.fromRestore(id, read.events, handle.header, handle.inheritedEventCount, read.eventState)
+}
+
+function messages(session: Session) {
+  return session.deriveMessages().map(({ role, content }) => ({ role, content }))
+}
+
+function prompt(text: string) {
+  return { role: 'system', content: [{ type: 'text', text }] }
+}
+const human = { role: 'user', content: [{ type: 'text' as const, text: 'question' }] }
+
+function assertRequestHistory(events: readonly SessionEvent[], handle: SessionHandle) {
+  const requests = events.filter(event => event.type === 'request/header')
+  expect(requests).toHaveLength(3)
+  const projected = requests.map((event) => {
+    expect(event.data.header).toEqual({ config })
+    const prefix = Session.fromRestore(id, events.slice(0, event.seq + 1), handle.header, SessionLogOffset(0), 'shared-frozen')
+    return messages(prefix)
+  })
+  expect(projected).toEqual([[prompt('initial'), human], [prompt('changed'), human], [human]])
+}
+
+describe('V2 system prompts through current Session and JSONL persistence', () => {
+  it.each([undefined, ''])('prepares prompts and clear (%j) read-only, then publishes V3 without replacing V2', async (clear) => {
+    const sourcePath = await writeV2([
+      { type: 'turn/start', data: { turn: 1 } },
+      { type: 'step/start', data: { turn: 1, step: 1 } },
+      user('question'), request('initial'), request('changed'), request(clear),
+      { type: 'step/end', data: { turn: 1, step: 1 } },
+      { type: 'turn/end', data: { turn: 1, reason: { kind: 'completed' } } },
+    ])
+    const original = await observeFile(sourcePath)
+    const directory = dirname(sourcePath)
+    const directoryBefore = await readdir(directory)
+    const ctx = await mount()
+    const reader = await ctx.sessionPersistence.open(id, 'read')
+    let prepared: readonly SessionEvent[]
+    try {
+      expect(reader.header.version).toBe(3)
+      prepared = (await reader.read()).events
+      expect(prepared.map(event => event.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])
+      expect(prepared.filter(event => event.type !== 'system/message').map(event => [event.type, event.time])).toEqual([
+        ['turn/start', 10], ['step/start', 11], ['user/message', 12],
+        ['request/header', 13], ['request/header', 14], ['request/header', 15],
+        ['step/end', 16], ['turn/end', 17],
+      ])
+      expect(prepared.filter(event => event.type === 'system/message').map(event => ({
+        seq: event.seq, content: event.data.message.content,
+        surfaceOp: event.surfaceOp, sourceEventSeqs: event.sourceEventSeqs,
+      }))).toEqual([
+        { seq: 2, content: [], surfaceOp: 'append', sourceEventSeqs: undefined },
+        { seq: 4, content: prompt('initial').content, surfaceOp: { op: 'replace', start: 2, end: 2 }, sourceEventSeqs: [2] },
+        { seq: 6, content: prompt('changed').content, surfaceOp: { op: 'replace', start: 4, end: 4 }, sourceEventSeqs: [4] },
+        { seq: 8, content: [], surfaceOp: { op: 'replace', start: 6, end: 6 }, sourceEventSeqs: [6] },
+      ])
+      assertRequestHistory(prepared, reader)
+      const session = await restore(reader)
+      expect(messages(session)).toEqual([human])
+      expect(session.surface.nodes).toEqual([8, 3])
+      expect(session.eventAt(SessionSeq(8))?.type).toBe('system/message')
+    } finally {
+      await reader.close()
+    }
+    expect(await observeFile(sourcePath)).toEqual(original)
+    expect(await readdir(directory)).toEqual(directoryBefore)
+    const independent = await mount()
+    const independentReader = await independent.sessionPersistence.open(id, 'read')
+    try {
+      expect((await independentReader.read()).events).toEqual(prepared)
+      await independent.sessionPersistence.flush()
+    } finally {
+      await independentReader.close()
+    }
+    expect(await observeFile(sourcePath)).toEqual(original)
+    expect(await readdir(directory)).toEqual(directoryBefore)
+
+    const writer = await ctx.sessionPersistence.open(id, 'write')
+    try {
+      expect((await writer.read()).events).toEqual(prepared)
+    } finally {
+      await writer.close()
+    }
+    expect(await observeFile(sourcePath)).toEqual(original)
+    expect((await readdir(directory)).filter(name => name.endsWith('.jsonl')).sort())
+      .toEqual(['session.v2.jsonl', 'session.v3.jsonl'])
+    const publishedPath = generationLogPath(root, undefined, id, 3, 'none')
+    const published = (await readFile(publishedPath, 'utf8')).trimEnd().split('\n').map(line => JSON.parse(line) as unknown)
+    expect(published[0]).toMatchObject({ type: 'session', version: 3, id })
+    expect(published.slice(1)).toEqual(prepared)
+
+    const reopened = await mount()
+    const reopenedReader = await reopened.sessionPersistence.open(id, 'read')
+    try {
+      expect((await reopenedReader.read()).events).toEqual(prepared)
+      assertRequestHistory(prepared, reopenedReader)
+      const session = await restore(reopenedReader)
+      expect(messages(session)).toEqual([human])
+      expect(session.surface.nodes).toEqual([8, 3])
+    } finally {
+      await reopenedReader.close()
+    }
+    expect(await observeFile(sourcePath)).toEqual(original)
+  })
+
+  it('remaps local references and keeps the inherited cut stable through publication and a current Session append', async () => {
+    const sourcePath = await writeV2([
+      { type: 'turn/start', data: { turn: 1 } },
+      { type: 'step/start', data: { turn: 1, step: 1 } },
+      user('a'), request('seed prompt'), user('b'),
+      { type: 'compaction/prune', data: { shadowedRange: { start: 2, end: 4 }, shadowedSeqs: [2, 4], shadowedTokenCount: 20 } },
+      { ...user('summary'), surfaceOp: { op: 'replace', start: 2, end: 4 }, sourceEventSeqs: [2, 4] },
+      { type: 'command/run', data: { commandId: 'command', name: 'test', source: { kind: 'user' } } },
+      { type: 'command/done', data: { commandId: 'command', kind: 'success', sourceEventSeq: 6 } },
+      { type: 'session/title', data: { title: 'title', messageSeqs: [2, 4], source: { kind: 'fallback' } } },
+      { type: 'step/end', data: { turn: 1, step: 1 } },
+      { type: 'turn/end', data: { turn: 1, reason: { kind: 'completed' } } },
+      { type: 'session/end-seed', data: { inherited: true } },
+    ], true)
+    const original = await observeFile(sourcePath)
+    const ctx = await mount()
+    const reader = await ctx.sessionPersistence.open(id, 'read')
+    let prepared: readonly SessionEvent[]
+    try {
+      prepared = (await reader.read()).events
+      expect(reader.inheritedEventCount).toBe(14)
+      expect(prepared[7]).toMatchObject({ type: 'compaction/prune', data: { shadowedRange: { start: 3, end: 6 }, shadowedSeqs: [3, 6] } })
+      expect(prepared[8]).toMatchObject({ type: 'user/message', surfaceOp: { op: 'replace', start: 3, end: 6 }, sourceEventSeqs: [3, 6] })
+      expect(prepared[10]).toMatchObject({ type: 'command/done', data: { sourceEventSeq: 8 } })
+      expect(prepared[11]).toMatchObject({ type: 'session/title', data: { messageSeqs: [3, 6] } })
+      expect(prepared[14]).toMatchObject({ type: 'session/end-seed', seq: 14, data: { inherited: true } })
+      const session = await restore(reader)
+      expect(session.inheritedEventCount).toBe(14)
+      expect(session.firstLiveSeq).toBe(15)
+      expect(session.surface.nodes).toEqual([4, 8])
+      expect(messages(session)).toEqual([prompt('seed prompt'), { role: 'user', content: [{ type: 'text', text: 'summary' }] }])
+      expect(session.ownEvents()).toEqual([prepared[14]])
+    } finally {
+      await reader.close()
+    }
+    expect(await observeFile(sourcePath)).toEqual(original)
+    expect(await readdir(dirname(sourcePath))).toEqual(['session.v2.jsonl'])
+
+    const writer = await ctx.sessionPersistence.open(id, 'write')
+    let expected: readonly SessionEvent[]
+    try {
+      expect((await writer.read()).events).toEqual(prepared)
+      const session = await restore(writer)
+      session.append('turn/start', { turn: 2 })
+      session.append('step/start', { turn: 2, step: 1 })
+      session.append('system/message', {
+        turn: 2, step: 1,
+        message: freezeMessage({ role: 'system', id: MessageId('resumed-system'), content: [{ type: 'text', text: 'resumed prompt' }], source: { kind: 'plugin', plugin: '@deepseek-ai/dsh-system-prompt' } }),
+      }, { surfaceOp: { op: 'replace', start: SessionSeq(4), end: SessionSeq(4) }, sourceEventSeqs: [SessionSeq(4)] })
+      session.append('step/end', { turn: 2, step: 1 })
+      session.append('turn/end', { turn: 2, reason: { kind: 'completed' } })
+      expected = session.snapshotEvents()
+      await writer.append(session.snapshotEvents(session.firstLiveSeq))
+      await writer.flush()
+    } finally {
+      await writer.close()
+    }
+    const reopened = await mount()
+    const reopenedReader = await reopened.sessionPersistence.open(id, 'read')
+    try {
+      expect((await reopenedReader.read()).events).toEqual(expected)
+      expect(reopenedReader.inheritedEventCount).toBe(14)
+      const session = await restore(reopenedReader)
+      expect(session.inheritedEventCount).toBe(14)
+      expect(session.isOwnSeq(SessionSeq(13))).toBe(false)
+      expect(session.isOwnSeq(SessionSeq(14))).toBe(true)
+      expect(session.surface.nodes).toEqual([17, 8])
+      expect(messages(session)).toEqual([prompt('resumed prompt'), { role: 'user', content: [{ type: 'text', text: 'summary' }] }])
+    } finally {
+      await reopenedReader.close()
+    }
+    expect(await observeFile(sourcePath)).toEqual(original)
+  })
+
+  it.each(['read', 'write'] as const)('refuses unsupported pre-step V2 during %s open without falling back to V1', async (access) => {
+    const sourcePath = await writeV2([
+      { type: 'turn/start', data: { turn: 1 } }, user('too early'),
+      { type: 'step/start', data: { turn: 1, step: 1 } }, request('cannot reorder'),
+    ])
+    const lowerPath = generationLogPath(root, undefined, id, 1, 'none')
+    await writeFile(lowerPath, JSON.stringify({ type: 'session', version: 1, id, createdAt: 1, delegationDepth: 0 }) + '\n')
+    const original = await observeFile(sourcePath)
+    const lower = await observeFile(lowerPath)
+    const ctx = await mount()
+    await expect(ctx.sessionPersistence.open(id, access)).rejects.toBeInstanceOf(SessionFormatUnsupportedError)
+    await expect(ctx.sessionPersistence.open(id, access)).rejects.toThrow(/before first step/)
+    expect(await observeFile(sourcePath)).toEqual(original)
+    expect(await observeFile(lowerPath)).toEqual(lower)
+    expect((await readdir(dirname(sourcePath))).filter(name => name !== 'session.lock').sort())
+      .toEqual(['session.v1.jsonl', 'session.v2.jsonl'])
+  })
+
+  it('persists native V3 system appends after the protected head without converting them to user messages', async () => {
+    const ctx = await mount()
+    const session = Session.create(id)
+    const writer = await ctx.sessionPersistence.create(session.header)
+    try {
+      session.append('turn/start', { turn: 1 })
+      session.append('step/start', { turn: 1, step: 1 })
+      session.append('system/message', {
+        turn: 1, step: 1,
+        message: freezeMessage({ role: 'system', id: MessageId('head'), content: [{ type: 'text', text: 'head prompt' }], source: { kind: 'plugin', plugin: '@deepseek-ai/dsh-system-prompt' } }),
+      }, { surfaceOp: 'append' })
+      session.append('user/message', freezeMessage({ role: 'user', id: MessageId('question'), content: human.content, source: { kind: 'user' } }), { surfaceOp: 'append' })
+      session.append('system/message', {
+        turn: 1, step: 1,
+        message: freezeMessage({ role: 'system', id: MessageId('context'), content: [{ type: 'text', text: 'tail context' }], source: { kind: 'plugin', plugin: 'context-plugin' } }),
+      }, { surfaceOp: 'append' })
+      session.append('step/end', { turn: 1, step: 1 })
+      session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
+      await writer.append(session.snapshotEvents())
+      await writer.flush()
+    } finally {
+      await writer.close()
+    }
+    const reopened = await mount()
+    const reader = await reopened.sessionPersistence.open(id, 'read')
+    try {
+      expect((await reader.read()).events).toEqual(session.snapshotEvents())
+      const restored = await restore(reader)
+      expect(restored.surface.nodes).toEqual([2, 3, 4])
+      expect(messages(restored)).toEqual([prompt('head prompt'), human, prompt('tail context')])
+      expect(restored.deriveMessages().map(message => message.id)).toEqual(['head', 'question', 'context'])
+    } finally {
+      await reader.close()
+    }
+  })
+})