perf(conversation): reuse unchanged location projections
This commit is contained in:
1 parent
203e2440ac
commit
81431381d6
8 files changed
+169
-31
No files matched your search
@@ -364,6 +364,13 @@ function projectAssistant(context: ConversationNodeContext<AssistantState>): Ass
|
||||
}
|
||||
}
|
||||
|
||||
function publishedAssistantData(
|
||||
context: ConversationNodeContext<AssistantState>,
|
||||
): Readonly<AssistantChatData> | undefined {
|
||||
const location = context.start?.location ?? context.matches.at(-1)?.location
|
||||
return location?.kind === 'step' ? location.step.data.get('assistant-step') : undefined
|
||||
}
|
||||
|
||||
/** Per-step Assistant streaming/final/interruption Definition. */
|
||||
export const assistantDefinition: ConversationNodeDefinition<AssistantState> = {
|
||||
kind: 'assistant-step',
|
||||
@@ -427,16 +434,19 @@ export const assistantDefinition: ConversationNodeDefinition<AssistantState> = {
|
||||
}
|
||||
},
|
||||
buildViewNode: (context) => {
|
||||
const projected = projectAssistant(context)
|
||||
if (projected === undefined) return null
|
||||
if (projected.settled === undefined && !projected.visible) {
|
||||
const state = context.state ?? fallbackState(context)
|
||||
if (state === undefined) return null
|
||||
const state = context.state ?? fallbackState(context)
|
||||
if (state === undefined) return null
|
||||
const data = publishedAssistantData(context)
|
||||
if (data === undefined) return null
|
||||
const settled = data.finalNode
|
||||
const visible = settled === undefined ? state.visibleBlocks > 0 : hasVisibleContent(data.blocks)
|
||||
if (settled === undefined && !visible) {
|
||||
const current = context.current.get('chat')
|
||||
if (!state.hidden || current === undefined || current === null) return null
|
||||
}
|
||||
return chatNode(context, 'assistant-step', projected.anchorSeq, projected.data, {
|
||||
visibility: projected.settled?.interrupted === true || projected.visible ? 'visible' : 'hidden',
|
||||
const anchorSeq = settled?.seq ?? state.firstVisibleSeq ?? context.matches[0]?.event.seq ?? 0
|
||||
return chatNode(context, 'assistant-step', anchorSeq, data, {
|
||||
visibility: settled?.interrupted === true || visible ? 'visible' : 'hidden',
|
||||
})
|
||||
},
|
||||
}
|
||||
|
||||
@@ -7,7 +7,7 @@ import type {} from '@deepseek-ai/dsh-llm-retry/types'
|
||||
import { isAppendSurfaceEvent } from '@deepseek-ai/dsh-session/surface'
|
||||
import type {} from '@deepseek-ai/dsh-tools/types'
|
||||
import { hasAssistantReplyContent } from '../contract/assistant-content.ts'
|
||||
import type { AssistantChatData, FinalAssistantChatData } from '../contract/chat-nodes.ts'
|
||||
import type { AssistantChatData, ChatNode, FinalAssistantChatData } from '../contract/chat-nodes.ts'
|
||||
import {
|
||||
decodeTurnProcess, encodeTurnProcess, isSubagentDelegationTool,
|
||||
type TurnProcessSignature, type TurnProcessSpec,
|
||||
@@ -34,6 +34,8 @@ interface TurnProcessState {
|
||||
readonly assistantStartByStep: ReadonlyMap<number, number>
|
||||
readonly messageCountByStep: ReadonlyMap<number, number>
|
||||
readonly otherStartSeq?: number
|
||||
readonly controlAnchorSeq?: number
|
||||
readonly messageCount: number
|
||||
readonly toolCallCount: number
|
||||
readonly subagentCount: number
|
||||
}
|
||||
@@ -109,6 +111,7 @@ function fallbackState(context: ConversationNodeContext<TurnProcessState>): Turn
|
||||
turn,
|
||||
assistantStartByStep: new Map(),
|
||||
messageCountByStep: new Map(),
|
||||
messageCount: 0,
|
||||
toolCallCount: 0,
|
||||
subagentCount: 0,
|
||||
}
|
||||
@@ -130,15 +133,12 @@ function latestAnswer(turn: TurnLocation): Readonly<FinalAssistantChatData> | nu
|
||||
}
|
||||
|
||||
function processSpec(state: TurnProcessState, turn: TurnLocation): TurnProcessSpec | null {
|
||||
const controlAnchorSeq = Math.min(
|
||||
state.otherStartSeq ?? Number.POSITIVE_INFINITY,
|
||||
...state.assistantStartByStep.values(),
|
||||
)
|
||||
if (!Number.isFinite(controlAnchorSeq)) return null
|
||||
const controlAnchorSeq = state.controlAnchorSeq
|
||||
if (controlAnchorSeq === undefined) return null
|
||||
const answer = latestAnswer(turn)
|
||||
const counts = {
|
||||
messageCount: answer === null
|
||||
? [...state.messageCountByStep.values()].reduce((total, count) => total + count, 0)
|
||||
? state.messageCount
|
||||
: [...state.messageCountByStep]
|
||||
.filter(([step]) => step < answer.step)
|
||||
.reduce((total, [, count]) => total + count, 0),
|
||||
@@ -185,7 +185,7 @@ function updateProcessState(state: TurnProcessState, event: ConversationEvent):
|
||||
&& hasAssistantReplyContent(toAssistantBlocks(event.data.message.content))) {
|
||||
const messageCountByStep = new Map(current.messageCountByStep)
|
||||
messageCountByStep.set(event.data.step, (messageCountByStep.get(event.data.step) ?? 0) + 1)
|
||||
current = { ...current, messageCountByStep }
|
||||
current = { ...current, messageCountByStep, messageCount: current.messageCount + 1 }
|
||||
}
|
||||
if (event.type === 'tool/call') {
|
||||
const subagent = isSubagentDelegationTool(event.data.name)
|
||||
@@ -198,12 +198,22 @@ function updateProcessState(state: TurnProcessState, event: ConversationEvent):
|
||||
const evidence = processEvidence(event)
|
||||
if (evidence === undefined) return current
|
||||
if (evidence.kind === 'other') {
|
||||
return current.otherStartSeq === undefined ? { ...current, otherStartSeq: evidence.seq } : current
|
||||
return current.otherStartSeq === undefined
|
||||
? {
|
||||
...current,
|
||||
otherStartSeq: evidence.seq,
|
||||
controlAnchorSeq: Math.min(current.controlAnchorSeq ?? Number.POSITIVE_INFINITY, evidence.seq),
|
||||
}
|
||||
: current
|
||||
}
|
||||
if (current.assistantStartByStep.has(evidence.step)) return current
|
||||
const assistantStartByStep = new Map(current.assistantStartByStep)
|
||||
assistantStartByStep.set(evidence.step, evidence.seq)
|
||||
return { ...current, assistantStartByStep }
|
||||
return {
|
||||
...current,
|
||||
assistantStartByStep,
|
||||
controlAnchorSeq: Math.min(current.controlAnchorSeq ?? Number.POSITIVE_INFINITY, evidence.seq),
|
||||
}
|
||||
}
|
||||
|
||||
/** Turn-scoped process range and answer-boundary Definition. */
|
||||
@@ -233,6 +243,7 @@ export const turnProcessDefinition: ConversationNodeDefinition<TurnProcessState>
|
||||
turn: match.event.data.turn,
|
||||
assistantStartByStep: new Map(),
|
||||
messageCountByStep: new Map(),
|
||||
messageCount: 0,
|
||||
toolCallCount: 0,
|
||||
subagentCount: 0,
|
||||
}
|
||||
@@ -246,24 +257,54 @@ export const turnProcessDefinition: ConversationNodeDefinition<TurnProcessState>
|
||||
}
|
||||
return 'immediate'
|
||||
},
|
||||
buildLocationData: (context, scope) => {
|
||||
buildLocationData: (context, scope, previous) => {
|
||||
if (scope !== 'turn') return null
|
||||
const state = context.state ?? fallbackState(context)
|
||||
if (state === undefined) return null
|
||||
const turn = turnLocation(context)
|
||||
if (turn === undefined) return null
|
||||
const current = context.current.get('chat') as ChatNode | null | undefined
|
||||
const latestStep = turn.steps.at(-1)
|
||||
if (previous?.kind === 'turn'
|
||||
&& previous.key === 'turn-process'
|
||||
&& current?.kind === 'turn-process'
|
||||
&& current.data.answerAnchorSeq === null
|
||||
&& current.data.controlAnchorSeq === state.controlAnchorSeq
|
||||
&& current.data.messageCount === state.messageCount
|
||||
&& current.data.toolCallCount === state.toolCallCount
|
||||
&& current.data.subagentCount === state.subagentCount
|
||||
&& turn.status !== 'closed'
|
||||
&& latestStep?.status !== 'closed') return previous
|
||||
const spec = processSpec(state, turn)
|
||||
return spec === null ? null : {
|
||||
if (spec === null) return null
|
||||
const value = encodeTurnProcess(spec)
|
||||
if (previous?.kind === 'turn'
|
||||
&& previous.turn === spec.turn
|
||||
&& previous.key === 'turn-process'
|
||||
&& previous.value === value) return previous
|
||||
return {
|
||||
kind: 'turn',
|
||||
turn: turn.turn,
|
||||
key: 'turn-process',
|
||||
value: encodeTurnProcess(spec),
|
||||
value,
|
||||
}
|
||||
},
|
||||
buildViewNode: (context) => {
|
||||
const turn = turnLocation(context)
|
||||
const signature = turn?.data.get('turn-process')
|
||||
if (turn === undefined || signature === undefined) return null
|
||||
const current = context.current.get('chat') as ChatNode | null | undefined
|
||||
const state = context.state
|
||||
if (current?.kind === 'turn-process'
|
||||
&& state !== undefined
|
||||
&& current.data.answerAnchorSeq === null
|
||||
&& current.data.controlAnchorSeq === state.controlAnchorSeq
|
||||
&& current.data.messageCount === state.messageCount
|
||||
&& current.data.toolCallCount === state.toolCallCount
|
||||
&& current.data.subagentCount === state.subagentCount
|
||||
&& turn.status !== 'closed'
|
||||
&& turn.steps.at(-1)?.status !== 'closed'
|
||||
&& current.location === (context.start?.location ?? context.matches[0]?.location)) return current
|
||||
const data = decodeTurnProcess(signature)
|
||||
return chatNode(
|
||||
context,
|
||||
|
||||
@@ -134,8 +134,9 @@ function hasText(data: AssistantChatData): data is FinalAssistantChatData {
|
||||
}
|
||||
|
||||
function tailData(context: ConversationNodeContext<TurnTailState>): TurnTailChatData | null {
|
||||
const end = context.state?.end
|
||||
?? context.matches.find(match => match.event.type === 'turn/end')
|
||||
const end = context.state === undefined
|
||||
? context.matches.find(match => match.event.type === 'turn/end')
|
||||
: context.state.end
|
||||
if (end?.event.type !== 'turn/end') return null
|
||||
const turn = turnLocation(context)
|
||||
if (turn === undefined) return null
|
||||
|
||||
@@ -608,6 +608,30 @@ describe('built-in conversation node Definitions', () => {
|
||||
expect(settled.answerStep).toBe(1)
|
||||
})
|
||||
|
||||
it('reuses the open Turn-process projection across continuing Assistant chunks', () => {
|
||||
const value = assembler([
|
||||
at(1, 'turn/start', { turn: 1 }),
|
||||
at(2, 'step/start', { turn: 1, step: 1 }),
|
||||
at(3, 'assistant/chunk', {
|
||||
turn: 1, step: 1, chunk: { type: 'reasoning-delta', index: 0, text: 'first' },
|
||||
}),
|
||||
])
|
||||
const before = snapshot(value)
|
||||
const processNode = node(before, 'turn-process')
|
||||
const processSignature = before.timeline.turns.get(1)?.data.get('turn-process')
|
||||
const assistantNode = node(before, 'assistant-step')
|
||||
|
||||
value.append(at(4, 'assistant/chunk', {
|
||||
turn: 1, step: 1, chunk: { type: 'reasoning-delta', index: 0, text: ' second' },
|
||||
}))
|
||||
value.flush()
|
||||
|
||||
const after = snapshot(value)
|
||||
expect(after.timeline.turns.get(1)?.data.get('turn-process')).toBe(processSignature)
|
||||
expect(node(after, 'turn-process')).toBe(processNode)
|
||||
expect(node(after, 'assistant-step')).not.toBe(assistantNode)
|
||||
})
|
||||
|
||||
it('anchors a streamed non-text answer from its block start', () => {
|
||||
const value = assembler([
|
||||
at(50, 'turn/start', { turn: 5 }),
|
||||
@@ -640,6 +664,9 @@ describe('built-in conversation node Definitions', () => {
|
||||
const runningSnapshot = snapshot(value)
|
||||
const running = node(runningSnapshot, 'assistant-step')
|
||||
expect(running?.data).toMatchObject({ status: 'running', blocks: [{ kind: 'text', text: 'streaming' }] })
|
||||
expect(running?.location.kind === 'step'
|
||||
? running.location.step.data.get('assistant-step')
|
||||
: undefined).toBe(running?.data)
|
||||
const order = runningSnapshot.order
|
||||
|
||||
value.append(at(4, 'assistant/message', {
|
||||
@@ -654,6 +681,9 @@ describe('built-in conversation node Definitions', () => {
|
||||
expect(settled?.key).toBe(running?.key)
|
||||
expect(settledSnapshot.order).toBe(order)
|
||||
expect(settled?.data).toMatchObject({ status: 'settled', blocks: [{ kind: 'text', text: 'settled' }] })
|
||||
expect(settled?.location.kind === 'step'
|
||||
? settled.location.step.data.get('assistant-step')
|
||||
: undefined).toBe(settled?.data)
|
||||
|
||||
const interruptedValue = assembler([
|
||||
at(10, 'turn/start', { turn: 2 }),
|
||||
|
||||
@@ -210,11 +210,14 @@ export interface ConversationNodeDefinition<State = unknown> {
|
||||
* the same Location key.
|
||||
* @param context - latest complete Context.
|
||||
* @param scope - Location hierarchy level currently being materialized.
|
||||
* @returns current Location value, or null while unavailable.
|
||||
* @param previous - value from the preceding materialization; return it when unchanged.
|
||||
* @returns current Location value, preserving value identity while unchanged,
|
||||
* or null while unavailable.
|
||||
*/
|
||||
buildLocationData?(
|
||||
context: ConversationNodeContext<State>,
|
||||
scope: ConversationLocationDataScope,
|
||||
previous: ConversationLocationData | null,
|
||||
): ConversationLocationData | null
|
||||
/**
|
||||
* Materialize one final Node for this Definition's declared view target.
|
||||
|
||||
@@ -826,9 +826,10 @@ export class ConversationNodeAssembler implements ConversationViewSnapshotStore
|
||||
private buildLocationData(
|
||||
context: InternalContext,
|
||||
scope: ConversationLocationDataScope,
|
||||
previous: ConversationLocationData | null,
|
||||
): ConversationLocationData | null {
|
||||
if (context.definition.buildLocationData === undefined) return null
|
||||
const data = context.definition.buildLocationData(contextSnapshot(context), scope)
|
||||
const data = context.definition.buildLocationData(contextSnapshot(context), scope, previous)
|
||||
if (data === null) return null
|
||||
if (data.kind !== scope) {
|
||||
throw new Error(
|
||||
@@ -853,7 +854,7 @@ export class ConversationNodeAssembler implements ConversationViewSnapshotStore
|
||||
const entries: { owner: string; data: ConversationLocationData }[] = []
|
||||
for (const scope of LOCATION_DATA_SCOPES) {
|
||||
for (const context of this.contexts.values()) {
|
||||
const data = this.buildLocationData(context, scope)
|
||||
const data = this.buildLocationData(context, scope, context.locationData[scope])
|
||||
context.locationData[scope] = data
|
||||
if (data !== null) entries.push({ owner: context.key, data })
|
||||
}
|
||||
@@ -869,9 +870,10 @@ export class ConversationNodeAssembler implements ConversationViewSnapshotStore
|
||||
const changes: ConversationLocationDataChange[] = []
|
||||
for (const context of this.dirty) {
|
||||
const previous = context.locationData[scope]
|
||||
const next = this.buildLocationData(context, scope)
|
||||
const next = this.buildLocationData(context, scope, previous)
|
||||
if (previous === next) continue
|
||||
context.locationData[scope] = next
|
||||
if (previous !== next) changes.push({ owner: context.key, previous, next })
|
||||
changes.push({ owner: context.key, previous, next })
|
||||
}
|
||||
changed = this.locationIndex.applyData(changes) || changed
|
||||
}
|
||||
|
||||
@@ -1005,6 +1005,52 @@ describe('ConversationNodeAssembler', () => {
|
||||
.toEqual({ step: 2, turn: 2 })
|
||||
})
|
||||
|
||||
it('does not republish an equivalent Location value', () => {
|
||||
const apply = vi.fn()
|
||||
const initialValue = { value: 1 }
|
||||
const changedValue = { value: 2 }
|
||||
const definition: ConversationNodeDefinition<{ readonly value: ScopeProbeStepData }> = {
|
||||
kind: 'scope-probe',
|
||||
match: (event) => {
|
||||
if (event.type === 'step/start') return { id: '1:1', role: 'start' }
|
||||
if ((event.type as string) === 'scope-probe/update') return { id: '1:1', role: 'update' }
|
||||
return null
|
||||
},
|
||||
start: () => ({ value: initialValue }),
|
||||
update: (_context, match) => ({
|
||||
value: (match.event.data as unknown as { changed: boolean }).changed
|
||||
? changedValue
|
||||
: initialValue,
|
||||
}),
|
||||
buildLocationData: (context, scope, previous) => {
|
||||
if (scope !== 'step' || context.state === undefined) return null
|
||||
if (previous?.kind === 'step'
|
||||
&& previous.turn === 1
|
||||
&& previous.step === 1
|
||||
&& previous.key === 'scope-probe'
|
||||
&& previous.value === context.state.value) return previous
|
||||
return { kind: 'step', turn: 1, step: 1, key: 'scope-probe', value: context.state.value }
|
||||
},
|
||||
}
|
||||
const assembler = new ConversationNodeAssembler(
|
||||
new TestEventDefinitions([definition]),
|
||||
new TestViewDefinitions([testView(apply)]),
|
||||
)
|
||||
assembler.replaceWindow([
|
||||
input(at(1, 'turn/start', { turn: 1 })),
|
||||
input(at(2, 'step/start', { turn: 1, step: 1 })),
|
||||
], false)
|
||||
assembler.flush()
|
||||
|
||||
assembler.append(input(at(3, 'scope-probe/update', { turn: 1, step: 1, changed: false })))
|
||||
expect(assembler.flush()).toBe(false)
|
||||
expect(apply).not.toHaveBeenCalled()
|
||||
|
||||
assembler.append(input(at(4, 'scope-probe/update', { turn: 1, step: 1, changed: true })))
|
||||
expect(assembler.flush()).toBe(true)
|
||||
expect(apply).toHaveBeenCalledOnce()
|
||||
})
|
||||
|
||||
it('updates existing turn Locations when their Step membership changes', () => {
|
||||
const apply = vi.fn()
|
||||
const definition: ConversationNodeDefinition<null> = {
|
||||
|
||||
@@ -177,14 +177,19 @@ export const deliverablesDefinition: ConversationNodeDefinition<DeliverablesStat
|
||||
? context.state
|
||||
: { ...context.state, produced: [...context.state.produced, { seq: match.event.seq, path }] }
|
||||
},
|
||||
buildLocationData: (context, scope) => scope !== 'turn' || context.state === undefined
|
||||
? null
|
||||
: {
|
||||
buildLocationData: (context, scope, previous) => {
|
||||
if (scope !== 'turn' || context.state === undefined) return null
|
||||
if (previous?.kind === 'turn'
|
||||
&& previous.turn === context.state.turn
|
||||
&& previous.key === 'deliverables'
|
||||
&& previous.value.produced === context.state.produced) return previous
|
||||
return {
|
||||
kind: 'turn',
|
||||
turn: context.state.turn,
|
||||
key: 'deliverables',
|
||||
value: { produced: context.state.produced },
|
||||
},
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
/**
|
||||
|
||||
Reference in new issue
Block a user