fix(trajectory): bound work during long streams

packages/client/runtime/src/client/session-history/history-fold.ts

@@ -78,28 +78,61 @@ function isTokenDelta(chunk: SessionEvent<'assistant/chunk'>['data']['chunk']):
 
 function foldContexts(
   events: readonly SessionEvent[],
-  baseSeq: number,
 ): readonly FoldedContext[] {
   const replay: SessionEvent[] = []
-  const surface = new SurfaceManager(replay, baseSeq)
+  const originalSeqs: number[] = []
+  const rebasedSeqByOriginal = new Map<number, number>()
+  const surface = new SurfaceManager(replay)
   const contexts: FoldedContext[] = []
   let generation = 0
   let originSeq: number | undefined
+  const originalNodes = () => surface.nodes.map((seq) => {
+    const original = originalSeqs[seq]
+    if (original === undefined) throw new Error(`rebased surface seq ${seq} has no origin`)
+    return original
+  })
   for (const event of events) {
-    if (isSurfaceEvent(event) && event.surfaceOp !== 'append') {
+    if (!isSurfaceEvent(event)) continue
+    if (event.surfaceOp !== 'append') {
       contexts.push({
         generation,
-        nodes: [...surface.nodes],
+        nodes: originalNodes(),
         ...(originSeq === undefined ? {} : { originSeq }),
       })
       generation++
       originSeq = event.seq
     }
-    replay.push(event)
+    const rebasedSeq = replay.length
+    const {
+      sourceEventSeqs: rawSources,
+      ...eventWithoutSources
+    } = event as SessionEvent & { sourceEventSeqs?: readonly number[] }
+    const mappedSourceEventSeqs = rawSources?.flatMap((seq) => {
+      const rebased = rebasedSeqByOriginal.get(seq)
+      return rebased === undefined ? [] : [rebased]
+    })
+    const sourceEventSeqs = mappedSourceEventSeqs?.length === 0
+      ? undefined
+      : mappedSourceEventSeqs
+    const surfaceOp = event.surfaceOp === 'append'
+      ? event.surfaceOp
+      : {
+        ...event.surfaceOp,
+        start: rebasedSeqByOriginal.get(event.surfaceOp.start) ?? event.surfaceOp.start,
+        end: rebasedSeqByOriginal.get(event.surfaceOp.end) ?? event.surfaceOp.end,
+      }
+    originalSeqs.push(event.seq)
+    rebasedSeqByOriginal.set(event.seq, rebasedSeq)
+    replay.push({
+      ...eventWithoutSources,
+      seq: rebasedSeq,
+      surfaceOp,
+      ...(sourceEventSeqs === undefined ? {} : { sourceEventSeqs }),
+    } as SessionEvent)
   }
   contexts.push({
     generation,
-    nodes: [...surface.nodes],
+    nodes: originalNodes(),
     ...(originSeq === undefined ? {} : { originSeq }),
   })
   return contexts
@@ -327,6 +360,7 @@ export function projectConversationHistory(
 ): ConversationHistoryProjection {
   const events = entries.map(entry => entry.event)
   const baseSeq = events[0]?.seq ?? 0
+  const eventsBySeq = new Map(events.map(event => [event.seq, event]))
   const callIndex = new Map<string, CallIndexEntry>()
   const resultViews = new Map<number, ToolResultView>()
   const assistantSteps = new Map<string, AssistantStepMetadata>()
@@ -396,7 +430,7 @@ export function projectConversationHistory(
   const materialize = (seq: number): ConversationNode | undefined => {
     const cached = nodeCache.get(seq)
     if (cached !== undefined) return cached
-    const event = events[seq - baseSeq]
+    const event = eventsBySeq.get(seq)
     if (event === undefined || !isSurfaceEligibleType(event.type)) return
     const node = materializeNode(
       event,
@@ -422,7 +456,7 @@ export function projectConversationHistory(
     }]
   } else {
     try {
-      contexts = foldContexts(events, baseSeq).map((context): ConversationContext => {
+      contexts = foldContexts(events).map((context): ConversationContext => {
         const nodes = context.nodes.flatMap((seq) => {
           const node = materialize(seq)
           return node === undefined ? [] : [node]
@@ -435,7 +469,7 @@ export function projectConversationHistory(
             nodes,
           }
         }
-        const originEvent = events[context.originSeq - baseSeq]
+        const originEvent = eventsBySeq.get(context.originSeq)
         return {
           id: context.generation,
           parentId: context.generation - 1,

packages/client/runtime/src/client/session-history/source.ts

@@ -6,7 +6,9 @@ import { transportError } from '@deepseek-ai/dsh-host-apiproxy/api'
 import type {
   SessionHistoryFace, SessionHistorySnapshot,
 } from '../contract/session-history.ts'
-import { createHistoryInspection } from '../sessions/history.ts'
+import {
+  compactHistoryInspectionEntries, createHistoryInspection,
+} from '../sessions/history.ts'
 import { Notifier } from '../sessions/notifier.ts'
 import { PartialAccumulator } from '../sessions/partial.ts'
 
@@ -18,7 +20,8 @@ function isAborted(signal: AbortSignal | undefined): boolean {
 
 /** Independent raw-history owner used only by inspection consumers. */
 export class SessionHistorySource implements SessionHistoryFace {
-  private entries: readonly HistoryEntry[] = []
+  private entries: HistoryEntry[] = []
+  private inspectionEntries: readonly HistoryEntry[] = []
   private baseSeq = 0
   private hasMore = false
   private state: SessionHistorySnapshot['state'] = 'cold'
@@ -135,6 +138,7 @@ export class SessionHistorySource implements SessionHistoryFace {
     this.liveBuffer = []
     this.subscribedLastSeq = null
     this.entries = []
+    this.inspectionEntries = []
     this.baseSeq = 0
     this.hasMore = false
     this.state = 'cold'
@@ -250,6 +254,7 @@ export class SessionHistorySource implements SessionHistoryFace {
           return
         }
         this.entries = [...older, ...this.entries]
+        this.inspectionEntries = compactHistoryInspectionEntries([...this.entries])
         this.baseSeq = older[0]?.event.seq ?? this.baseSeq
         this.hasMore = result.value.hasMore
       } catch (error) {
@@ -281,6 +286,7 @@ export class SessionHistorySource implements SessionHistoryFace {
       this.entries = [...prefix, ...tail]
     }
     this.baseSeq = this.entries[0]?.event.seq ?? 0
+    this.inspectionEntries = compactHistoryInspectionEntries([...this.entries])
     const buffered = this.liveBuffer
     this.liveBuffer = []
     for (const entry of buffered) this.appendLive(entry)
@@ -314,7 +320,11 @@ export class SessionHistorySource implements SessionHistoryFace {
   private appendLive(entry: HistoryEntry): void {
     const tailSeq = this.tailSeq()
     if (tailSeq !== null && entry.event.seq <= tailSeq) return
-    this.entries = [...this.entries, entry]
+    this.entries.push(entry)
+    this.inspectionEntries = [...this.inspectionEntries, entry]
+    if (entry.event.type === 'assistant/message') {
+      this.inspectionEntries = compactHistoryInspectionEntries(this.inspectionEntries)
+    }
   }
 
   /** Append a chunk against the cached finalized projection; false means no visible publish. */
@@ -326,7 +336,7 @@ export class SessionHistorySource implements SessionHistoryFace {
     if (!isVisibleAssistantChunk(chunk.type)) {
       const inspection = this.currentInspection()
       this.appendLive(entry)
-      this.inspectionCache = { entries: this.entries, value: inspection }
+      this.inspectionCache = { entries: this.inspectionEntries, value: inspection }
       return false
     }
     const base = this.currentInspection()
@@ -345,7 +355,7 @@ export class SessionHistorySource implements SessionHistoryFace {
     this.streamPartial.push(chunk)
     this.appendLive(entry)
     this.inspectionCache = {
-      entries: this.entries,
+      entries: this.inspectionEntries,
       value: { ...base, partial: this.streamPartial.toPartial() },
     }
     return true
@@ -410,8 +420,8 @@ export class SessionHistorySource implements SessionHistoryFace {
 
   /** Inspection pinned to the source's current immutable entry array. */
   private currentInspection(): SessionHistorySnapshot['inspection'] {
-    if (this.inspectionCache?.entries !== this.entries) {
-      const entries = this.entries
+    if (this.inspectionCache?.entries !== this.inspectionEntries) {
+      const entries = this.inspectionEntries
       this.inspectionCache = {
         entries,
         value: createHistoryInspection(() => entries),

packages/client/runtime/src/client/sessions/history.ts

@@ -7,6 +7,24 @@ import type { ConversationContext } from './conversation-context.ts'
 import { projectConversationHistory } from '../session-history/history-fold.ts'
 import { inspectRequests, type RequestView } from './request-inspection.ts'
 
+function assistantStepKey(turn: number, step: number): string {
+  return `${turn}\u0000${step}`
+}
+
+function isFirstTokenCandidate(entry: HistoryEntry): boolean {
+  const event = entry.event
+  if (event.type !== 'assistant/chunk') return false
+  switch (event.data.chunk.type) {
+    case 'text-delta':
+    case 'reasoning-delta':
+      return event.data.chunk.text !== ''
+    case 'tool-call-delta':
+      return event.data.chunk.argumentsDelta !== '' || event.data.chunk.name !== undefined
+    default:
+      return false
+  }
+}
+
 /** Lazily derived inspection data for one immutable session-history window. */
 export interface SessionHistoryInspection {
   eventNodes: readonly ConversationNode[]
@@ -19,6 +37,47 @@ export interface SessionHistoryInspection {
   codeDispatches: ReadonlyMap<string, readonly CodeSubCall[]>
 }
 
+/**
+ * Remove completed-step token payloads that no inspection projection reads.
+ * The first visible token preserves timing, usage chunks preserve accounting,
+ * and unfinished steps retain every chunk for live or interrupted content.
+ * @param entries - Contiguous raw history entries in sequence order.
+ * @returns A projection-equivalent, usually much smaller entry ledger.
+ */
+export function compactHistoryInspectionEntries(
+  entries: readonly HistoryEntry[],
+): readonly HistoryEntry[] {
+  const completedSteps = new Set<string>()
+  for (const { event } of entries) {
+    if (event.type === 'assistant/message') {
+      completedSteps.add(assistantStepKey(event.data.turn, event.data.step))
+    }
+  }
+
+  const firstTokenSteps = new Set<string>()
+  const compacted: HistoryEntry[] = []
+  let changed = false
+  for (const entry of entries) {
+    const event = entry.event
+    if (event.type !== 'assistant/chunk') {
+      compacted.push(entry)
+      continue
+    }
+    const key = assistantStepKey(event.data.turn, event.data.step)
+    if (!completedSteps.has(key) || event.data.chunk.type === 'usage') {
+      compacted.push(entry)
+      continue
+    }
+    if (isFirstTokenCandidate(entry) && !firstTokenSteps.has(key)) {
+      firstTokenSteps.add(key)
+      compacted.push(entry)
+    } else {
+      changed = true
+    }
+  }
+  return changed ? compacted : entries
+}
+
 /**
  * Create a lazy inspection projection over an immutable history window.
  * Conversation consumers retain the cheap wrapper; only Trajectory snapshots

packages/client/runtime/tests/history-fold.spec.ts

@@ -2,6 +2,8 @@ import { createMessage } from '@deepseek-ai/dsh-llm'
 import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
 import { describe, expect, it } from 'vitest'
 import { projectConversationHistory } from '../src/client/session-history/history-fold.ts'
+import { compactHistoryInspectionEntries } from '../src/client/sessions/history.ts'
+import { inspectRequests } from '../src/client/sessions/request-inspection.ts'
 import { ev } from './event-script.ts'
 
 const at = (seq: number, event: Record<string, unknown>): SessionEvent =>
@@ -122,4 +124,35 @@ describe('projectConversationHistory', () => {
       requestConfig: { provider: 'fake', model: 'first' },
     })
   })
+
+  it('drops completed token payloads without changing inspection projections', () => {
+    const events = [
+      ev.user(0, 'before'),
+      ev.stepStart(1, 1, 0),
+      ev.chunkStart(2, 1),
+      ev.chunkText(3, 1, ''),
+      ev.chunkText(4, 1, 'first'),
+      ev.chunkText(5, 1, ' discarded'),
+      at(6, { type: 'assistant/chunk', data: {
+        turn: 1,
+        step: 0,
+        chunk: { type: 'usage', usage: { inputTokens: 4, outputTokens: 2 } },
+      } }),
+      ev.assistant(7, 1, 'first discarded'),
+      ev.compactSummary(8, 'summary', 0, 7),
+      ev.compactCheckpoint(9, 8, 0, 7),
+      ev.stepStart(10, 2, 0),
+      ev.chunkStart(11, 2),
+      ev.chunkText(12, 2, 'interrupted'),
+      ev.turnEnd(13, 2, 'aborted'),
+    ]
+    const raw = events.map(event => ({ event }))
+    const compacted = compactHistoryInspectionEntries(raw)
+
+    expect(compacted.map(entry => entry.event.seq)).toEqual([
+      0, 1, 4, 6, 7, 8, 9, 10, 11, 12, 13,
+    ])
+    expect(projectConversationHistory(compacted)).toEqual(projectConversationHistory(raw))
+    expect(inspectRequests(compacted)).toEqual(inspectRequests(raw))
+  })
 })