diff --git a/packages/api/session-controller/src/client/contract/events.ts b/packages/api/session-controller/src/client/contract/events.ts index f71c4be0b0..2f8bc48f82 100644 --- a/packages/api/session-controller/src/client/contract/events.ts +++ b/packages/api/session-controller/src/client/contract/events.ts @@ -2,6 +2,66 @@ import { notifySubscribers, type ObservableSnapshot } from '@deepseek-ai/dsh-client-store' import type { SessionEventEntry } from '../../types.ts' +interface EventWindowLeaf { + readonly kind: 'leaf' + readonly entries: readonly SessionEventEntry[] + readonly length: number +} + +interface EventWindowConcat { + readonly kind: 'concat' + readonly left: EventWindowNode + readonly right: EventWindowNode + readonly length: number +} + +type EventWindowNode = EventWindowLeaf | EventWindowConcat + +function leaf(entries: readonly SessionEventEntry[]): EventWindowLeaf { + return { kind: 'leaf', entries, length: entries.length } +} + +function concat(left: EventWindowNode, right: EventWindowNode): EventWindowConcat { + return { kind: 'concat', left, right, length: left.length + right.length } +} + +function materialize(node: EventWindowNode): readonly SessionEventEntry[] { + if (node.kind === 'leaf') return node.entries + const entries = new Array(node.length) + const pending: EventWindowNode[] = [node] + let index = 0 + while (pending.length > 0) { + const current = pending.pop() as EventWindowNode + if (current.kind === 'concat') { + pending.push(current.right, current.left) + continue + } + for (const entry of current.entries) { + entries[index] = entry + index += 1 + } + } + return entries +} + +function windowSnapshot( + node: EventWindowNode, + hasMore: boolean, + revision: number, + change: SessionEventChange, +): SessionEventWindow { + let entries: readonly SessionEventEntry[] | undefined + return { + get entries() { + entries ??= materialize(node) + return entries + }, + hasMore, + revision, + change, + } +} + /** Exact delta that produced the latest event-window revision. */ export type SessionEventChange = | { readonly kind: 'replace'; readonly entries: readonly SessionEventEntry[] } @@ -22,12 +82,13 @@ export type SessionEventSource = ObservableSnapshot /** Session-owned event feed; every accepted window mutation publishes synchronously. */ export class MutableSessionEventSource implements SessionEventSource { private readonly listeners = new Set<() => void>() - private snapshot: SessionEventWindow = { - entries: [], - hasMore: false, - revision: 0, - change: { kind: 'replace', entries: [] }, - } + private window: EventWindowNode = leaf([]) + private snapshot: SessionEventWindow = windowSnapshot( + this.window, + false, + 0, + { kind: 'replace', entries: [] }, + ) /** @returns the cached event-window snapshot. */ getSnapshot(): SessionEventWindow { return this.snapshot } @@ -48,7 +109,8 @@ export class MutableSessionEventSource implements SessionEventSource { * @param hasMore - whether older history remains. */ replace(entries: readonly SessionEventEntry[], hasMore: boolean): void { - this.publish(entries, hasMore, { kind: 'replace', entries }) + this.window = leaf(entries) + this.publish(hasMore, { kind: 'replace', entries }) } /** @@ -57,7 +119,8 @@ export class MutableSessionEventSource implements SessionEventSource { * @param hasMore - whether still older history remains. */ prepend(entries: readonly SessionEventEntry[], hasMore: boolean): void { - this.publish([...entries, ...this.snapshot.entries], hasMore, { kind: 'prepend', entries }) + this.window = concat(leaf(entries), this.window) + this.publish(hasMore, { kind: 'prepend', entries }) } /** @@ -65,18 +128,19 @@ export class MutableSessionEventSource implements SessionEventSource { * @param entry - live tail entry. */ append(entry: SessionEventEntry): void { - this.publish([...this.snapshot.entries, entry], this.snapshot.hasMore, { + const entries = [entry] + this.window = concat(this.window, leaf(entries)) + this.publish(this.snapshot.hasMore, { kind: 'append', - entries: [entry], + entries, }) } private publish( - entries: readonly SessionEventEntry[], hasMore: boolean, change: SessionEventChange, ): void { - this.snapshot = { entries, hasMore, revision: this.snapshot.revision + 1, change } + this.snapshot = windowSnapshot(this.window, hasMore, this.snapshot.revision + 1, change) notifySubscribers(this.listeners, '[session-controller] event feed') } } diff --git a/packages/api/session-controller/tests/client-contract.client.spec.ts b/packages/api/session-controller/tests/client-contract.client.spec.ts index 18919e66e9..9b3a7fa015 100644 --- a/packages/api/session-controller/tests/client-contract.client.spec.ts +++ b/packages/api/session-controller/tests/client-contract.client.spec.ts @@ -54,6 +54,27 @@ describe('Client Session contracts', () => { expect(listener).toHaveBeenCalledTimes(3) }) + it('does not traverse the complete event window while appending', () => { + const feed = new MutableSessionEventSource() + const first = entry(1) + const base = [first] + const iterate = vi.fn(Array.prototype[Symbol.iterator].bind(base)) + Object.defineProperty(base, Symbol.iterator, { value: iterate }) + feed.replace(base, false) + iterate.mockClear() + + const before = feed.getSnapshot() + const live = entry(2) + feed.append(live) + const after = feed.getSnapshot() + + expect(iterate).not.toHaveBeenCalled() + expect(before.entries).toEqual([first]) + expect(after.entries).toEqual([first, live]) + expect(after.entries).toBe(after.entries) + expect(iterate).toHaveBeenCalledOnce() + }) + it('folds Error and non-Error carrier rejections into Client failures', () => { expect(transportResult(new Error('transport unavailable'))).toEqual({ ok: false, diff --git a/packages/client/ui-conversation/src/client/conversation/assembly.ts b/packages/client/ui-conversation/src/client/conversation/assembly.ts index 2a996d6296..cf0c4380b7 100644 --- a/packages/client/ui-conversation/src/client/conversation/assembly.ts +++ b/packages/client/ui-conversation/src/client/conversation/assembly.ts @@ -159,9 +159,18 @@ export class UiConversation extends Service { const rebuild = (): void => { for (const record of this.bindings.values()) record.binding.rebuild() } + let rebuildQueued = false + const scheduleRebuild = (): void => { + if (rebuildQueued) return + rebuildQueued = true + queueMicrotask(() => { + rebuildQueued = false + rebuild() + }) + } ctx.effect(() => { - const disposeEvents = this.events.subscribe(rebuild) - const disposeViews = this.views.subscribe(rebuild) + const disposeEvents = this.events.subscribe(scheduleRebuild) + const disposeViews = this.views.subscribe(scheduleRebuild) return () => { disposeViews() disposeEvents() diff --git a/packages/client/ui-conversation/tests/conversation-registry.client.spec.ts b/packages/client/ui-conversation/tests/conversation-registry.client.spec.ts index 06e76ce8c2..3df509dcdf 100644 --- a/packages/client/ui-conversation/tests/conversation-registry.client.spec.ts +++ b/packages/client/ui-conversation/tests/conversation-registry.client.spec.ts @@ -226,15 +226,21 @@ describe('Conversation registries', () => { expect(views.entries()).toEqual([]) }) - it('rebuilds every resident Conversation binding after each registry change', async () => { + it('coalesces one turn of registry changes into one rebuild per resident Conversation', async () => { const { uiConversation, binding, events, views } = await bootRegistries() uiConversation.binding(binding) const rebuild = vi.spyOn(ConversationNodeAssembler.prototype, 'rebuildRegistry') events.register(eventDefinition('message')) + views.register(viewDefinition('chat')) + events.register(eventDefinition('tool')) + expect(rebuild).not.toHaveBeenCalled() + + await Promise.resolve() expect(rebuild).toHaveBeenCalledOnce() - views.register(viewDefinition('chat')) + views.register(viewDefinition('trajectory')) + await Promise.resolve() expect(rebuild).toHaveBeenCalledTimes(2) rebuild.mockRestore() })