fix: reduce streaming stutter in the web chat

.changeset/server-ws-nodelay.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/server": patch
+---
+
+Reduce streaming latency by disabling Nagle's algorithm on WebSocket connections.

.changeset/smooth-web-streaming.md

@@ -0,0 +1,5 @@
+---
+"@moonshot-ai/kimi-web": patch
+---
+
+Fix stuttery streaming in the web chat by coalescing rapid token updates into a single render per frame.

apps/kimi-web/src/composables/client/eventBatcher.ts

@@ -0,0 +1,80 @@
+// apps/kimi-web/src/composables/client/eventBatcher.ts
+// Coalesce high-frequency streaming events onto the next animation frame.
+//
+// Pure logic (no Vue, no DOM) so it is unit-testable in isolation. See
+// useKimiWebClient.ts for where it is wired into the WS event pipeline.
+
+import type { AppEvent } from '../../api/types';
+
+// Events that merely append a chunk to something already streaming. They can
+// arrive dozens to hundreds of times per second, so they are worth coalescing.
+const RENDER_EVENT_TYPES: ReadonlySet<AppEvent['type']> = new Set<AppEvent['type']>([
+  'assistantDelta',
+  'agentDelta',
+  'toolOutput',
+  'taskProgress',
+]);
+
+/** True for high-frequency render-only events that are safe to delay to the
+    next animation frame. Everything else (lifecycle / control-flow) must apply
+    immediately so turn-end cleanup etc. is not delayed by a throttled rAF. */
+export function isRenderEvent(appEvent: AppEvent): boolean {
+  return RENDER_EVENT_TYPES.has(appEvent.type);
+}
+
+function defaultScheduleFrame(cb: () => void): number {
+  return typeof requestAnimationFrame === 'function'
+    ? requestAnimationFrame(cb)
+    : (setTimeout(cb, 16) as unknown as number);
+}
+
+/**
+ * Coalesce batchable items onto a single scheduled callback, while applying
+ * non-batchable items immediately.
+ *
+ * A non-batchable item first drains any pending batchable items (in arrival
+ * order) so overall ordering is preserved — a lifecycle event never overtakes
+ * the deltas that arrived before it.
+ *
+ * The returned handle is itself callable (enqueue) and also exposes `flush()`
+ * to synchronously drain pending batchable items. Callers that replace state
+ * authoritatively (e.g. applying a server snapshot) must `flush()` first so
+ * stale queued deltas are not applied on top of the new state.
+ */
+export interface EventBatcher<T> {
+  (item: T): void;
+  /** Synchronously drain any pending batchable items in arrival order. */
+  flush(): void;
+}
+
+export function createEventBatcher<T>(
+  process: (item: T) => void,
+  isBatchable: (item: T) => boolean,
+  schedule: (cb: () => void) => number = defaultScheduleFrame,
+): EventBatcher<T> {
+  let pending: T[] = [];
+  let handle: number | null = null;
+
+  const drain = (): void => {
+    handle = null;
+    if (pending.length === 0) return;
+    const batch = pending;
+    pending = [];
+    for (const item of batch) process(item);
+  };
+
+  const enqueue = ((item: T) => {
+    if (isBatchable(item)) {
+      pending.push(item);
+      if (handle === null) handle = schedule(drain);
+      return;
+    }
+    // Immediate item: flush pending batchables first to preserve order.
+    drain();
+    process(item);
+  }) as EventBatcher<T>;
+
+  enqueue.flush = drain;
+
+  return enqueue;
+}

apps/kimi-web/src/composables/useKimiWebClient.ts

@@ -17,6 +17,7 @@ import {
   saveWorkspaceOrder,
   STORAGE_KEYS,
 } from '../lib/storage';
+import { createEventBatcher, isRenderEvent } from './client/eventBatcher';
 import { useAppearance } from './client/useAppearance';
 import { useNotification } from './client/useNotification';
 import { useTaskPoller } from './client/useTaskPoller';
@@ -27,6 +28,7 @@ import { useWorkspaceState } from './client/useWorkspaceState';
 const appearance = useAppearance();
 const notification = useNotification();
 import type {
+  AppEvent,
   AppApprovalRequest,
   AppConfig,
   AppGoal,
@@ -462,6 +464,13 @@ function forgetSession(sessionId: string): void {
   // buffered event for this id would otherwise be reduced and recreate the very
   // per-session maps we are about to delete.
   eventConn?.unsubscribe(sessionId);
+  // Drain the streaming-event batcher too. unsubscribe() stops future server
+  // frames, but events already queued for the next animation frame would
+  // otherwise survive and be reduced AFTER the maps below are cleared —
+  // recreating entries like messagesBySession[id] and lastSeqBySession[id].
+  // That would make hasLoadedMessages() treat the stale empty cache as
+  // authoritative and skip the next snapshot fetch for this id.
+  enqueueEvent.flush();
   removeSession(sessionId);
   removeSessionMessages(sessionId);
   delete rawState.approvalsBySession[sessionId];
@@ -658,6 +667,88 @@ function applyEvent(event: ReturnType<typeof toAppEvent>, sessionId: string, seq
   }
 }
 
+// ---------------------------------------------------------------------------
+// Streaming event batching
+// ---------------------------------------------------------------------------
+//
+// High-frequency "append a chunk" events (assistant/agent deltas, tool/task
+// output) can arrive dozens to hundreds of times per second. Applying each one
+// synchronously triggers a full Vue re-render per event, which saturates the
+// main thread and makes the stream look janky (see messagesToTurns / Markdown).
+//
+// We coalesce those render-only events onto the next animation frame so Vue
+// commits a single render per frame. Lifecycle / control-flow events
+// (sessionStatusChanged, messageCreated, approval*, question*, ...) are applied
+// immediately: they are infrequent, and some (e.g. sessionStatusChanged idle)
+// drive turn-end cleanup that must not be delayed by a throttled rAF in a
+// background tab. Ordering is preserved by draining any pending render events
+// before applying an immediate event.
+
+type PendingEvent = { appEvent: AppEvent; meta: { sessionId: string; seq: number } };
+
+function processEvent(appEvent: AppEvent, meta: { sessionId: string; seq: number }): void {
+  // meta carries wire-level seq/sessionId so the reducer can advance
+  // lastSeqBySession[sessionId] = seq. Compaction completion appends a
+  // persistent divider marker in the reducer (TUI parity: the scrollback
+  // is kept, only a marker line records the compaction).
+  applyEvent(appEvent, meta.sessionId, meta.seq);
+
+  const sideTarget = sideChat.sideChatTargetBySession.value[meta.sessionId];
+  if (sideTarget) {
+    const { agentId } = sideTarget;
+    const parentId = meta.sessionId;
+    if (appEvent.type === 'agentDelta' && appEvent.agentId === agentId) {
+      if (appEvent.delta.text) {
+        sideChat.appendSideChatAssistantText(agentId, parentId, appEvent.delta.text);
+      }
+    } else if (appEvent.type === 'agentTurnEnded' && appEvent.agentId === agentId) {
+      sideChat.finishSideChatAgent(agentId, parentId);
+    } else if (appEvent.type === 'taskProgress' && appEvent.taskId === agentId) {
+      sideChat.appendSideChatAssistantText(agentId, parentId, appEvent.outputChunk);
+    } else if (appEvent.type === 'taskCompleted' && appEvent.taskId === agentId) {
+      sideChat.finishSideChatAgent(agentId, parentId, appEvent.outputPreview);
+    }
+  }
+
+  // The daemon's prompt.submitted event is projected as a user messageCreated
+  // carrying the real prompt_id. When the HTTP submit response is lost
+  // (timeout / network error) this is the fallback that lets Stop work.
+  if (
+    appEvent.type === 'messageCreated' &&
+    appEvent.message.role === 'user' &&
+    appEvent.message.promptId !== undefined
+  ) {
+    const sid = appEvent.message.sessionId;
+    if (rawState.promptIdBySession[sid] !== appEvent.message.promptId) {
+      rawState.promptIdBySession = {
+        ...rawState.promptIdBySession,
+        [sid]: appEvent.message.promptId,
+      };
+    }
+  }
+
+  if (appEvent.type === 'assistantDelta' && meta.sessionId === rawState.activeSessionId) {
+    appearance.recordMoonDelta((appEvent.delta.text?.length ?? 0) + (appEvent.delta.thinking?.length ?? 0));
+  }
+
+  // Turn-end cleanup for the session the event belongs to — including
+  // sessions running in the background (see onSessionIdle).
+  // Turn-end: both 'idle' and 'aborted' mean the prompt is no longer in
+  // flight, so both must flush in-flight/queued state. (Awaiting-* is still
+  // in flight — it's waiting on the user — and must NOT flush.)
+  if (
+    appEvent.type === 'sessionStatusChanged' &&
+    (appEvent.status === 'idle' || appEvent.status === 'aborted')
+  ) {
+    onSessionIdle(appEvent.sessionId, appEvent.status);
+  }
+}
+
+const enqueueEvent = createEventBatcher<PendingEvent>(
+  ({ appEvent, meta }) => processEvent(appEvent, meta),
+  ({ appEvent }) => isRenderEvent(appEvent),
+);
+
 // ---------------------------------------------------------------------------
 // WS subscription (lazy, only when a session is selected)
 // ---------------------------------------------------------------------------
@@ -685,64 +776,20 @@ function connectEventsIfNeeded(): void {
         return;
       }
 
-      // meta carries wire-level seq/sessionId so the reducer can advance
-      // lastSeqBySession[sessionId] = seq. Compaction completion appends a
-      // persistent divider marker in the reducer (TUI parity: the scrollback
-      // is kept, only a marker line records the compaction).
-      applyEvent(appEvent, meta.sessionId, meta.seq);
-
-      const sideTarget = sideChat.sideChatTargetBySession.value[meta.sessionId];
-      if (sideTarget) {
-        const { agentId } = sideTarget;
-        const parentId = meta.sessionId;
-        if (appEvent.type === 'agentDelta' && appEvent.agentId === agentId) {
-          if (appEvent.delta.text) {
-            sideChat.appendSideChatAssistantText(agentId, parentId, appEvent.delta.text);
-          }
-        } else if (appEvent.type === 'agentTurnEnded' && appEvent.agentId === agentId) {
-          sideChat.finishSideChatAgent(agentId, parentId);
-        } else if (appEvent.type === 'taskProgress' && appEvent.taskId === agentId) {
-          sideChat.appendSideChatAssistantText(agentId, parentId, appEvent.outputChunk);
-        } else if (appEvent.type === 'taskCompleted' && appEvent.taskId === agentId) {
-          sideChat.finishSideChatAgent(agentId, parentId, appEvent.outputPreview);
-        }
-      }
-
-      // The daemon's prompt.submitted event is projected as a user messageCreated
-      // carrying the real prompt_id. When the HTTP submit response is lost
-      // (timeout / network error) this is the fallback that lets Stop work.
-      if (
-        appEvent.type === 'messageCreated' &&
-        appEvent.message.role === 'user' &&
-        appEvent.message.promptId !== undefined
-      ) {
-        const sid = appEvent.message.sessionId;
-        if (rawState.promptIdBySession[sid] !== appEvent.message.promptId) {
-          rawState.promptIdBySession = {
-            ...rawState.promptIdBySession,
-            [sid]: appEvent.message.promptId,
-          };
-        }
-      }
-
-      if (appEvent.type === 'assistantDelta' && meta.sessionId === rawState.activeSessionId) {
-        appearance.recordMoonDelta((appEvent.delta.text?.length ?? 0) + (appEvent.delta.thinking?.length ?? 0));
-      }
-
-      // Turn-end cleanup for the session the event belongs to — including
-      // sessions running in the background (see onSessionIdle).
-      // Turn-end: both 'idle' and 'aborted' mean the prompt is no longer in
-      // flight, so both must flush in-flight/queued state. (Awaiting-* is still
-      // in flight — it's waiting on the user — and must NOT flush.)
-      if (
-        appEvent.type === 'sessionStatusChanged' &&
-        (appEvent.status === 'idle' || appEvent.status === 'aborted')
-      ) {
-        onSessionIdle(appEvent.sessionId, appEvent.status);
-      }
+      // Coalesce high-frequency render events onto the next animation frame;
+      // everything else is applied immediately. See createEventBatcher /
+      // processEvent above.
+      enqueueEvent({ appEvent, meta });
     },
 
     onResync(sessionId: string, currentSeq: number, epoch?: string) {
+      // Flush streaming deltas already queued so they render on the
+      // pre-snapshot state (the snapshot is authoritative and will overwrite
+      // them). Stragglers that arrive during the snapshot fetch are drained
+      // again right before the snapshot write inside syncSessionFromSnapshot,
+      // so they are applied to the pre-snapshot array too rather than on top
+      // of the fresh snapshot (which would duplicate text / tool output).
+      enqueueEvent.flush();
       // The server-announced cursor is only a hint; the snapshot fetch
       // returns the authoritative {asOfSeq, epoch} and re-subscribes.
       if (epoch !== undefined) epochBySession[sessionId] = epoch;
@@ -960,6 +1007,13 @@ async function syncSessionFromSnapshot(sessionId: string): Promise<SyncSessionRe
     const api = getKimiWebApi();
     const snap = await api.getSessionSnapshot(sessionId);
 
+    // Drain any queued streaming deltas before the snapshot replaces
+    // messagesBySession[sessionId]. The snapshot is authoritative (it already
+    // contains everything up to asOfSeq); applying stale queued deltas on top
+    // of it would duplicate text / tool output. Flushing here applies them to
+    // the pre-snapshot array, which the snapshot then overwrites.
+    enqueueEvent.flush();
+
     updateSession(sessionId, (s) => ({
       ...snap.session,
       model:
@@ -1016,6 +1070,11 @@ function hasLoadedMessages(sessionId: string): boolean {
 function subscribeToSessionEvents(sessionId: string): void {
   connectEventsIfNeeded();
   if (eventConn) {
+    // Apply any queued streaming deltas before re-subscribing so the transcript
+    // is current. (These deltas are volatile — never replayed by the server and
+    // they don't advance lastSeqBySession — but flushing here is cheap and
+    // future-proofs the cursor if the batching set ever changes.)
+    enqueueEvent.flush();
     const seq = rawState.lastSeqBySession[sessionId] ?? 0;
     const epoch = epochBySession[sessionId];
     eventConn.subscribe(sessionId, { seq, epoch });