fix(history): preserve per-delta replay

packages/api/session-controller/src/client/sessions/history-records.ts

@@ -1,99 +1,18 @@
-/** Compact client folding for packed Assistant delta runs in history responses. */
+/** Lossless client decoding for packed Assistant delta runs in history responses. */
 
-import type { ChunkRow } from '@deepseek-ai/dsh-session/chunk-rows'
-import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
+import { decodeStorageRecord } from '@deepseek-ai/dsh-session/chunk-rows'
 import type {
   SessionEventEntry,
   SessionHistoryRecord,
 } from '../../types.ts'
 
-/** Resolve one packed member's original timestamp from the row's delta gaps. */
-function memberTime(row: ChunkRow, index: number): number {
-  let time = row.time0
-  for (let cursor = 0; cursor < index; cursor++) time += row.data.dt[cursor] as number
-  return time
-}
-
-/** Build one coalesced text or reasoning event from a contiguous member slice. */
-function textEvent(
-  row: Extract<ChunkRow, { type: 'text-chunks' | 'reasoning-chunks' }>,
-  start: number,
-  text: string,
-): SessionEvent<'assistant/chunk'> {
-  return {
-    type: 'assistant/chunk',
-    seq: row.seq0 + start,
-    time: memberTime(row, start),
-    data: {
-      turn: row.data.turn,
-      step: row.data.step,
-      chunk: row.type === 'text-chunks'
-        ? { type: 'text-delta', index: row.data.index, text }
-        : { type: 'reasoning-delta', index: row.data.index, text },
-    },
-  }
-}
-
 /**
- * Coalesce one packed run into the smallest event set that preserves the
- * conversation fold's accumulated content, first-token time, and first
- * non-whitespace visibility boundary. Exact token boundaries remain available
- * in the wire row to consumers that explicitly decode it.
- * @param row - one validated packed history record.
- * @returns At most two Assistant chunk events for the ordinary UI fold.
- */
-export function coalesceHistoryChunkRun(row: ChunkRow): SessionEvent<'assistant/chunk'>[] {
-  if (row.type === 'tool-call-chunks') {
-    const firstToken = row.data.name === undefined
-      ? row.data.args.findIndex(fragment => fragment !== '')
-      : 0
-    const start = firstToken < 0 ? 0 : firstToken
-    return [{
-      type: 'assistant/chunk',
-      seq: row.seq0 + start,
-      time: memberTime(row, start),
-      data: {
-        turn: row.data.turn,
-        step: row.data.step,
-        chunk: {
-          type: 'tool-call-delta',
-          index: row.data.index,
-          id: row.data.id,
-          ...row.data.name === undefined ? {} : { name: row.data.name },
-          argumentsDelta: row.data.args.join(''),
-        },
-      },
-    }]
-  }
-
-  const texts = row.data.texts
-  const firstToken = texts.findIndex(text => text !== '')
-  const tokenStart = firstToken < 0 ? 0 : firstToken
-  let visibleStart = -1
-  let accumulated = ''
-  for (let index = 0; index < texts.length; index++) {
-    accumulated += texts[index] as string
-    if (accumulated.trim() !== '') {
-      visibleStart = index
-      break
-    }
-  }
-  if (visibleStart > tokenStart) {
-    return [
-      textEvent(row, tokenStart, texts.slice(0, visibleStart).join('')),
-      textEvent(row, visibleStart, texts.slice(visibleStart).join('')),
-    ]
-  }
-  return [textEvent(row, tokenStart, texts.join(''))]
-}
-
-/**
- * Convert history wire records into compact event inputs for the ordinary UI.
+ * Convert history wire records into exact event inputs for Conversation Definitions.
  * @param records - validated lossless history transport records.
- * @returns Ordinary entries unchanged and packed runs coalesced for folding.
+ * @returns Ordinary entries unchanged and packed runs expanded member-for-member.
  */
 export function historyEntries(records: readonly SessionHistoryRecord[]): SessionEventEntry[] {
   return records.flatMap(record => 'event' in record
     ? [record]
-    : coalesceHistoryChunkRun(record.chunks).map(event => ({ event })))
+    : decodeStorageRecord(record.chunks).map(event => ({ event })))
 }

packages/api/session-controller/tests/history-records.client.spec.ts

@@ -1,99 +1,15 @@
-/** Packed history record folding without token-by-token browser expansion. */
+/** Packed history records decode to the exact Session event stream. */
 
 import { describe, expect, it } from 'vitest'
 import { CallId } from '@deepseek-ai/dsh-llm/brand'
-import type { ChunkRow } from '@deepseek-ai/dsh-session/chunk-rows'
 import type { SessionHistoryRecord } from '../src/types.ts'
-import { coalesceHistoryChunkRun, historyEntries } from '../src/client/sessions/history-records.ts'
-
-describe('coalesceHistoryChunkRun', () => {
-  it('preserves first-token and first-visible boundaries with at most two text events', () => {
-    const row: ChunkRow = {
-      type: 'text-chunks',
-      seq0: 10,
-      time0: 100,
-      data: {
-        turn: 2,
-        step: 3,
-        index: 0,
-        dt: [1, 2, 3, 4],
-        texts: ['', ' ', '', 'hello', ' world'],
-      },
-    }
-
-    const events = coalesceHistoryChunkRun(row)
-    expect(events).toHaveLength(2)
-    expect(events.map(event => ({ seq: event.seq, time: event.time, text: event.data.chunk.type === 'text-delta' ? event.data.chunk.text : '' })))
-      .toEqual([
-        { seq: 11, time: 101, text: ' ' },
-        { seq: 13, time: 106, text: 'hello world' },
-      ])
-    expect(events[0]?.time).toBe(101)
-    expect(events.find(event => event.data.chunk.type === 'text-delta' && event.data.chunk.text.trim() !== '')?.time)
-      .toBe(106)
-  })
-
-  it('joins visible reasoning members into one event at the first non-empty member', () => {
-    const row: ChunkRow = {
-      type: 'reasoning-chunks',
-      seq0: 4,
-      time0: 50,
-      data: { turn: 1, step: 1, index: 2, dt: [5, 7], texts: ['', 'a', 'b'] },
-    }
-    const [event] = coalesceHistoryChunkRun(row)
-    expect(event).toMatchObject({
-      seq: 5,
-      time: 55,
-      data: { chunk: { type: 'reasoning-delta', index: 2, text: 'ab' } },
-    })
-  })
-
-  it('joins tool arguments while retaining name presence and first-token time', () => {
-    const named: ChunkRow = {
-      type: 'tool-call-chunks',
-      seq0: 20,
-      time0: 200,
-      data: {
-        turn: 2,
-        step: 4,
-        index: 1,
-        id: CallId('call-1'),
-        name: 'write',
-        dt: [2, 3],
-        args: ['', '{"x":', '1}'],
-      },
-    }
-    expect(coalesceHistoryChunkRun(named)).toMatchObject([{
-      seq: 20,
-      time: 200,
-      data: { chunk: { type: 'tool-call-delta', name: 'write', argumentsDelta: '{"x":1}' } },
-    }])
-
-    const unnamed: ChunkRow = {
-      type: 'tool-call-chunks',
-      seq0: 20,
-      time0: 200,
-      data: {
-        turn: 2,
-        step: 4,
-        index: 1,
-        id: CallId('call-1'),
-        dt: [2, 3],
-        args: ['', '', 'x'],
-      },
-    }
-    const [event] = coalesceHistoryChunkRun(unnamed)
-    expect(event).toMatchObject({ seq: 22, time: 205, data: { chunk: { argumentsDelta: 'x' } } })
-    expect(Object.hasOwn(event?.data.chunk ?? {}, 'name')).toBe(false)
-  })
-})
+import { historyEntries } from '../src/client/sessions/history-records.ts'
 
 describe('historyEntries', () => {
-  it('keeps ordinary entries and views while folding a packed run without expansion', () => {
+  it('keeps ordinary entries and expands every packed text member with its exact boundary', () => {
     const ordinary = {
       event: { type: 'turn/start', seq: 0, time: 1, data: { turn: 1 } },
-      view: { for: 'call', view: { card: 'generic' } },
-    } as unknown as SessionHistoryRecord
+    } as SessionHistoryRecord
     const packed: SessionHistoryRecord = {
       chunks: {
         type: 'text-chunks',
@@ -102,9 +18,48 @@ describe('historyEntries', () => {
         data: { turn: 1, step: 1, index: 0, dt: [1, 1, 1], texts: ['a', 'b', 'c', 'd'] },
       },
     }
+
     const entries = historyEntries([ordinary, packed])
-    expect(entries).toHaveLength(2)
+
+    expect(entries).toHaveLength(5)
     expect(entries[0]).toBe(ordinary)
-    expect(entries[1]?.event).toMatchObject({ seq: 1, data: { chunk: { text: 'abcd' } } })
+    expect(entries.slice(1).map(entry => ({
+      seq: entry.event.seq,
+      time: entry.event.time,
+      chunk: entry.event.type === 'assistant/chunk' ? entry.event.data.chunk : undefined,
+    }))).toEqual([
+      { seq: 1, time: 2, chunk: { type: 'text-delta', index: 0, text: 'a' } },
+      { seq: 2, time: 3, chunk: { type: 'text-delta', index: 0, text: 'b' } },
+      { seq: 3, time: 4, chunk: { type: 'text-delta', index: 0, text: 'c' } },
+      { seq: 4, time: 5, chunk: { type: 'text-delta', index: 0, text: 'd' } },
+    ])
+  })
+
+  it('preserves every tool-call fragment and optional-name presence', () => {
+    const packed: SessionHistoryRecord = {
+      chunks: {
+        type: 'tool-call-chunks',
+        seq0: 20,
+        time0: 200,
+        data: {
+          turn: 2,
+          step: 4,
+          index: 1,
+          id: CallId('call-1'),
+          dt: [2, 3],
+          args: ['', '{"x":', '1}'],
+        },
+      },
+    }
+
+    const events = historyEntries([packed]).map(entry => entry.event)
+
+    expect(events).toMatchObject([
+      { seq: 20, time: 200, data: { chunk: { argumentsDelta: '' } } },
+      { seq: 21, time: 202, data: { chunk: { argumentsDelta: '{"x":' } } },
+      { seq: 22, time: 205, data: { chunk: { argumentsDelta: '1}' } } },
+    ])
+    expect(events.every(event => event.type === 'assistant/chunk'
+      && !Object.hasOwn(event.data.chunk, 'name'))).toBe(true)
   })
 })

packages/api/session-controller/tests/history-transport.perf.client.ts

@@ -1,4 +1,4 @@
-/** Opt-in synthetic benchmark for packed session-history transport and folding. */
+/** Opt-in synthetic benchmark for packed session-history transport and exact replay. */
 
 import { createHash } from 'node:crypto'
 import { performance } from 'node:perf_hooks'
@@ -40,6 +40,8 @@ interface HeapPeaks<T> {
 
 interface FoldState {
   readonly blocks: readonly string[]
+  readonly deltaCount: number
+  readonly lastDeltaSeq?: number
   readonly firstTokenTime?: number
   readonly firstVisibleSeq?: number
   readonly firstVisibleTime?: number
@@ -176,7 +178,7 @@ function foldDefinition(kind: string, target: string): ConversationNodeDefinitio
       }
       return null
     },
-    start: () => ({ blocks: [] }),
+    start: () => ({ blocks: [], deltaCount: 0 }),
     update: (context, match) => {
       if (match.event.type !== 'assistant/chunk' || match.event.data.chunk.type !== 'reasoning-delta') {
         return context.state
@@ -188,6 +190,8 @@ function foldDefinition(kind: string, target: string): ConversationNodeDefinitio
       return {
         ...context.state,
         blocks,
+        deltaCount: context.state.deltaCount + 1,
+        lastDeltaSeq: match.event.seq,
         ...context.state.firstTokenTime === undefined ? { firstTokenTime: match.event.time } : {},
         ...visible && context.state.firstVisibleSeq === undefined
           ? { firstVisibleSeq: match.event.seq, firstVisibleTime: match.event.time }
@@ -242,7 +246,7 @@ function digest(value: unknown): string {
   return createHash('sha256').update(JSON.stringify(value)).digest('hex')
 }
 
-it('reports packed history transport and compact fold costs', () => {
+it('reports packed history transport and exact replay costs', () => {
   const fixture = timed(buildEvents)
 
   assemble(conversationInputs(fixture.value.slice(0, 1_000).map(event => ({ event }))))
@@ -332,7 +336,7 @@ it('reports packed history transport and compact fold costs', () => {
   expect(fixture.value.filter(event => event.type !== 'assistant/chunk')).toHaveLength(ORDINARY_EVENTS)
   expect(packedRows).toHaveLength(DELTA_RUNS)
   expect(packed.value).toHaveLength(696)
-  expect(packedPreparation.value).toHaveLength(696)
+  expect(packedPreparation.value).toHaveLength(LOGICAL_EVENTS)
   expect(digest(packedFold.value)).toBe(digest(rawFold.value))
   expect(packedClientHeap.value).toBe(rawClientHeap.value)
   expect(rawHostHeap.value).toBe(rawBytes)
@@ -351,7 +355,7 @@ it('reports packed history transport and compact fold costs', () => {
       deltaEvents: DELTA_EVENTS,
       deltaRuns: packedRows.length,
       packedRecords: packed.value.length,
-      compactFoldInputs: packedPreparation.value.length,
+      decodedEvents: packedPreparation.value.length,
     },
     bytes: {
       rawJson: rawBytes,
@@ -406,3 +410,44 @@ it('reports packed history transport and compact fold costs', () => {
     },
   })}\n`)
 }, 600_000)
+
+it('reports exact decoding cost for long whitespace-prefix runs', () => {
+  historyEntries([{
+    chunks: {
+      type: 'reasoning-chunks',
+      seq0: 0,
+      time0: TIME_ZERO,
+      data: { turn: 1, step: 1, index: 0, dt: [], texts: ['x'] },
+    },
+  }])
+  const results = [10_000, 20_000, 40_000].map((members) => {
+    const record: HistoryRecord = {
+      chunks: {
+        type: 'reasoning-chunks',
+        seq0: 0,
+        time0: TIME_ZERO,
+        data: {
+          turn: 1,
+          step: 1,
+          index: 0,
+          dt: Array.from({ length: members - 1 }, () => 1),
+          texts: Array.from({ length: members }, (_, index) => index === members - 1 ? 'x' : ' '),
+        },
+      },
+    }
+    const decoded = historyEntries([record])
+    const samplesMs = Array.from({ length: 5 }, () => timed(() => historyEntries([record])).ms)
+    expect(decoded).toHaveLength(members)
+    expect(decoded.at(-1)?.event).toMatchObject({
+      seq: members - 1,
+      time: TIME_ZERO + members - 1,
+      data: { chunk: { type: 'reasoning-delta', text: 'x' } },
+    })
+    return {
+      members,
+      medianMs: rounded(median(samplesMs)),
+      samplesMs: samplesMs.map(rounded),
+    }
+  })
+  process.stdout.write(`HISTORY_WHITESPACE_PREFIX_PERF_RESULT ${JSON.stringify(results)}\n`)
+})