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/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/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),