diff --git a/packages/client/runtime/src/client/session-history/source.ts b/packages/client/runtime/src/client/session-history/source.ts index 8de11b9da9..4f3e86be6a 100644 --- a/packages/client/runtime/src/client/session-history/source.ts +++ b/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 { @@ -188,7 +196,7 @@ export class SessionHistorySource implements SessionHistoryFace { private async doOpen(generation: number): Promise { 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 { 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' +} diff --git a/packages/client/runtime/src/client/sessions/partial.ts b/packages/client/runtime/src/client/sessions/partial.ts index 189febf253..242232f3c1 100644 --- a/packages/client/runtime/src/client/sessions/partial.ts +++ b/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 } } /** diff --git a/packages/client/runtime/tests/partial.spec.ts b/packages/client/runtime/tests/partial.spec.ts index 25681ea4cc..c1df190c33 100644 --- a/packages/client/runtime/tests/partial.spec.ts +++ b/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"' })) diff --git a/packages/client/runtime/tests/session-history-source.spec.ts b/packages/client/runtime/tests/session-history-source.spec.ts index 1375338e90..bcf25cd933 100644 --- a/packages/client/runtime/tests/session-history-source.spec.ts +++ b/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 diff --git a/packages/client/ui-trajectory/src/client/layout.ts b/packages/client/ui-trajectory/src/client/layout.ts index e180487630..0745a0c8df 100644 --- a/packages/client/ui-trajectory/src/client/layout.ts +++ b/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}…` diff --git a/packages/client/ui-trajectory/tests/layout.spec.tsx b/packages/client/ui-trajectory/tests/layout.spec.tsx index b90e6bf38a..b2271f9e9f 100644 --- a/packages/client/ui-trajectory/tests/layout.spec.tsx +++ b/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)