打开历史会话失败：同一 callId 出现第二次 tool/call 时汇编器抛 "received more than one start Match" #5247
Replies: 7 comments
|
This is a real, reproducible assembler defect, and your evidence (the The throw is real and (today) intentional.
if (role === 'start' && context?.start !== undefined) {
throw new Error(`conversation Context ${key} received more than one start Match`)
}The test suite actively pins this as a designed rejection: So a naive "don't throw on duplicate start" would break that test and, more importantly, silently drop a genuinely malformed duplicate. Why legacy sessions hit it and fresh code doesn't. The context key is This is the frontend counterpart of the server-side BlockAssembler index-reuse issue ( A discriminating fix direction (helps both the legacy case and keeps the malformed case loud): The key insight is that a duplicate start for the same key in a different turn/step is legitimate, while a duplicate within the same turn is malformed. Some options worth weighing:
Your distinction matters for the fix: The acceptance condition you set — open a session with repeated callIds without throwing and render it — is exactly the right spec. I'd suggest a regression test that seeds two turns, each with |
|
Local patch that fixed this for us (in case it helps the upstream fix design): in ConversationNodeAssembler (packages/client/ui-conversation/src/client/conversation/assembler.ts), we replaced the hard throw on duplicate start matches with sibling forking: a families map tracks base business key -> context keys; a repeated start for an already-started key forks a suffixed sibling context (key#2, key#3...); non-start matches route to the newest sibling whose startSeq <= event seq; the prepend batch path segments entries per start the same way. 3 new tests cover append / window-replay / prepend paths; full ui-conversation suite (337 tests) passes. Sessions with reused callIds (pwsh:0 etc.) now load completely. |
|
Thanks for the thorough analysis, @argszero — your breakdown helped us a lot. We went ahead and implemented option 2 (re-key on duplicate) as a local patch, and it works well for the legacy sessions we have:
Tests: we replaced the old "rejects duplicate start" test with sibling-fork expectations and added window-replay and prepend cases (e.g. To answer your question directly: runtime tolerance (option 1/2) is strongly preferable for us — option 3 (load-time normalization) would fix reading but leaves every new legacy-style write broken, and the writer side (short index ids) is still emitted by some providers/configurations in the wild, not just old logs. Caveat we are aware of: our patch does not discriminate "same invocation replayed" (genuinely malformed) from "new invocation, reused id" (legacy). It simply tolerates both by forking. If upstream wants to keep the malformed case loud, combining fork-on-duplicate with the turn/step discriminator from option 1 would give both: fork when turn/step differs, throw when the duplicate lands inside the same turn/step. We kept ours simple since client-side rendering has no real risk from over-tolerance — worst case, one malformed write renders as two nodes instead of crashing the whole conversation. Happy to share the diff if useful. |
|
@dafish-aurelia thanks for the follow-through — sibling-forking is a clean realization of option 2, and the "non-start matches route to the newest sibling whose startSeq <= event.seq" rule is the part that makes it robust across both prepend (older pages) and live append. Two notes: 1. On the caveat you flagged (not discriminating "same invocation replayed" from "new invocation, reused id"): for the client renderer, over-tolerance is the right trade — the worst case is one extra rendered node, not a crashed conversation. If upstream later wants to keep the malformed case loud, the turn/step discriminator you mention is the correct axis, and it aligns with the server-side analysis in #5268: a duplicate start landing in the same (turn, step) is genuinely malformed; a reused short id across turns is legitimate legacy data. Fork-when-different-turn, throw-when-same-turn is a coherent combination. 2. The deeper point from #5268: the fork is a reader-side tolerance, and it's the right stopgap for the legacy sessions you have. The root cause is that the provider-emitted id and the harness-internal invocation identity currently share one field ( Happy to take a look at the diff if you share it — the window-replay and prepend cases are the ones I'd want to see covered, and you've got both. |
|
Here's the diff as requested, @argszero. The assembler change (option 2) plus the three tests — including the window-replay and prepend cases you asked about: packages/client/ui-conversation/src/client/conversation/assembler.ts diff --git "a/G:\\Temp\\dsh-diff-5247\\assembler.orig.ts" "b/G:\\life\\Aurelia\\deepseek-harness-v0.1.2\\packages\\client\\ui-conversation\\src\\client\\conversation\\assembler.ts"
index a8267bd..809af1f 100644
--- "a/G:\\Temp\\dsh-diff-5247\\assembler.orig.ts"
+++ "b/G:\\life\\Aurelia\\deepseek-harness-v0.1.2\\packages\\client\\ui-conversation\\src\\client\\conversation\\assembler.ts"
@@ -158,6 +158,8 @@ export class ConversationNodeAssembler implements ConversationViewSnapshotStore
private readonly contexts = new Map<string, InternalContext>()
private readonly contextsByKind = new Map<string, InternalContext[]>()
private readonly contextsBySeq = new Map<number, Set<InternalContext>>()
+ /** Base business key -> sibling Context keys forked by repeated starts. */
+ private readonly families = new Map<string, string[]>()
private readonly inputs = new Map<number, SessionEventLikeEntry>()
private readonly locationIndex = new ConversationLocationIndex()
private readonly dirty = new Set<InternalContext>()
@@ -189,6 +191,7 @@ export class ConversationNodeAssembler implements ConversationViewSnapshotStore
this.contexts.clear()
this.contextsByKind.clear()
this.contextsBySeq.clear()
+ this.families.clear()
this.inputs.clear()
this.dirty.clear()
this.revised.clear()
@@ -419,34 +422,86 @@ export class ConversationNodeAssembler implements ConversationViewSnapshotStore
return publication
}
+ private createContext(
+ key: string,
+ definition: ConversationNodeDefinition,
+ id: string,
+ ): InternalContext {
+ const context: InternalContext = {
+ key,
+ kind: definition.kind,
+ id,
+ definition,
+ startSeq: undefined,
+ start: undefined,
+ matches: [],
+ state: undefined,
+ revision: 0,
+ current: new Map(),
+ locationData: emptyLocationData(),
+ dependencies: new Map(),
+ }
+ this.contexts.set(key, context)
+ return context
+ }
+
+ /** Track a Context under its base business key; forks share one entry list. */
+ private registerFamily(baseKey: string, key: string): void {
+ const list = this.families.get(baseKey) ?? []
+ if (!list.includes(key)) list.push(key)
+ this.families.set(baseKey, list)
+ }
+
+ /**
+ * Key for a start Match. Providers may reuse short business ids (e.g.
+ * `pwsh:0`), so a repeated start forks a suffixed sibling Context instead
+ * of crashing the feed. A Context that only collected updates so far
+ * adopts the arriving start.
+ */
+ private resolveStartKey(baseKey: string): string {
+ const list = this.families.get(baseKey) ?? []
+ const idle = list.find(key => this.contexts.get(key)?.startSeq === undefined)
+ if (idle !== undefined) return idle
+ if (list.length === 0 && !this.contexts.has(baseKey)) return baseKey
+ let ordinal = 2
+ let key = `${baseKey}#${ordinal}`
+ while (this.contexts.has(key)) key = `${baseKey}#${++ordinal}`
+ return key
+ }
+
+ /**
+ * Key for a non-start Match: the newest sibling whose start precedes this
+ * event, so updates land on their own run when ids repeat.
+ */
+ private resolveUpdateKey(baseKey: string, seq: number): string {
+ const list = this.families.get(baseKey)
+ if (list === undefined || list.length === 0) return baseKey
+ let best: string | undefined
+ let bestSeq = Number.NEGATIVE_INFINITY
+ for (const key of list) {
+ const startSeq = this.contexts.get(key)?.startSeq
+ if (startSeq !== undefined && startSeq <= seq && startSeq >= bestSeq) {
+ best = key
+ bestSeq = startSeq
+ }
+ }
+ if (best !== undefined) return best
+ const idle = list.find(key => this.contexts.get(key)?.startSeq === undefined)
+ return idle ?? (list[0] as string)
+ }
+
private acceptMatch(
definition: ConversationNodeDefinition,
id: string,
role: ConversationMatch['role'],
input: SessionEventLikeEntry,
): ConversationPublication {
- const key = conversationContextKey(definition.kind, id)
- let context = this.contexts.get(key)
- if (role === 'start' && context?.start !== undefined) {
- throw new Error(`conversation Context ${key} received more than one start Match`)
- }
- if (context === undefined) {
- context = {
- key,
- kind: definition.kind,
- id,
- definition,
- startSeq: undefined,
- start: undefined,
- matches: [],
- state: undefined,
- revision: 0,
- current: new Map(),
- locationData: emptyLocationData(),
- dependencies: new Map(),
- }
- this.contexts.set(key, context)
- }
+ const baseKey = conversationContextKey(definition.kind, id)
+ const key = role === 'start'
+ ? this.resolveStartKey(baseKey)
+ : this.resolveUpdateKey(baseKey, input.event.seq)
+ const context = this.contexts.get(key) ?? this.createContext(key, definition, id)
+ this.registerFamily(baseKey, key)
const match = conversationMatch(
key,
input,
@@ -487,58 +542,52 @@ export class ConversationNodeAssembler implements ConversationViewSnapshotStore
affected: Set<InternalContext>,
): void {
const startsByKind = new Map<string, InternalContext[]>()
- for (const [key, entries] of pending) {
- const first = entries[0]
- if (first === undefined) continue
- let context = this.contexts.get(key)
- if (context === undefined) {
- context = {
- key,
- kind: first.definition.kind,
- id: first.id,
- definition: first.definition,
- startSeq: undefined,
- start: undefined,
- matches: [],
- state: undefined,
- revision: 0,
- current: new Map(),
- locationData: emptyLocationData(),
- dependencies: new Map(),
- }
- this.contexts.set(key, context)
- }
+ for (const [baseKey, entries] of pending) {
+ if (entries[0] === undefined) continue
+ // A pending bucket may carry several starts when a provider reused one
+ // business id; split it into per-run segments, each owning a Context.
+ const ordered = [...entries].sort((left, right) => left.match.event.seq - right.match.event.seq)
+ let context: InternalContext | undefined
+ let additions: ConversationMatch[] = []
let discoveredStart: ConversationStartMatch | undefined
- const additions = entries
- .map((entry) => {
- if (entry.definition !== context.definition || entry.id !== context.id) {
- throw new Error(`conversation Context ${key} received inconsistent Definition identity`)
- }
- if (entry.match.role === 'start') {
- if (discoveredStart !== undefined || context.start !== undefined) {
- throw new Error(`conversation Context ${key} received more than one start Match`)
- }
- discoveredStart = entry.match
- }
- const owners = this.contextsBySeq.get(entry.match.event.seq) ?? new Set<InternalContext>()
- owners.add(context)
- this.contextsBySeq.set(entry.match.event.seq, owners)
- return entry.match
- })
- .sort((left, right) => left.event.seq - right.event.seq)
- context.matches = mergeMatches(context.key, additions, context.matches)
- if (discoveredStart !== undefined) {
- context.start = discoveredStart
- context.startSeq = discoveredStart.event.seq
- const starts = startsByKind.get(context.kind) ?? []
- starts.push(context)
- startsByKind.set(context.kind, starts)
+ const flushSegment = () => {
+ if (context === undefined || additions.length === 0) return
+ context.matches = mergeMatches(context.key, additions, context.matches)
+ if (discoveredStart !== undefined) {
+ context.start = discoveredStart
+ context.startSeq = discoveredStart.event.seq
+ const starts = startsByKind.get(context.kind) ?? []
+ starts.push(context)
+ startsByKind.set(context.kind, starts)
+ }
+ if (context.start !== undefined && context.matches[0] !== context.start) {
+ throw new Error(`conversation Context ${context.key} received an update before its start Match`)
+ }
+ affected.add(context)
+ this.dirty.add(context)
}
- if (context.start !== undefined && context.matches[0] !== context.start) {
- throw new Error(`conversation Context ${context.key} received an update before its start Match`)
+ for (const entry of ordered) {
+ if (entry.match.role === 'start' || context === undefined) {
+ flushSegment()
+ const key = entry.match.role === 'start'
+ ? this.resolveStartKey(baseKey)
+ : this.resolveUpdateKey(baseKey, entry.match.event.seq)
+ context = this.contexts.get(key) ?? this.createContext(key, entry.definition, entry.id)
+ this.registerFamily(baseKey, key)
+ additions = []
+ discoveredStart = entry.match.role === 'start'
+ ? entry.match as ConversationStartMatch
+ : undefined
+ }
+ if (entry.definition !== context.definition || entry.id !== context.id) {
+ throw new Error(`conversation Context ${context.key} received inconsistent Definition identity`)
+ }
+ const owners = this.contextsBySeq.get(entry.match.event.seq) ?? new Set<InternalContext>()
+ owners.add(context)
+ this.contextsBySeq.set(entry.match.event.seq, owners)
+ additions.push(entry.match)
}
- affected.add(context)
- this.dirty.add(context)
+ flushSegment()
}
for (const [kind, contexts] of startsByKind) this.indexStartedContexts(kind, contexts)
}
` ``
**packages/client/ui-conversation/tests/conversation-assembler.client.spec.ts**
```diff
diff --git "a/G:\\Temp\\dsh-diff-5247\\spec.orig.ts" "b/G:\\life\\Aurelia\\deepseek-harness-v0.1.2\\packages\\client\\ui-conversation\\tests\\conversation-assembler.client.spec.ts"
index 6465559..99e0ce6 100644
--- "a/G:\\Temp\\dsh-diff-5247\\spec.orig.ts"
+++ "b/G:\\life\\Aurelia\\deepseek-harness-v0.1.2\\packages\\client\\ui-conversation\\tests\\conversation-assembler.client.spec.ts"
@@ -1244,12 +1244,21 @@ describe('ConversationNodeAssembler', () => {
)).toThrow(/Definition "undefined-update" returned undefined from update/)
})
- it('rejects a duplicate start before mutating the existing Context', () => {
- const definition: ConversationNodeDefinition<number> = {
+ it('forks a duplicate start into a sibling Context instead of throwing', () => {
+ interface RunState { readonly callSeq: number; readonly results: number }
+ const definition: ConversationNodeDefinition<RunState> = {
kind: 'single-start',
- match: event => (event.type as string) === 'command/run' ? { id: 'one', role: 'start' } : null,
- start: (_context, match) => match.event.seq,
- update: context => context.state,
+ match: (event) => {
+ if ((event.type as string) === 'command/run') {
+ return { id: String((event.data as { commandId: string }).commandId), role: 'start' }
+ }
+ if ((event.type as string) === 'command/done') {
+ return { id: String((event.data as { commandId: string }).commandId), role: 'update' }
+ }
+ return null
+ },
+ start: (_context, match) => ({ callSeq: match.event.seq, results: 0 }),
+ update: context => ({ ...context.state, results: context.state.results + 1 }),
target: 'test',
buildViewNode: context => node(context, context.state),
}
@@ -1262,10 +1271,107 @@ describe('ConversationNodeAssembler', () => {
], false)
assembler.flush()
- expect(() => assembler.append(
- input(at(2, 'command/run', { commandId: 'two', name: 'x' })),
- )).toThrow(/received more than one start Match/)
+ // A provider reusing the same business id must not crash the feed.
+ assembler.append(input(at(2, 'command/run', { commandId: 'one', name: 'x' })))
+ assembler.append(input(at(3, 'command/done', { commandId: 'one', kind: 'success' })))
+ assembler.append(input(at(4, 'command/run', { commandId: 'one', name: 'x' })))
+ assembler.append(input(at(5, 'command/done', { commandId: 'one', kind: 'success' })))
assembler.flush()
- expect([...testSnapshot(assembler)?.nodes.values() ?? []][0]?.data).toBe(1)
+
+ const nodes = [...testSnapshot(assembler)?.nodes.values() ?? []]
+ expect(nodes.map(value => value.data)).toEqual([
+ { callSeq: 1, results: 0 },
+ { callSeq: 2, results: 1 },
+ { callSeq: 4, results: 1 },
+ ])
+ expect(new Set(nodes.map(value => value.key)).size).toBe(3)
+ })
+
+ it('forks duplicate starts replayed inside one window into ordered siblings', () => {
+ interface RunState { readonly callSeq: number; readonly results: number }
+ const definition: ConversationNodeDefinition<RunState> = {
+ kind: 'dup-tool',
+ match: (event) => {
+ if (event.type === 'tool/call') return { id: String(event.data.callId), role: 'start' }
+ if (event.type === 'tool/result') {
+ return { id: String(event.data.message.source.callId), role: 'update' }
+ }
+ return null
+ },
+ start: (_context, match) => ({ callSeq: match.event.seq, results: 0 }),
+ update: context => ({ ...context.state, results: context.state.results + 1 }),
+ target: 'test',
+ buildViewNode: context => node(context, context.state),
+ }
+ const assembler = new ConversationNodeAssembler(
+ new TestEventDefinitions([definition]),
+ new TestViewDefinitions([testView()]),
+ )
+ assembler.replaceWindow([
+ input(at(1, 'tool/call', { turn: 1, step: 1, callId: 'pwsh:0', name: 'x', arguments: '{}' })),
+ input(at(2, 'tool/result', {
+ turn: 1, step: 1,
+ message: { source: { type: 'tool-result', callId: 'pwsh:0' }, content: [], isError: false },
+ })),
+ input(at(3, 'tool/call', { turn: 1, step: 1, callId: 'pwsh:0', name: 'x', arguments: '{}' })),
+ input(at(4, 'tool/result', {
+ turn: 1, step: 1,
+ message: { source: { type: 'tool-result', callId: 'pwsh:0' }, content: [], isError: false },
+ })),
+ ], false)
+ assembler.flush()
+
+ const nodes = [...testSnapshot(assembler)?.nodes.values() ?? []]
+ expect(nodes.map(value => value.data)).toEqual([
+ { callSeq: 1, results: 1 },
+ { callSeq: 3, results: 1 },
+ ])
+ })
+
+ it('forks duplicate starts arriving through prepend and keeps updates on their own run', () => {
+ interface RunState { readonly callSeq: number; readonly results: number }
+ const definition: ConversationNodeDefinition<RunState> = {
+ kind: 'dup-tool',
+ match: (event) => {
+ if (event.type === 'tool/call') return { id: String(event.data.callId), role: 'start' }
+ if (event.type === 'tool/result') {
+ return { id: String(event.data.message.source.callId), role: 'update' }
+ }
+ return null
+ },
+ start: (_context, match) => ({ callSeq: match.event.seq, results: 0 }),
+ update: context => ({ ...context.state, results: context.state.results + 1 }),
+ target: 'test',
+ buildViewNode: context => node(context, context.state),
+ }
+ const assembler = new ConversationNodeAssembler(
+ new TestEventDefinitions([definition]),
+ new TestViewDefinitions([testView()]),
+ )
+ assembler.replaceWindow([
+ input(at(3, 'tool/call', { turn: 1, step: 1, callId: 'pwsh:0', name: 'x', arguments: '{}' })),
+ ], true)
+ assembler.flush()
+
+ assembler.prepend([
+ input(at(1, 'tool/call', { turn: 1, step: 1, callId: 'pwsh:0', name: 'x', arguments: '{}' })),
+ input(at(2, 'tool/result', {
+ turn: 1, step: 1,
+ message: { source: { type: 'tool-result', callId: 'pwsh:0' }, content: [], isError: false },
+ })),
+ ], false)
+ assembler.flush()
+ // A later live result belongs to the newer sibling, not the prepended one.
+ assembler.append(input(at(4, 'tool/result', {
+ turn: 1, step: 1,
+ message: { source: { type: 'tool-result', callId: 'pwsh:0' }, content: [], isError: false },
+ })))
+ assembler.flush()
+
+ const nodes = [...testSnapshot(assembler)?.nodes.values() ?? []]
+ expect(nodes.map(value => value.data)).toEqual([
+ { callSeq: 3, results: 1 },
+ { callSeq: 1, results: 1 },
+ ])
})
})
` ``
Local build against `v0.1.2-alpha.1`; full ui-conversation suite passes (337 tests). |
|
Reviewed the diff — it's a coherent realization of option 2, and the three tests cover exactly the paths I wanted to see (single-window duplicate, window-replay, prepend). A few notes from the read-through:
Solid patch — thanks for sharing it. For upstream adoption, the natural next step is the |
|
Thanks for the careful review, @argszero — and good catch on the suffix scheme. We adopted the nit locally: the fork separator is now a NUL character ( Agreed on the rest: the routing rule stays the load-bearing part, and over-tolerance remains a documented renderer trade for us; the turn/step discriminator (fork across turns, throw within one) is noted as the optional tightening if upstream wants malformed duplicates loud again. And yes — the |
Uh oh!
There was an error while loading. Please reload this page.
打开某些较旧的历史会话时前端打不开，会话面板空白——前端 conversation assembler 在遇到同一会话上下文重复收到
role: start的tool/call事件时直接throw，订阅整条事件流的 feed subscriber 因此中断。例如打开标题为「v0.1.2-alpha.1」的会话即复现，控制台报错：
复现、预期与验收
callId的多次tool/call（如某些模型的工具会以pwsh:0、grep:0这种短 id 写回；在一条 16 轮会话里实锤撞车：pwsh:0×45、grep:0×11、read:0×8、grep:1×5、pwsh:1×7、write:0×4、edit:0×3、todo_write:0×4）。trajectory-tool-calldefinition 时，assembler.ts的acceptMatch判定context.start !== undefined即抛错，BoundConversation 进而崩溃。contexts未按 state 分组时不允许重复 start）。callId收到第二次 start 时不再抛错——容许将该调用视为同 id 的另一次调用（同 context 重开/覆盖状态）或安全降级为 warn，而不是让整个会话进不来。All reactions