perf(ui-trajectory): reuse finalized stream projections

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

@@ -1,12 +1,14 @@
 import type {
   HistoryEntry, IApiClient, MuxFrame, RpcError, SessionId,
 } from '@deepseek-ai/dsh-client-connection/client'
+import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
 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 { Notifier } from '../sessions/notifier.ts'
+import { PartialAccumulator } from '../sessions/partial.ts'
 
 const HISTORY_PAGE_MESSAGES = 50
 
@@ -33,6 +35,9 @@ export class SessionHistorySource implements SessionHistoryFace {
     entries: readonly HistoryEntry[]
     value: SessionHistorySnapshot['inspection']
   } | null = null
+  private streamPublishToken: object | null = null
+  private streamBaseInspection: SessionHistorySnapshot['inspection'] | null = null
+  private streamPartial: PartialAccumulator | null = null
   private snapshotCache: SessionHistorySnapshot
   private readonly notifier = new Notifier(() => {
     this.snapshotCache = this.buildSnapshot()
@@ -125,7 +130,7 @@ export class SessionHistorySource implements SessionHistoryFace {
     if (this.state !== 'cold') {
       this.state = 'cold'
       this.error = null
-      this.notifier.markDirty()
+      this.publishDirtyNow()
     }
   }
 
@@ -143,7 +148,7 @@ export class SessionHistorySource implements SessionHistoryFace {
     this.hasMore = false
     this.state = 'cold'
     this.error = null
-    this.notifier.markDirty()
+    this.publishDirtyNow()
     void this.loadForConsumers()
   }
 
@@ -155,6 +160,9 @@ export class SessionHistorySource implements SessionHistoryFace {
     this.openPromise = null
     this.olderPromise = null
     this.liveBuffer = []
+    this.streamPublishToken = null
+    this.streamBaseInspection = null
+    this.streamPartial = null
   }
 
   private open(): Promise<void> {
@@ -188,7 +196,7 @@ export class SessionHistorySource implements SessionHistoryFace {
   private async doOpen(generation: number): Promise<void> {
     this.state = 'loading'
     this.error = null
-    this.notifier.markDirty()
+    this.publishDirtyNow()
     try {
       let { result } = await this.api.sessions.history({
         sessionId: this.sessionId,
@@ -222,7 +230,7 @@ export class SessionHistorySource implements SessionHistoryFace {
       /* v8 ignore next -- transportError always returns the error branch. */
       this.error = folded.ok ? null : folded.error
     } finally {
-      if (generation === this.generation) this.notifier.markDirty()
+      if (generation === this.generation) this.publishDirtyNow()
     }
   }
 
@@ -261,7 +269,7 @@ export class SessionHistorySource implements SessionHistoryFace {
     const settled = operation.finally(() => {
       if (this.olderPromise !== settled) return
       this.olderPromise = null
-      this.notifier.markDirty()
+      this.publishDirtyNow()
     })
     this.olderPromise = settled
     return settled
@@ -286,7 +294,7 @@ export class SessionHistorySource implements SessionHistoryFace {
     const buffered = this.liveBuffer
     this.liveBuffer = []
     for (const entry of buffered) this.appendLive(entry)
-    this.notifier.markDirty()
+    this.publishDirtyNow()
   }
 
   private acceptLive(entry: HistoryEntry): void {
@@ -301,8 +309,16 @@ export class SessionHistorySource implements SessionHistoryFace {
       void this.repairGap()
       return
     }
+    if (
+      entry.event.type === 'assistant/chunk'
+      && entry.event.data.chunk.type !== 'usage'
+    ) {
+      if (!this.appendIncrementalChunk(entry, entry.event)) return
+      this.publishStreamDirty()
+      return
+    }
     this.appendLive(entry)
-    this.notifier.markDirty()
+    this.publishDirtyNow()
   }
 
   private appendLive(entry: HistoryEntry): void {
@@ -311,6 +327,66 @@ export class SessionHistorySource implements SessionHistoryFace {
     this.entries = [...this.entries, entry]
   }
 
+  /** Append a chunk against the cached finalized projection; false means no visible publish. */
+  private appendIncrementalChunk(
+    entry: HistoryEntry,
+    event: SessionEvent<'assistant/chunk'>,
+  ): boolean {
+    const { turn, step, chunk } = event.data
+    if (!isVisibleAssistantChunk(chunk.type)) {
+      const inspection = this.currentInspection()
+      this.appendLive(entry)
+      this.inspectionCache = { entries: this.entries, value: inspection }
+      return false
+    }
+    const base = this.streamBaseInspection ?? this.currentInspection()
+    this.streamBaseInspection = base
+    if (
+      this.streamPartial === null
+      || this.streamPartial.turn !== turn
+      || this.streamPartial.step !== step
+    ) {
+      const current = base.partial
+      this.streamPartial = new PartialAccumulator(
+        turn,
+        step,
+        current?.turn === turn && current.step === step ? current.blocks : [],
+      )
+    }
+    this.streamPartial.push(chunk)
+    this.appendLive(entry)
+    this.inspectionCache = {
+      entries: this.entries,
+      value: { ...base, partial: this.streamPartial.toPartial() },
+    }
+    return true
+  }
+
+  /** Coalesce token-stream projection and rendering work to one publish per browser frame. */
+  private publishStreamDirty(): void {
+    if (this.streamPublishToken !== null) return
+    const token = {}
+    this.streamPublishToken = token
+    const publish = () => {
+      if (this.streamPublishToken !== token) return
+      this.streamPublishToken = null
+      this.notifier.markDirty()
+    }
+    if (typeof globalThis.requestAnimationFrame === 'function') {
+      globalThis.requestAnimationFrame(publish)
+    } else {
+      queueMicrotask(publish)
+    }
+  }
+
+  /** Publish structural changes immediately and invalidate an older scheduled stream publish. */
+  private publishDirtyNow(): void {
+    this.streamPublishToken = null
+    this.streamBaseInspection = null
+    this.streamPartial = null
+    this.notifier.markDirty()
+  }
+
   private async repairGap(): Promise<void> {
     if (this.stitching) return
     this.stitching = true
@@ -335,6 +411,16 @@ export class SessionHistorySource implements SessionHistoryFace {
   }
 
   private buildSnapshot(): SessionHistorySnapshot {
+    return {
+      state: this.state,
+      error: this.error,
+      hasMore: this.hasMore,
+      inspection: this.currentInspection(),
+    }
+  }
+
+  /** Inspection pinned to the source's current immutable entry array. */
+  private currentInspection(): SessionHistorySnapshot['inspection'] {
     if (this.inspectionCache?.entries !== this.entries) {
       const entries = this.entries
       this.inspectionCache = {
@@ -342,11 +428,14 @@ export class SessionHistorySource implements SessionHistoryFace {
         value: createHistoryInspection(() => entries),
       }
     }
-    return {
-      state: this.state,
-      error: this.error,
-      hasMore: this.hasMore,
-      inspection: this.inspectionCache.value,
-    }
+    return this.inspectionCache.value
   }
 }
+
+function isVisibleAssistantChunk(type: string): boolean {
+  return type === 'block-start'
+    || type === 'text-delta'
+    || type === 'reasoning-delta'
+    || type === 'tool-call-delta'
+    || type === 'block-end'
+}

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

@@ -13,8 +13,18 @@ export class PartialAccumulator {
   private changed = true
   private snapshot: PartialAssistant
 
-  constructor(readonly turn: number, readonly step: number) {
-    this.snapshot = { turn, step, blocks: [] }
+  /**
+   * @param turn - Owning agent turn.
+   * @param step - Owning model step.
+   * @param initialBlocks - Materialized prefix when accumulation begins after history replay.
+   */
+  constructor(
+    readonly turn: number,
+    readonly step: number,
+    initialBlocks: readonly AssistantBlock[] = [],
+  ) {
+    this.blocks = [...initialBlocks]
+    this.snapshot = { turn, step, blocks: initialBlocks }
   }
 
   /**

packages/client/runtime/tests/partial.spec.ts

@@ -41,6 +41,12 @@ describe('PartialAccumulator', () => {
     expect(acc.toPartial().blocks).toEqual([{ kind: 'reasoning', text: '思考' }])
   })
 
+  it('continues from a materialized history prefix', () => {
+    const acc = new PartialAccumulator(1, 0, [{ kind: 'text', text: '已有' }])
+    acc.push(chunk({ type: 'text-delta', index: 0, text: '增量' }))
+    expect(acc.toPartial().blocks).toEqual([{ kind: 'text', text: '已有增量' }])
+  })
+
   it('folds tool-call deltas: first id pins callId, late name overrides, argsRaw concatenates', () => {
     const acc = new PartialAccumulator(1, 0)
     acc.push(chunk({ type: 'tool-call-delta', index: 0, id: 'c1', argumentsDelta: '{"a"' }))

packages/client/runtime/tests/session-history-source.spec.ts

@@ -1,4 +1,4 @@
-import { describe, expect, it } from 'vitest'
+import { afterEach, describe, expect, it, vi } from 'vitest'
 import type { SessionEvent } from '@deepseek-ai/dsh-session/types'
 import type { SessionId } from '@deepseek-ai/dsh-client-connection/client'
 import { SessionHistorySource } from '../src/client/session-history/source.ts'
@@ -7,6 +7,10 @@ import { entries, ev, plainTurn } from './event-script.ts'
 
 const SID = 'history-s1' as SessionId
 
+afterEach(() => {
+  vi.unstubAllGlobals()
+})
+
 function histResponse(events: SessionEvent[], hasMore = false) {
   return Promise.resolve(ok({ events: entries(events) as never[], hasMore }))
 }
@@ -52,6 +56,71 @@ describe('SessionHistorySource', () => {
       .toEqual([1, 3, 6])
   })
 
+  it('publishes multiple assistant chunks once per browser frame', async () => {
+    const api = new FakeApiClient()
+    api.onHistory = () => histResponse(plainTurn(0, 0, '问', '答'))
+    const source = new SessionHistorySource(SID, api)
+    await source.loadAll()
+    const frames: FrameRequestCallback[] = []
+    vi.stubGlobal('requestAnimationFrame', (callback: FrameRequestCallback) => {
+      frames.push(callback)
+      return frames.length
+    })
+    let notifications = 0
+    const unsubscribe = source.subscribe(() => { notifications++ })
+    const before = source.getSnapshot().inspection
+    const finalizedNodes = before.eventNodes
+    const requests = before.requests
+    const contexts = before.contexts
+
+    for (const event of [
+      ev.chunkStart(6, 1),
+      ev.chunkText(7, 1, 'stream '),
+      ev.chunkText(8, 1, 'content'),
+    ]) {
+      source.handleMuxFrame({
+        type: 'session/event',
+        sessionId: SID,
+        event,
+      })
+    }
+
+    expect(frames).toHaveLength(1)
+    expect(notifications).toBe(0)
+    frames[0]?.(0)
+    await Promise.resolve()
+
+    expect(notifications).toBe(1)
+    const streamed = source.getSnapshot().inspection
+    expect(streamed.eventNodes).toBe(finalizedNodes)
+    expect(streamed.requests).toBe(requests)
+    expect(streamed.contexts).toBe(contexts)
+    expect(streamed.partial?.blocks).toEqual([
+      { kind: 'text', text: 'stream content' },
+    ])
+
+    source.handleMuxFrame({
+      type: 'session/event',
+      sessionId: SID,
+      event: ev.chunkText(9, 1, ' then final'),
+    })
+    source.handleMuxFrame({
+      type: 'session/event',
+      sessionId: SID,
+      event: ev.assistant(10, 1, 'stream content then final'),
+    })
+    await Promise.resolve()
+
+    expect(notifications).toBe(2)
+    const finalized = source.getSnapshot().inspection
+    expect(finalized.eventNodes).not.toBe(finalizedNodes)
+    expect(finalized.partial).toBeNull()
+    frames[1]?.(0)
+    await Promise.resolve()
+    expect(notifications).toBe(2)
+    unsubscribe()
+  })
+
   it('stops loading when an older page fails to advance', async () => {
     const api = new FakeApiClient()
     api.onHistory = payload => payload.beforeSeq === undefined

packages/client/ui-trajectory/src/client/layout.ts

@@ -12,6 +12,7 @@ import type {
   RequestView,
   ToolResultNode,
 } from '@deepseek-ai/dsh-client-runtime/client'
+import { extractMarkdownPlainText } from '@deepseek-ai/dsh-client-ui-primitives'
 import type {
   TrajectoryCellProps,
   TrajectorySourceBlock,
@@ -925,13 +926,7 @@ function summarizeText(text: string): string {
  */
 export function trajectoryPreviewText(text: string): string {
   const source = text.slice(0, PREVIEW_SOURCE_CHARACTERS)
-  const compact = source
-    .replace(/!\[([^\]]*)\]\([^)]*\)/g, '$1')
-    .replace(/\[([^\]]+)\]\([^)]*\)/g, '$1')
-    .replace(/(^|\s)(?:#{1,6}|[-+*>])\s+/g, '$1')
-    .replace(/[*_~`]+/g, '')
-    .replace(/\s+/g, ' ')
-    .trim()
+  const compact = extractMarkdownPlainText(source).replace(/\s+/g, ' ').trim()
   const preview = compact.slice(0, PREVIEW_OUTPUT_CHARACTERS).trimEnd()
   return source.length < text.length || preview.length < compact.length
     ? `${preview}…`

packages/client/ui-trajectory/tests/layout.spec.tsx

@@ -177,7 +177,7 @@ describe('deriveTrajectoryLayout', () => {
   })
 
   it('bounds a long Markdown-like thinking preview while retaining its full detail', () => {
-    const thinking = `# Investigation\n\n**finding** ${'- repeated detail '.repeat(1_000)}`
+    const thinking = `# Investigation\n\n**NAVIGATION_OK file_path** ${'- repeated detail '.repeat(1_000)}`
     const nodes = [{
       kind: 'assistant', seq: 1, time: 5_000, turn: 1, step: 0,
       blocks: [{ kind: 'reasoning', text: thinking }],
@@ -189,7 +189,7 @@ describe('deriveTrajectoryLayout', () => {
     const message = turns[0]?.groups.flatMap(group => group.cells)
       .find(cell => cell.kind === 'message')
 
-    expect(message?.text.startsWith('Investigation finding')).toBe(true)
+    expect(message?.text.startsWith('Investigation NAVIGATION_OK file_path')).toBe(true)
     expect(message?.text.endsWith('…')).toBe(true)
     expect(message?.text.length).toBeLessThanOrEqual(513)
     expect(message?.thinkingDetail).toBe(thinking)