diff --git a/.changeset/fix-web-resync-step-blob.md b/.changeset/fix-web-resync-step-blob.md new file mode 100644 index 0000000000..e9b29d680e --- /dev/null +++ b/.changeset/fix-web-resync-step-blob.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +web: Fix a running multi-step turn rendering a duplicated wall of text after the page reconnects or refreshes mid-turn. diff --git a/apps/kimi-web/src/api/daemon/agentEventProjector.ts b/apps/kimi-web/src/api/daemon/agentEventProjector.ts index 4d26110c40..bcfefbb5d3 100644 --- a/apps/kimi-web/src/api/daemon/agentEventProjector.ts +++ b/apps/kimi-web/src/api/daemon/agentEventProjector.ts @@ -101,9 +101,9 @@ interface SessionState { // Assistant message tracking currentAssistantMsgId: string | undefined; - // Per-turn accumulated stream lengths — aligned against the wire `offset` - // on volatile delta frames (v2 sync protocol) to skip duplicates and - // detect gaps after a snapshot seed. + // Per-step accumulated stream lengths — aligned against the (step-relative) + // wire `offset` on volatile delta frames (v2 sync protocol) to skip + // duplicates and detect gaps after a snapshot seed. turnTextLen: number; turnThinkLen: number; @@ -500,9 +500,10 @@ export interface AgentProjector { /** * Seed mid-turn state from a session snapshot's `in_flight_turn` (v2 sync): * resets per-session state, builds the partially-streamed assistant message - * (thinking + text + running tool_use parts), and returns the messageCreated - * AppEvent to apply to the reducer. Live deltas continue appending; their - * wire `offset` aligns against the seeded text so the overlap window around + * (thinking + text + running tool_use parts — the current step only; earlier + * steps arrive via the transcript), and returns the messageCreated AppEvent + * to apply to the reducer. Live deltas continue appending; their wire + * `offset` aligns against the seeded text so the overlap window around * snapshot/subscribe is exact. Session status is NOT seeded here — the REST * snapshot's `session.status` is the authoritative value. */ @@ -573,6 +574,7 @@ export function createAgentProjector(): AgentProjector { s.toolStartTimes.set(tool.toolCallId, Date.now()); } s.currentAssistantMsgId = msg.id; + // Seeded step-relative lengths; the next turn.step.started resets both. s.turnTextLen = turn.assistantText.length; s.turnThinkLen = turn.thinkingText.length; @@ -706,7 +708,7 @@ export function createAgentProjector(): AgentProjector { if (turnId !== undefined) { s.turnPromptId.set(turnId, existingPromptId); } - // Fresh turn → fresh per-turn stream offsets. + // Fresh turn → fresh step stream offsets. s.turnTextLen = 0; s.turnThinkLen = 0; break; @@ -725,6 +727,12 @@ export function createAgentProjector(): AgentProjector { if (turnId !== undefined) s.turnPromptId.set(turnId, promptId); } + // Fresh step → fresh stream offsets: the server's delta `offset` is + // step-relative, so without this reset every delta from step 2 on is + // silently skipped or misread as a gap. + s.turnTextLen = 0; + s.turnThinkLen = 0; + // Create a new pending assistant message const msg = startAssistantMessage(s, sessionId, promptId); s.currentAssistantMsgId = msg.id; diff --git a/apps/kimi-web/src/composables/messagesToTurns.ts b/apps/kimi-web/src/composables/messagesToTurns.ts index cddd47a63e..a08e584b19 100644 --- a/apps/kimi-web/src/composables/messagesToTurns.ts +++ b/apps/kimi-web/src/composables/messagesToTurns.ts @@ -347,13 +347,14 @@ interface Group { /** Client-side measured duration from turn.started to turn.ended (ms). */ durationMs?: number; /** - * Content signatures already folded into this group, used to drop a duplicate - * assistant message. The same logical reply can reach us under two different - * ids — e.g. the streamed copy plus the persisted copy after a reload — and - * since both share the promptId they'd otherwise merge and render the text + - * tool cards twice. Dedupe by exact content so a turn shows each reply once. + * Normalized signatures already folded into this group, used to drop a + * duplicate assistant message. The same logical reply can reach us under two + * different ids — e.g. the streamed copy plus the persisted copy after a + * reload — and since both share the promptId they'd otherwise merge and + * render the text + tool cards twice. Dedupe by normalized content (see + * `contentSig` / `covers`) so a turn shows each reply once. */ - seenSigs: Set; + foldedSigs: ContentSig[]; } // --------------------------------------------------------------------------- @@ -485,6 +486,57 @@ function parsePlanSavedPath(output: string[] | undefined): string | undefined { return undefined; } +/** + * Normalize an assistant message's content for duplicate detection. The same + * logical reply reaches us as both a persisted transcript message and a + * streamed copy (live deltas, or a resync seed from `in_flight_turn`), and the + * two differ in ways that must not defeat the dedup: the persisted thinking + * part may carry a provider `signature`, the seeded tool card may carry + * progress `outputLines`, and the seeded copy concatenates each stream into a + * single part instead of keeping the model's part boundaries. Reduce to the + * concatenated stream text plus sorted tool-call ids — a toolCallId is unique + * per call, so identical id sets mean the same logical message. + */ +interface ContentSig { + text: string; + thinking: string; + toolIds: string[]; + rest: string[]; +} + +function contentSig(content: AppMessage['content']): ContentSig { + let text = ''; + let thinking = ''; + const toolIds: string[] = []; + const rest: string[] = []; + for (const c of content) { + if (c.type === 'text') text += c.text; + else if (c.type === 'thinking') thinking += c.thinking; + else if (c.type === 'toolUse') toolIds.push(c.toolCallId); + else rest.push(JSON.stringify(c)); + } + toolIds.sort(); + rest.sort(); + return { text, thinking, toolIds, rest }; +} + +/** + * Whether an already-folded message's signature fully covers `incoming` — + * i.e. `incoming` is a duplicate of it. Subset, not equality: a resync seed + * carries only the still-running tools of a parallel batch (finished ones + * left `running_tools`), so its id set is a strict subset of the persisted + * message's. Empty text/thinking in `incoming` adds nothing and counts as + * covered. + */ +function covers(folded: ContentSig, incoming: ContentSig): boolean { + if (incoming.text !== '' && incoming.text !== folded.text) return false; + if (incoming.thinking !== '' && incoming.thinking !== folded.thinking) return false; + return ( + incoming.toolIds.every((id) => folded.toolIds.includes(id)) && + incoming.rest.every((j) => folded.rest.includes(j)) + ); +} + export function messagesToTurns( messages: AppMessage[], approvals: AppApprovalRequest[], @@ -614,6 +666,28 @@ export function messagesToTurns( } } + /** + * Fold the volatile extras of a dropped duplicate into the group: a resync + * seed's tool cards carry live progress (`outputLines` from + * `in_flight_turn.running_tools[].last_progress`) that the persisted copy + * lacks — without this, a mid-tool refresh blanks the card's latest output + * until the next progress frame. Never overwrite output a tool result + * already settled. + */ + function mergeVolatileExtras(g: Group, content: AppMessage['content']): void { + for (const c of content) { + if (c.type !== 'toolUse' || !c.outputLines?.length) continue; + const idx = g.tools.findIndex((t) => t.id === c.toolCallId); + if (idx === -1) continue; + const tool = g.tools[idx]!; + if (tool.output !== undefined) continue; + const updated: ToolCall = { ...tool, output: c.outputLines }; + g.tools[idx] = updated; + const blk = g.blocks.find((b) => b.kind === 'tool' && b.tool.id === c.toolCallId); + if (blk && blk.kind === 'tool') blk.tool = updated; + } + } + function resolveMediaUrl( c: AppMessage['content'][number], ): { url: string; kind: 'image' | 'video'; fileId?: string } | undefined { @@ -769,7 +843,7 @@ export function messagesToTurns( blocks: [], approval: undefined, approvalId: undefined, - seenSigs: new Set(), + foldedSigs: [], durationMs: msg.durationMs, }; } else if (pendingGroup !== null && pendingGroup.promptId === undefined && pid !== undefined) { @@ -781,10 +855,15 @@ export function messagesToTurns( // Drop an assistant message whose content was already folded into this group // (a duplicate streamed-vs-persisted copy sharing the promptId), so the turn - // doesn't render the same text + tools twice. - const sig = JSON.stringify(msg.content); - if (group.promptId !== undefined && group.seenSigs.has(sig)) continue; - group.seenSigs.add(sig); + // doesn't render the same text + tools twice. The duplicate can still carry + // volatile extras the persisted copy lacks (tool progress), so merge those + // into the existing cards before dropping it. + const sig = contentSig(msg.content); + if (group.promptId !== undefined && group.foldedSigs.some((folded) => covers(folded, sig))) { + mergeVolatileExtras(group, msg.content); + continue; + } + group.foldedSigs.push(sig); absorbContent(group, msg.content); } diff --git a/apps/kimi-web/test/agent-event-projector.test.ts b/apps/kimi-web/test/agent-event-projector.test.ts index 29be76f607..36bf5ec581 100644 --- a/apps/kimi-web/test/agent-event-projector.test.ts +++ b/apps/kimi-web/test/agent-event-projector.test.ts @@ -219,3 +219,80 @@ describe('session status single-sourcing', () => { ); }); }); + +describe('step-boundary delta alignment', () => { + it('resets stream offsets at step boundaries — a post-step delta ahead of local state signals a gap', () => { + const projector = createAgentProjector(); + projector.project('turn.started', { turnId: 1 }, 's1'); + projector.project('turn.step.started', { turnId: 1, step: 1 }, 's1'); + projector.project('assistant.delta', { turnId: 1, delta: 'step-one text' }, 's1', { offset: 0 }); + projector.project('turn.step.completed', { turnId: 1, step: 1 }, 's1'); + projector.project('turn.step.started', { turnId: 1, step: 2 }, 's1'); + + const events = projector.project('assistant.delta', { turnId: 1, delta: 'tail' }, 's1', { offset: 12 }); + expect(events).toContainEqual( + expect.objectContaining({ type: 'historyCompacted', reason: 'delta_gap' }), + ); + }); + + it('appends step-2 deltas to the fresh step message at step-relative offsets', () => { + const projector = createAgentProjector(); + projector.project('turn.started', { turnId: 1 }, 's1'); + projector.project('turn.step.started', { turnId: 1, step: 1 }, 's1'); + projector.project('assistant.delta', { turnId: 1, delta: 'step one' }, 's1', { offset: 0 }); + projector.project('turn.step.completed', { turnId: 1, step: 1 }, 's1'); + + const step2 = projector.project('turn.step.started', { turnId: 1, step: 2 }, 's1'); + const created = step2.find((e) => e.type === 'messageCreated'); + const msgId = (created as { message: { id: string } } | undefined)?.message.id; + expect(msgId).toBeDefined(); + + // Offset restarts at 0 for the new step and appends to ITS message. + const events = projector.project('assistant.delta', { turnId: 1, delta: 'step two' }, 's1', { offset: 0 }); + expect(events).toContainEqual( + expect.objectContaining({ + type: 'assistantDelta', + messageId: msgId, + delta: { text: 'step two' }, + }), + ); + }); + + it('seeds only the current step and aligns live deltas against the seeded length', () => { + const projector = createAgentProjector(); + const seeded = projector.seedInFlight('s1', { + turnId: 7, + promptId: 'pr_1', + thinkingText: 'step two thinking', + assistantText: 'step two partial', + runningTools: [{ toolCallId: 'tc_1', name: 'bash', args: { command: 'ls' } }], + }); + const created = seeded.find((e) => e.type === 'messageCreated'); + const message = (created as { message: { id: string; content: unknown[] } } | undefined)?.message; + expect(message).toBeDefined(); + + expect(message!.content).toEqual([ + { type: 'thinking', thinking: 'step two thinking' }, + { type: 'text', text: 'step two partial' }, + { type: 'toolUse', toolCallId: 'tc_1', toolName: 'bash', input: { command: 'ls' } }, + ]); + + const dup = projector.project('assistant.delta', { turnId: 7, delta: 'two part' }, 's1', { offset: 5 }); + expect(dup).toEqual([]); + + const cont = projector.project( + 'assistant.delta', + { turnId: 7, delta: ' continues' }, + 's1', + { offset: 'step two partial'.length }, + ); + expect(cont).toContainEqual( + expect.objectContaining({ + type: 'assistantDelta', + messageId: message!.id, + contentIndex: 3, + delta: { text: ' continues' }, + }), + ); + }); +}); diff --git a/apps/kimi-web/test/turn-logic.test.ts b/apps/kimi-web/test/turn-logic.test.ts index 6b379bdfe2..b7c6191d6d 100644 --- a/apps/kimi-web/test/turn-logic.test.ts +++ b/apps/kimi-web/test/turn-logic.test.ts @@ -336,6 +336,176 @@ describe('messagesToTurns', () => { }); }); +describe('messagesToTurns resync dedup', () => { + it('drops a resync-seeded copy that differs only by signature, progress, and part boundaries', () => { + const turns = messagesToTurns( + [ + // Persisted transcript copy: thinking carries the provider signature, + // text split at the model's part boundary, plain tool_use. + message( + 'a1', + 'assistant', + [ + { type: 'thinking', thinking: 'let me check', signature: 'sig-abc' }, + { type: 'text', text: 'I will ' }, + { type: 'text', text: 'run ls' }, + { type: 'toolUse', toolCallId: 'tool-1', toolName: 'bash', input: { command: 'ls' } }, + ], + { promptId: 'p1' }, + ), + // Resync seed from in_flight_turn: no signature, streams concatenated + // into single parts, tool card carrying live progress outputLines. + message( + 'seed', + 'assistant', + [ + { type: 'thinking', thinking: 'let me check' }, + { type: 'text', text: 'I will run ls' }, + { + type: 'toolUse', + toolCallId: 'tool-1', + toolName: 'bash', + input: { command: 'ls' }, + outputLines: ['total 8'], + }, + ], + { promptId: 'p1' }, + ), + ], + [], + undefined, + true, + ); + + expect(turns).toHaveLength(1); + expect(turns[0]?.thinking).toBe('let me check'); + expect(turns[0]?.text).toBe('I will \nrun ls'); + expect(turns[0]?.tools).toHaveLength(1); + // The seed's live progress survives the dedup — the persisted card had none. + expect(turns[0]?.tools?.[0]?.output).toEqual(['total 8']); + }); + + it('keeps the seeded message when the transcript has no copy of the current step yet', () => { + const turns = messagesToTurns( + [ + message('a1', 'assistant', [{ type: 'text', text: 'step one done' }], { promptId: 'p1' }), + message('seed', 'assistant', [{ type: 'text', text: 'step two streami' }], { + promptId: 'p1', + }), + ], + [], + undefined, + true, + ); + + expect(turns).toHaveLength(1); + expect(turns[0]?.text).toBe('step one done\nstep two streami'); + }); + + it('keeps a following step whose content differs', () => { + const turns = messagesToTurns( + [ + message( + 'a1', + 'assistant', + [ + { type: 'text', text: 'step one' }, + { type: 'toolUse', toolCallId: 'tool-1', toolName: 'bash', input: { command: 'ls' } }, + ], + { promptId: 'p1' }, + ), + message('a2', 'assistant', [{ type: 'text', text: 'step two' }], { promptId: 'p1' }), + ], + [], + undefined, + false, + ); + + expect(turns).toHaveLength(1); + expect(turns[0]?.text).toBe('step one\nstep two'); + }); + + it('does not overwrite a settled tool result with seed progress', () => { + const turns = messagesToTurns( + [ + message( + 'a1', + 'assistant', + [{ type: 'toolUse', toolCallId: 'tool-1', toolName: 'bash', input: { command: 'ls' } }], + { promptId: 'p1' }, + ), + message('t1', 'tool', [ + { type: 'toolResult', toolCallId: 'tool-1', output: 'real result' }, + ]), + message( + 'seed', + 'assistant', + [ + { + type: 'toolUse', + toolCallId: 'tool-1', + toolName: 'bash', + input: { command: 'ls' }, + outputLines: ['stale progress'], + }, + ], + { promptId: 'p1' }, + ), + ], + [], + undefined, + false, + ); + + expect(turns[0]?.tools).toHaveLength(1); + expect(turns[0]?.tools?.[0]?.output).toEqual(['real result']); + }); + + it('drops a seed whose finished parallel tools left running_tools (subset of the persisted message)', () => { + const turns = messagesToTurns( + [ + message( + 'a1', + 'assistant', + [ + { type: 'text', text: 'running two tools' }, + { type: 'toolUse', toolCallId: 'tool-a', toolName: 'bash', input: { command: 'a' } }, + { type: 'toolUse', toolCallId: 'tool-b', toolName: 'bash', input: { command: 'b' } }, + ], + { promptId: 'p1' }, + ), + message('t1', 'tool', [{ type: 'toolResult', toolCallId: 'tool-a', output: 'a done' }]), + // tool-a finished and left running_tools; the seed carries only tool-b. + message( + 'seed', + 'assistant', + [ + { type: 'text', text: 'running two tools' }, + { + type: 'toolUse', + toolCallId: 'tool-b', + toolName: 'bash', + input: { command: 'b' }, + outputLines: ['b progress'], + }, + ], + { promptId: 'p1' }, + ), + ], + [], + undefined, + true, + ); + + expect(turns).toHaveLength(1); + expect(turns[0]?.text).toBe('running two tools'); + expect(turns[0]?.tools).toMatchObject([ + { id: 'tool-a', status: 'ok', output: ['a done'] }, + { id: 'tool-b', status: 'running', output: ['b progress'] }, + ]); + }); +}); + describe('latestTodos', () => { it('returns the newest todo write and ignores later read-only queries', () => { expect( diff --git a/packages/kap-server/src/transport/ws/v1/inFlightTurnTracker.ts b/packages/kap-server/src/transport/ws/v1/inFlightTurnTracker.ts index 22c9742738..8fadc4bb65 100644 --- a/packages/kap-server/src/transport/ws/v1/inFlightTurnTracker.ts +++ b/packages/kap-server/src/transport/ws/v1/inFlightTurnTracker.ts @@ -8,9 +8,11 @@ * dispatch queue — keeping accumulated text, the journal watermark, and fan-out * order mutually consistent. * - * `apply()` returns the pre-append character offset for text-delta frames; the - * broadcast layer stamps it on the wire envelope so clients align live deltas - * against snapshot text exactly (skip duplicates, detect gaps). + * Text accumulation is step-relative: `assistantText` / `thinkingText` reset at + * every `turn.step.started` because completed steps already live in the snapshot + * transcript; running tools are kept (a call without `tool.result` still needs + * seeding). The stamped delta `offset` is thus the pre-append offset within the + * current step, and clients reset their alignment counters at step boundaries. * * Only main-agent activity is tracked: subagent deltas share the session id but * describe a different stream and would corrupt the accumulation. @@ -65,6 +67,14 @@ export class InFlightTurnTracker { this.bySession.delete(sessionId); return {}; } + case 'turn.step.started': { + // Prior steps' text is already in the transcript; keep running tools. + const turn = this.bySession.get(sessionId); + if (!turn || turn.turnId !== event.turnId) return {}; + turn.assistantText = ''; + turn.thinkingText = ''; + return {}; + } case 'assistant.delta': { const turn = this.bySession.get(sessionId); if (!turn || turn.turnId !== event.turnId) return {}; diff --git a/packages/kap-server/test/inFlightTurnTracker.test.ts b/packages/kap-server/test/inFlightTurnTracker.test.ts index cd0230b8e3..d361d4047d 100644 --- a/packages/kap-server/test/inFlightTurnTracker.test.ts +++ b/packages/kap-server/test/inFlightTurnTracker.test.ts @@ -77,4 +77,50 @@ describe('InFlightTurnTracker', () => { t.apply(SID, ev({ type: 'tool.result', turnId: 1, toolCallId: 'tc1' })); expect(t.get(SID)?.running_tools).toEqual([]); }); + + it('resets text accumulation at step boundaries (step-relative in-flight text)', () => { + const t = new InFlightTurnTracker(); + t.apply(SID, ev({ type: 'turn.started', turnId: 1 })); + t.apply(SID, ev({ type: 'turn.step.started', turnId: 1, step: 1 })); + t.apply(SID, ev({ type: 'thinking.delta', turnId: 1, delta: 'think-1' })); + t.apply(SID, ev({ type: 'assistant.delta', turnId: 1, delta: 'text-1' })); + t.apply(SID, ev({ type: 'turn.step.completed', turnId: 1, step: 1 })); + + t.apply(SID, ev({ type: 'turn.step.started', turnId: 1, step: 2 })); + t.apply(SID, ev({ type: 'assistant.delta', turnId: 1, delta: 'text-2' })); + + expect(t.get(SID)).toMatchObject({ assistant_text: 'text-2', thinking_text: '' }); + }); + + it('reports step-relative offsets that restart at 0 each step', () => { + const t = new InFlightTurnTracker(); + t.apply(SID, ev({ type: 'turn.started', turnId: 1 })); + t.apply(SID, ev({ type: 'turn.step.started', turnId: 1, step: 1 })); + expect(t.apply(SID, ev({ type: 'assistant.delta', turnId: 1, delta: 'ab' }))).toEqual({ offset: 0 }); + expect(t.apply(SID, ev({ type: 'assistant.delta', turnId: 1, delta: 'cd' }))).toEqual({ offset: 2 }); + + t.apply(SID, ev({ type: 'turn.step.started', turnId: 1, step: 2 })); + expect(t.apply(SID, ev({ type: 'assistant.delta', turnId: 1, delta: 'x' }))).toEqual({ offset: 0 }); + }); + + it('keeps running tools across step boundaries while resetting text', () => { + const t = new InFlightTurnTracker(); + t.apply(SID, ev({ type: 'turn.started', turnId: 1 })); + t.apply(SID, ev({ type: 'turn.step.started', turnId: 1, step: 1 })); + t.apply(SID, ev({ type: 'tool.call.started', turnId: 1, toolCallId: 'tc1', name: 'bash' })); + t.apply(SID, ev({ type: 'assistant.delta', turnId: 1, delta: 'text-1' })); + + t.apply(SID, ev({ type: 'turn.step.started', turnId: 1, step: 2 })); + + expect(t.get(SID)?.assistant_text).toBe(''); + expect(t.get(SID)?.running_tools).toEqual([{ tool_call_id: 'tc1', name: 'bash' }]); + }); + + it('ignores step boundaries for a mismatched turn', () => { + const t = new InFlightTurnTracker(); + t.apply(SID, ev({ type: 'turn.started', turnId: 1 })); + t.apply(SID, ev({ type: 'assistant.delta', turnId: 1, delta: 'keep' })); + t.apply(SID, ev({ type: 'turn.step.started', turnId: 99, step: 2 })); + expect(t.get(SID)?.assistant_text).toBe('keep'); + }); }); diff --git a/packages/protocol/src/rest/snapshot.ts b/packages/protocol/src/rest/snapshot.ts index 642772b695..1cc531cbeb 100644 --- a/packages/protocol/src/rest/snapshot.ts +++ b/packages/protocol/src/rest/snapshot.ts @@ -48,9 +48,9 @@ export type InFlightToolCall = z.infer; export const inFlightTurnSchema = z.object({ turn_id: z.number().int().nonnegative(), - /** Assistant text accumulated from `assistant.delta` so far. */ + /** Assistant text accumulated from `assistant.delta` in the current step (reset on `turn.step.started`; earlier steps are in `messages`). */ assistant_text: z.string(), - /** Thinking text accumulated from `thinking.delta` so far. */ + /** Thinking text accumulated from `thinking.delta` in the current step (reset on `turn.step.started`). */ thinking_text: z.string(), /** Tool calls started but without a `tool.result` yet. */ running_tools: z.array(inFlightToolCallSchema),