fix(web): make mid-turn delta offsets step-relative

.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.

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;

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' },
+      }),
+    );
+  });
+});

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 {};

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');
+  });
 });

packages/protocol/src/rest/snapshot.ts

@@ -48,9 +48,9 @@ export type InFlightToolCall = z.infer<typeof inFlightToolCallSchema>;
 
 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),