From f1fad7222ccd3f66c1cae6c5b9c009230227cd2f Mon Sep 17 00:00:00 2001 From: qer Date: Thu, 25 Jun 2026 13:06:38 +0800 Subject: [PATCH] fix: reduce streaming stutter in the web chat (#1085) * fix(web): coalesce streaming token updates into one render per frame * fix(server): disable Nagle on WebSocket socket for lower streaming latency * fix(web): flush pending streaming deltas before re-subscribing * fix(web): flush pending streaming deltas before forgetting a session --- .changeset/server-ws-nodelay.md | 5 + .changeset/smooth-web-streaming.md | 5 + .../src/composables/client/eventBatcher.ts | 80 +++++++++ .../src/composables/useKimiWebClient.ts | 169 ++++++++++++------ apps/kimi-web/test/event-batcher.test.ts | 150 ++++++++++++++++ .../src/services/gateway/wsGatewayService.ts | 5 + 6 files changed, 359 insertions(+), 55 deletions(-) create mode 100644 .changeset/server-ws-nodelay.md create mode 100644 .changeset/smooth-web-streaming.md create mode 100644 apps/kimi-web/src/composables/client/eventBatcher.ts create mode 100644 apps/kimi-web/test/event-batcher.test.ts diff --git a/.changeset/server-ws-nodelay.md b/.changeset/server-ws-nodelay.md new file mode 100644 index 000000000..f46df8fa9 --- /dev/null +++ b/.changeset/server-ws-nodelay.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/server": patch +--- + +Reduce streaming latency by disabling Nagle's algorithm on WebSocket connections. diff --git a/.changeset/smooth-web-streaming.md b/.changeset/smooth-web-streaming.md new file mode 100644 index 000000000..bff9ee8fe --- /dev/null +++ b/.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. diff --git a/apps/kimi-web/src/composables/client/eventBatcher.ts b/apps/kimi-web/src/composables/client/eventBatcher.ts new file mode 100644 index 000000000..a42d02836 --- /dev/null +++ b/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 = new Set([ + '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 { + (item: T): void; + /** Synchronously drain any pending batchable items in arrival order. */ + flush(): void; +} + +export function createEventBatcher( + process: (item: T) => void, + isBatchable: (item: T) => boolean, + schedule: (cb: () => void) => number = defaultScheduleFrame, +): EventBatcher { + 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; + + enqueue.flush = drain; + + return enqueue; +} diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index 11b6b6f6a..d593f4bb3 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -18,6 +18,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'; @@ -28,6 +29,7 @@ import { useWorkspaceState } from './client/useWorkspaceState'; const appearance = useAppearance(); const notification = useNotification(); import type { + AppEvent, AppApprovalRequest, AppConfig, AppGoal, @@ -463,6 +465,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]; @@ -659,6 +668,88 @@ function applyEvent(event: ReturnType, 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( + ({ appEvent, meta }) => processEvent(appEvent, meta), + ({ appEvent }) => isRenderEvent(appEvent), +); + // --------------------------------------------------------------------------- // WS subscription (lazy, only when a session is selected) // --------------------------------------------------------------------------- @@ -686,64 +777,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; @@ -961,6 +1008,13 @@ async function syncSessionFromSnapshot(sessionId: string): Promise ({ ...snap.session, model: @@ -1019,6 +1073,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 }); diff --git a/apps/kimi-web/test/event-batcher.test.ts b/apps/kimi-web/test/event-batcher.test.ts new file mode 100644 index 000000000..09360e6db --- /dev/null +++ b/apps/kimi-web/test/event-batcher.test.ts @@ -0,0 +1,150 @@ +// apps/kimi-web/test/event-batcher.test.ts +// Unit tests for the streaming-event coalescing logic. +// +// These verify the batcher's behaviour (coalesce + preserve order + immediate +// passthrough for non-batchable items). They deliberately do NOT try to assert +// "Vue renders once" — that is a property of Vue's scheduler and is covered by +// manual perf verification, not by a unit test. + +import { describe, expect, it } from 'vitest'; +import { createEventBatcher, isRenderEvent } from '../src/composables/client/eventBatcher'; +import type { AppEvent } from '../src/api/types'; + +interface FakeSchedule { + schedule: (cb: () => void) => number; + calls: () => number; + flush: () => void; +} + +// A synchronous, manually-triggered scheduler. Stores the most recent callback; +// `flush()` runs it. Lets tests drive the batcher without real rAF / timers. +function fakeSchedule(): FakeSchedule { + let cb: (() => void) | null = null; + let count = 0; + return { + schedule(fn) { + count += 1; + cb = fn; + return count; + }, + calls: () => count, + flush() { + const fn = cb; + cb = null; + fn?.(); + }, + }; +} + +describe('createEventBatcher', () => { + it('coalesces consecutive batchable items into one scheduled flush, in order', () => { + const processed: string[] = []; + const f = fakeSchedule(); + const enqueue = createEventBatcher((s) => processed.push(s), (s) => s.startsWith('d'), f.schedule); + + enqueue('d1'); + enqueue('d2'); + enqueue('d3'); + + expect(processed).toEqual([]); // nothing processed yet + expect(f.calls()).toBe(1); // scheduled exactly once + + f.flush(); + expect(processed).toEqual(['d1', 'd2', 'd3']); + }); + + it('applies a non-batchable item immediately when the queue is empty', () => { + const processed: string[] = []; + const f = fakeSchedule(); + const enqueue = createEventBatcher((s) => processed.push(s), (s) => s.startsWith('d'), f.schedule); + + enqueue('X'); + + expect(processed).toEqual(['X']); + expect(f.calls()).toBe(0); // never scheduled + }); + + it('drains pending batchables before applying an immediate item', () => { + const processed: string[] = []; + const f = fakeSchedule(); + const enqueue = createEventBatcher((s) => processed.push(s), (s) => s.startsWith('d'), f.schedule); + + enqueue('d1'); + enqueue('d2'); + enqueue('X'); // immediate → must flush d1, d2 first + + expect(processed).toEqual(['d1', 'd2', 'X']); + + // The rAF scheduled for d1 is now stale; firing it must be a harmless no-op. + f.flush(); + expect(processed).toEqual(['d1', 'd2', 'X']); + }); + + it('preserves arrival order across mixed batchable and immediate items', () => { + const processed: string[] = []; + const f = fakeSchedule(); + const enqueue = createEventBatcher((s) => processed.push(s), (s) => s.startsWith('d'), f.schedule); + + enqueue('d1'); // queued + enqueue('d2'); // queued + enqueue('A'); // immediate → drains d1, d2, then A + enqueue('d3'); // queued again + f.flush(); // drains d3 + + expect(processed).toEqual(['d1', 'd2', 'A', 'd3']); + }); + + it('reschedules after a flush when new batchable items arrive', () => { + const processed: string[] = []; + const f = fakeSchedule(); + const enqueue = createEventBatcher((s) => processed.push(s), (s) => s.startsWith('d'), f.schedule); + + enqueue('d1'); + f.flush(); + expect(processed).toEqual(['d1']); + + enqueue('d2'); + expect(f.calls()).toBe(2); // scheduled a second time + + f.flush(); + expect(processed).toEqual(['d1', 'd2']); + }); + + it('flush() drains pending batchables synchronously without the scheduler', () => { + const processed: string[] = []; + const f = fakeSchedule(); + const enqueue = createEventBatcher((s) => processed.push(s), (s) => s.startsWith('d'), f.schedule); + + enqueue('d1'); + enqueue('d2'); + expect(processed).toEqual([]); + + enqueue.flush(); // synchronous drain, no scheduler callback needed + expect(processed).toEqual(['d1', 'd2']); + }); + + it('flush() on an empty queue is a no-op', () => { + const processed: string[] = []; + const f = fakeSchedule(); + const enqueue = createEventBatcher((s) => processed.push(s), (s) => s.startsWith('d'), f.schedule); + + enqueue.flush(); + expect(processed).toEqual([]); + }); +}); + +describe('isRenderEvent', () => { + it.each(['assistantDelta', 'agentDelta', 'toolOutput', 'taskProgress'])( + 'treats %s as batchable', + (type) => { + expect(isRenderEvent({ type } as AppEvent)).toBe(true); + }, + ); + + it.each(['messageCreated', 'messageUpdated', 'sessionStatusChanged', 'approvalRequested', 'configChanged'])( + 'treats %s as immediate', + (type) => { + expect(isRenderEvent({ type } as AppEvent)).toBe(false); + }, + ); +}); diff --git a/packages/server/src/services/gateway/wsGatewayService.ts b/packages/server/src/services/gateway/wsGatewayService.ts index 0abbfc38e..7c06b98f0 100644 --- a/packages/server/src/services/gateway/wsGatewayService.ts +++ b/packages/server/src/services/gateway/wsGatewayService.ts @@ -63,6 +63,11 @@ export class WSGateway extends Disposable implements IWSGateway { socket.destroy(); return; } + // Disable Nagle's algorithm: streaming chat sends many small frames (one per + // token delta), and Nagle + the client's delayed ACK can bunch them into + // ~40 ms clusters, making the stream look stuttery. Trade a little bandwidth + // for lower latency. + socket.setNoDelay(true); this.wss.handleUpgrade(req, socket, head, (ws) => this.onConnect(ws, req)); }