From 6eadb62eb769d6dd4c9bc64ea345b6e18007f0a4 Mon Sep 17 00:00:00 2001 From: Jesse_Chen Date: Tue, 1 Sep 2026 21:27:15 +0000 Subject: [PATCH] perf(chat): coalesce stream events per frame and pace text release Every NDJSON event used to commit its own React update and re-parse the whole partial answer through react-markdown, so long replies grew quadratically slower. Stream events now land in a frame buffer that flushes at most once per animation frame, releases answer and thinking text at a steady pace with a twelve-frame catch-up, and settles synchronously on completion, failure and abort. Streaming markdown is split at the last completed block so only the tail is re-parsed each frame. Applied to both the consultation hook and the rectification chat. BUG-473 Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_01JUei7K13cYxLHE3Axe4A45 --- docs/BUG_HISTORY.md | 16 ++ .../src/components/chat-message-content.tsx | 57 ++++- frontend/src/components/chat-message-row.tsx | 1 + .../components/rectification-agentic-chat.tsx | 106 +++++---- frontend/src/hooks/use-consultation-run.ts | 79 ++++--- frontend/src/lib/chat-markdown-split.ts | 69 ++++++ frontend/src/lib/stream-frame-buffer.ts | 212 ++++++++++++++++++ frontend/tests/chat-markdown-split.test.ts | 74 ++++++ .../tests/home-streaming-render-split.test.ts | 47 +++- frontend/tests/stream-frame-buffer.test.ts | 210 +++++++++++++++++ 10 files changed, 778 insertions(+), 93 deletions(-) create mode 100644 frontend/src/lib/chat-markdown-split.ts create mode 100644 frontend/src/lib/stream-frame-buffer.ts create mode 100644 frontend/tests/chat-markdown-split.test.ts create mode 100644 frontend/tests/stream-frame-buffer.test.ts diff --git a/docs/BUG_HISTORY.md b/docs/BUG_HISTORY.md index 13f0ec14..64b141c6 100644 --- a/docs/BUG_HISTORY.md +++ b/docs/BUG_HISTORY.md @@ -7266,3 +7266,19 @@ - 相关记录：BUG-453、BUG-463 - 复发自：BUG-463（idle 不再追问，拒答路径与 occupation blocking 仍挡住采用卡） - 修复版本：待发布 + +## BUG-473 | 流式回答每个网络事件都全量重渲染，长回答越写越卡 + +- 状态：resolved +- 首次发现：2026-09-01 +- 最近更新：2026-09-01 +- 影响面：`use-consultation-run.ts` NDJSON 回调、`rectification-agentic-chat.tsx` `send` 循环、`chat-message-content.tsx` +- 用户现象：回答越长越卡，文字一坨一坨蹦出来而不是流出来；思考步骤更新时整条消息跟着抖。 +- 触发条件：任何一次流式咨询或校正回合，回答超过一两千字后明显。 +- 根因：服务端每个模型 text-delta 立即发一条 `answer.delta`，客户端每条事件各自 `setStreamingReply` / `setMessages`；每次提交都对整段部分回答跑 `parseAgentReply`、`applyThinkingSectionProgress`、`splitSpokenAnswerAndTechniqueAudit` 和 `react-markdown` 全量 parse，随回答长度呈平方级增长。 +- 修复：新增 `lib/stream-frame-buffer.ts`，所有流式事件先落到累加器，`requestAnimationFrame` 合并成每帧最多一次提交；正文与思考文本按 `max(2, ceil(积压/12))` 每帧匀速释放，突发积压约十二帧追平，页面隐藏时退化为 250ms 定时并一次放完，`run.completed`/失败/中止时同步冲干净。`chat-message-content.tsx` 在流式时把已完成段落（最后一个空行之前，不切进代码围栏、列表项之间或表格）交给按内容记忆的前缀组件，只有尾块每帧重 parse；结算后仍整篇一次 parse。 +- 验证：`tests/stream-frame-buffer.test.ts`（释放公式、200 token/50 帧只提交 50 次、突发 12 帧追平、隐藏退化、reset/dispose）、`tests/chat-markdown-split.test.ts`（切分规则、前缀稳定、前缀与尾块分开 parse）、`tests/home-streaming-render-split.test.ts` 新增按帧驱动的渲染次数上限。 +- 防复发：流式状态必须经 `stream-frame-buffer` 提交，不得在事件回调里直接 `setState`；流式期间的 Markdown 渲染必须走前缀/尾块切分。 +- 相关记录：BUG-474 +- 复发自：无 +- 修复版本：待发布 diff --git a/frontend/src/components/chat-message-content.tsx b/frontend/src/components/chat-message-content.tsx index bec2ea9d..24cafa41 100644 --- a/frontend/src/components/chat-message-content.tsx +++ b/frontend/src/components/chat-message-content.tsx @@ -1,10 +1,11 @@ "use client"; -import { useEffect, useState, type ReactNode } from "react"; +import { memo, useEffect, useState, type ReactNode } from "react"; import { prefetchOnIdle } from "@/components/chat-chunk-prefetch"; import { plainParagraphs } from "@/components/chat-message-paragraphs"; import { TechniqueAuditDisclosure } from "@/components/technique-audit-disclosure"; +import { splitStableMarkdown } from "@/lib/chat-markdown-split"; import type { TechniqueAuditRow } from "@/lib/consultation-agent-events"; import { resolveTechniqueAuditRows, @@ -41,14 +42,56 @@ function useMarkdownRenderer() { return renderer; } +function renderProse(text: string, renderMarkdown: MarkdownRenderer | undefined): ReactNode { + if (!text) return null; + return renderMarkdown + ? renderMarkdown(text) + : (plainParagraphs(text) ?? []).map((paragraph, index) => ( +

{paragraph}

+ )); +} + +/** + * The completed part of a streaming answer. `memo` keeps React from calling the + * markdown parser again while `text` is unchanged, so a frame that only grew + * the tail costs one small parse instead of one over the whole answer. + */ +const StableMarkdownPrefix = memo(function StableMarkdownPrefix({ + text, + renderMarkdown, +}: Readonly<{ + text: string; + renderMarkdown: MarkdownRenderer | undefined; +}>) { + return <>{renderProse(text, renderMarkdown)}; +}); + +export function StreamingMarkdown({ + text, + renderMarkdown, +}: Readonly<{ + text: string; + renderMarkdown: MarkdownRenderer | undefined; +}>) { + const split = splitStableMarkdown(text); + return ( + <> + {split.stable ? : null} + {renderProse(split.tail, renderMarkdown)} + + ); +} + export function ChatMessageContent({ text, auditRows, vargaSentence, + streaming = false, }: { text: string; auditRows?: readonly TechniqueAuditRow[]; vargaSentence?: string | null; + streaming?: boolean; }) { const renderMarkdown = useMarkdownRenderer(); const split = splitSpokenAnswerAndTechniqueAudit(text); @@ -60,11 +103,13 @@ export function ChatMessageContent({
{spoken ? (
- {renderMarkdown - ? renderMarkdown(spoken) - : (plainParagraphs(spoken) ?? []).map((paragraph, index) => ( -

{paragraph}

- ))} + {streaming + ? + : renderMarkdown + ? renderMarkdown(spoken) + : (plainParagraphs(spoken) ?? []).map((paragraph, index) => ( +

{paragraph}

+ ))}
) : null} {vargaSentence ?

{vargaSentence}

: null} diff --git a/frontend/src/components/chat-message-row.tsx b/frontend/src/components/chat-message-row.tsx index 03d0a1c2..401cc4cd 100644 --- a/frontend/src/components/chat-message-row.tsx +++ b/frontend/src/components/chat-message-row.tsx @@ -116,6 +116,7 @@ export function ChatMessageRow({ text={message.text} auditRows={message.agentExecutionReceipt?.techniqueAuditTable} vargaSentence={showThinkingPanel ? null : vargaSentence} + streaming={message.state !== "settled"} /> ) : null; diff --git a/frontend/src/components/rectification-agentic-chat.tsx b/frontend/src/components/rectification-agentic-chat.tsx index 1872245f..563b70f0 100644 --- a/frontend/src/components/rectification-agentic-chat.tsx +++ b/frontend/src/components/rectification-agentic-chat.tsx @@ -4,7 +4,8 @@ import { ArrowUp, Square } from "lucide-react"; import { useCallback, useEffect, useId, useLayoutEffect, useRef, useState } from "react"; import { createPortal } from "react-dom"; import { parseAgentReply } from "@/lib/agent-reply"; -import { nextActivityView, type ChatMessage, type ChatMessageView } from "@/lib/chat-message-view"; +import { nextActivityView, type AgentActivityView, type ChatMessage, type ChatMessageView } from "@/lib/chat-message-view"; +import { createStreamFrameBuffer } from "@/lib/stream-frame-buffer"; import { completeActivityTrace, completeActivityTraceStep, @@ -494,6 +495,30 @@ export function RectificationAgenticChat(props: RectificationAgenticChatProps) { let activityReceiptState = createRectificationActivityReceiptState(); let completedReceipt = receiptFromRectificationActivityState(activityReceiptState); let completedTurnId: string | undefined; + let currentActivity: AgentActivityView | undefined = { + phase: "evidence-validation", + label: "正在处理…", + startedAt: Date.now(), + }; + // Every stream event lands in `frames`; it commits at most once per animation + // frame and releases text at a steady pace. The loop below never calls + // setMessages for a live turn directly except on `attempt.reset`. + const frames = createStreamFrameBuffer({ + initialMeta: null, + flush: (frame) => { + const text = frame.answer; + setMessages((current) => current.map((message) => message.renderKey === assistantRenderKey + ? { + ...message, + text, + activityTrace, + completedReceipt, + state: text.trim() ? "streaming" : "thinking", + activity: currentActivity, + } + : message)); + }, + }); const abortController = new AbortController(); runAbort.current = abortController; try { @@ -569,35 +594,29 @@ export function RectificationAgenticChat(props: RectificationAgenticChatProps) { if (event.type === "answer.delta" && typeof event.text === "string") { raw = event.replace === true ? event.text : raw + event.text; activityTrace = freezeLiveThink(activityTrace); - setMessages((current) => current.map((message) => message.renderKey === assistantRenderKey - ? { - ...message, - text: raw, - activityTrace, - state: raw.trim() ? "streaming" : "thinking", - activity: nextActivityView(message.activity, { - phase: "answer-composition", - label: "正在组织回答…", - }), - } - : message)); + currentActivity = nextActivityView(currentActivity, { + phase: "answer-composition", + label: "正在组织回答…", + }); + frames.setAnswer(raw); } else if (event.type === "activity.changed" && isPublicRectificationActivity(event.activity)) { const activity = event.activity; - setMessages((current) => current.map((message) => message.renderKey === assistantRenderKey - ? { - ...message, - activity: nextActivityView(message.activity, { - phase: "evidence-validation", - label: RECTIFICATION_ACTIVITY_PROGRESS_LABELS[activity], - }), - } - : message)); + currentActivity = nextActivityView(currentActivity, { + phase: "evidence-validation", + label: RECTIFICATION_ACTIVITY_PROGRESS_LABELS[activity], + }); + frames.touch(); } else if (event.type === "attempt.reset") { raw = ""; activityTrace = emptyActivityTrace(); activityReceiptState = createRectificationActivityReceiptState(); completedReceipt = receiptFromRectificationActivityState(activityReceiptState); completedTurnId = undefined; + currentActivity = nextActivityView(undefined, { + phase: "evidence-validation", + label: "正在处理…", + }); + frames.reset(); setMessages((current) => current.map((message) => message.renderKey === assistantRenderKey ? { ...message, @@ -608,10 +627,7 @@ export function RectificationAgenticChat(props: RectificationAgenticChatProps) { completedReceipt: undefined, failed: false, turnId: undefined, - activity: nextActivityView(undefined, { - phase: "evidence-validation", - label: "正在处理…", - }), + activity: currentActivity, } : message)); } else if (event.type === "run.failed") { @@ -634,17 +650,12 @@ export function RectificationAgenticChat(props: RectificationAgenticChatProps) { tool, RECTIFICATION_TOOL_PROGRESS_LABELS[tool], ); - setMessages((current) => current.map((message) => message.renderKey === assistantRenderKey - ? { - ...message, - activityTrace, - activity: nextActivityView(message.activity, { - phase: rectificationToolActivityPhase(tool), - label: RECTIFICATION_TOOL_PROGRESS_LABELS[tool], - completedTrail: rectificationCompletedTrail(activityReceiptState.completedSteps), - }), - } - : message)); + currentActivity = nextActivityView(currentActivity, { + phase: rectificationToolActivityPhase(tool), + label: RECTIFICATION_TOOL_PROGRESS_LABELS[tool], + completedTrail: rectificationCompletedTrail(activityReceiptState.completedSteps), + }); + frames.touch(); continue; } if (event.status !== "completed" && event.status !== "failed") continue; @@ -661,21 +672,16 @@ export function RectificationAgenticChat(props: RectificationAgenticChatProps) { : [], }); completedReceipt = receiptFromRectificationActivityState(activityReceiptState); - setMessages((current) => current.map((message) => message.renderKey === assistantRenderKey - ? { - ...message, - activityTrace, - completedReceipt, - activity: nextActivityView(message.activity, { - phase: rectificationToolActivityPhase(tool), - label: RECTIFICATION_TOOL_DONE_LABELS[tool], - completedTrail: rectificationCompletedTrail(activityReceiptState.completedSteps), - }), - } - : message)); + currentActivity = nextActivityView(currentActivity, { + phase: rectificationToolActivityPhase(tool), + label: RECTIFICATION_TOOL_DONE_LABELS[tool], + completedTrail: rectificationCompletedTrail(activityReceiptState.completedSteps), + }); + frames.touch(); } } } + frames.settle(); const parsed = completed && !streamFailed ? parseAgentReply(raw) : { text: "", title: undefined }; const succeeded = completed && !streamFailed && Boolean(parsed.text); @@ -722,6 +728,7 @@ export function RectificationAgenticChat(props: RectificationAgenticChatProps) { await loadCaseSnapshot(); } } catch (caught) { + frames.settle(); const aborted = caught instanceof DOMException ? caught.name === "AbortError" : caught instanceof Error && caught.name === "AbortError"; @@ -757,6 +764,7 @@ export function RectificationAgenticChat(props: RectificationAgenticChatProps) { : []; })); } finally { + frames.dispose(); if (runAbort.current === abortController) runAbort.current = null; setPending(false); } diff --git a/frontend/src/hooks/use-consultation-run.ts b/frontend/src/hooks/use-consultation-run.ts index 160fb1fa..b766adb2 100644 --- a/frontend/src/hooks/use-consultation-run.ts +++ b/frontend/src/hooks/use-consultation-run.ts @@ -22,6 +22,7 @@ import { reduceConsultationTimeline, } from "@/lib/consultation-run-timeline"; import { createNdjsonParser, type AgentExecutionReceipt, type ConsultationAgentPublicEvent } from "@/lib/consultation-agent-events"; +import { createStreamFrameBuffer } from "@/lib/stream-frame-buffer"; import { applyThinkingSectionProgress, upsertThinkingSection, @@ -659,6 +660,29 @@ export function useConsultationRun(params: ConsultationRunParams) { let thinkingSections: PublicThinkingSection[] = []; let streamedThinking = ""; let timelineState = emptyConsultationTimeline(); + let currentActivity: AgentActivityView | undefined; + // Every stream event lands in `frames`; it commits at most once per animation + // frame and releases text at a steady pace. Nothing below calls + // setStreamingReply directly while the response body is being read. + const frames = createStreamFrameBuffer({ + initialMeta: null, + flush: (frame) => { + const partialReply = parseAgentReply(frame.answer).text; + latestPartialReply = partialReply; + thinkingSections = applyThinkingSectionProgress(thinkingSections, partialReply); + setStreamingReply({ + sessionId, + text: partialReply, + thinkingText: frame.thinking.trim() || undefined, + thinkingSections: thinkingSections.length ? thinkingSections : undefined, + timeline: timelineState.rows, + activity: currentActivity, + }); + if (partialReply && pendingConsultation.current?.requestId === requestId) { + pendingConsultation.current = { ...pendingConsultation.current, partialReply }; + } + }, + }); try { const response = await fetch("/api/consult", { method: "POST", @@ -721,24 +745,6 @@ export function useConsultationRun(params: ConsultationRunParams) { const reader = response.body.getReader(); const decoder = new TextDecoder(); let answer = ""; - const updateStreamingAnswer = (activity?: AgentActivityView) => { - const partialReply = parseAgentReply(answer).text; - latestPartialReply = partialReply; - thinkingSections = applyThinkingSectionProgress(thinkingSections, partialReply); - setStreamingReply((current) => ({ - sessionId, - text: partialReply, - thinkingText: streamedThinking.trim() || undefined, - thinkingSections: thinkingSections.length ? thinkingSections : undefined, - timeline: timelineState.rows, - activity: activity - ? nextActivityView(current?.sessionId === sessionId ? current.activity : undefined, activity) - : current?.sessionId === sessionId ? current.activity : undefined, - })); - if (partialReply && pendingConsultation.current?.requestId === requestId) { - pendingConsultation.current = { ...pendingConsultation.current, partialReply }; - } - }; const updateActivity = (event: ConsultationAgentPublicEvent) => { let activity: AgentActivityView | undefined; if (event.type === "skill.started") { @@ -766,23 +772,23 @@ export function useConsultationRun(params: ConsultationRunParams) { } else if (event.type === "answer.delta") { activity = { phase: "answer-composition", label: CONSULTATION_COMPOSING_LABEL }; } - if (activity) updateStreamingAnswer(activity); + if (activity) { + currentActivity = nextActivityView(currentActivity, activity); + frames.touch(); + } }; if ((response.headers.get("content-type") ?? "").includes("application/x-ndjson")) { const parser = createNdjsonParser((event) => { timelineState = reduceConsultationTimeline(timelineState, event); - if (event.type === "answer.delta") answer += event.text; + frames.touch(); + if (event.type === "answer.delta") { + answer += event.text; + frames.setAnswer(answer); + } if (event.type === "thinking.delta" && typeof event.text === "string") { streamedThinking += event.text; - setStreamingReply((current) => ({ - sessionId, - text: current?.sessionId === sessionId ? current.text : parseAgentReply(answer).text, - thinkingText: streamedThinking.trim() || undefined, - thinkingSections: current?.sessionId === sessionId ? current.thinkingSections : thinkingSections, - timeline: timelineState.rows, - activity: current?.sessionId === sessionId ? current.activity : undefined, - })); + frames.setThinking(streamedThinking); } if (event.type === "thinking.section") { thinkingSections = applyThinkingSectionProgress( @@ -794,14 +800,6 @@ export function useConsultationRun(params: ConsultationRunParams) { }), parseAgentReply(answer).text, ); - setStreamingReply((current) => ({ - sessionId, - text: current?.sessionId === sessionId ? current.text : parseAgentReply(answer).text, - thinkingText: streamedThinking.trim() || undefined, - thinkingSections, - timeline: timelineState.rows, - activity: current?.sessionId === sessionId ? current.activity : undefined, - })); } if (event.type === "run.completed") { runCompleted = true; @@ -829,6 +827,7 @@ export function useConsultationRun(params: ConsultationRunParams) { parser.push(decoder.decode(value, { stream: true })); } parser.finish(decoder.decode()); + frames.settle(); if (truncatedFailure) { const reply = parseAgentReply(answer); if (!reply.text) throw new ConsultationResponseError(502, truncatedFailure.message); @@ -867,9 +866,11 @@ export function useConsultationRun(params: ConsultationRunParams) { const { done, value } = await reader.read(); if (done) break; answer += decoder.decode(value, { stream: true }); - updateStreamingAnswer(); + frames.setAnswer(answer); } answer += decoder.decode(); + frames.setAnswer(answer); + frames.settle(); } if (controller.signal.aborted) return Boolean(latestPartialReply); const reply = parseAgentReply(answer); @@ -915,6 +916,9 @@ export function useConsultationRun(params: ConsultationRunParams) { void refreshAccount(); return true; } catch (caught) { + // Whatever arrived before the failure is what gets kept, not just the + // part the pacing had released so far. + frames.settle(); const cancelled = controller.signal.aborted; const ownsInterface = pendingConsultation.current?.requestId === requestId; const partialReply = latestPartialReply; @@ -992,6 +996,7 @@ export function useConsultationRun(params: ConsultationRunParams) { } return Boolean(partialReply); } finally { + frames.dispose(); cancellationRequests.current.delete(requestId); const pending = pendingConsultation.current; if (pending?.requestId !== requestId || pending.phase !== "recovering") { diff --git a/frontend/src/lib/chat-markdown-split.ts b/frontend/src/lib/chat-markdown-split.ts new file mode 100644 index 00000000..7e73ed86 --- /dev/null +++ b/frontend/src/lib/chat-markdown-split.ts @@ -0,0 +1,69 @@ +/** + * Split streaming markdown into a stable prefix and a live tail. + * + * While an answer streams, only the tail can still change; everything before + * the last completed block is final. Rendering the prefix through a memoised + * component means each frame re-parses a paragraph, not the whole answer. + * + * The cut is only allowed at a blank line where both sides parse the same on + * their own as they would together: never inside a fenced code block, never + * between two items of the same list (a second `
    ` would add margin that + * the settled render does not have), and never inside a table or blockquote. + */ + +export type StableMarkdownSplit = Readonly<{ + stable: string; + tail: string; +}>; + +const FENCE = /^\s{0,3}(`{3,}|~{3,})/; +const LIST_ITEM = /^\s{0,3}(?:[-*+]|\d{1,9}[.)])\s/; +const INDENTED = /^\s{2,}\S/; +const TABLE_ROW = /^\s{0,3}\|/; + +function lastNonBlank(lines: readonly string[]): string | undefined { + return [...lines].reverse().find((line) => line.trim().length > 0); +} + +/** A blank line does not end a list or a table when the next block continues it. */ +function continuesPreviousBlock(previous: readonly string[], next: string): boolean { + const last = lastNonBlank(previous); + if (last === undefined) return false; + const previousIsList = LIST_ITEM.test(last) || INDENTED.test(last); + const nextIsList = LIST_ITEM.test(next) || INDENTED.test(next); + if (previousIsList && nextIsList) return true; + return TABLE_ROW.test(last) && TABLE_ROW.test(next); +} + +export function splitStableMarkdown(text: string): StableMarkdownSplit { + const lines = text.split("\n"); + let insideFence = false; + let cut = -1; + let currentBlock: string[] = []; + + for (let index = 0; index < lines.length; index += 1) { + const line = lines[index] ?? ""; + if (FENCE.test(line)) insideFence = !insideFence; + if (insideFence) { + currentBlock.push(line); + continue; + } + if (line.trim().length > 0) { + currentBlock.push(line); + continue; + } + // Blank line: the block that just ended is complete only if a non-blank + // line follows later, and the cut is safe only when the next block does + // not continue the previous one. + const nextIndex = lines.findIndex((candidate, at) => at > index && candidate.trim().length > 0); + if (nextIndex < 0) break; + const next = lines[nextIndex] ?? ""; + if (currentBlock.length > 0 && !continuesPreviousBlock(currentBlock, next)) cut = index; + currentBlock = []; + } + + if (cut < 0) return { stable: "", tail: text }; + const stable = lines.slice(0, cut).join("\n"); + const tail = lines.slice(cut + 1).join("\n"); + return { stable, tail }; +} diff --git a/frontend/src/lib/stream-frame-buffer.ts b/frontend/src/lib/stream-frame-buffer.ts new file mode 100644 index 00000000..c96b2c0b --- /dev/null +++ b/frontend/src/lib/stream-frame-buffer.ts @@ -0,0 +1,212 @@ +/** + * Frame-coalesced release of streamed agent output. + * + * Every network chunk used to become its own React commit, and each commit + * re-parsed the whole partial answer. This buffer sits between the event + * parser and `setState`: events mutate an accumulator, and at most one flush + * happens per animation frame. Answer and thinking text are released at a + * steady per-frame pace so a burst of chunks reads as flowing text instead of + * a jump, while a large backlog (reconnect, slow tab) catches up in roughly a + * dozen frames. + * + * Pure release arithmetic lives in exported functions so the policy is + * testable without a DOM; scheduling is injectable for the same reason. + */ + +export const STREAM_RELEASE_MIN_CHARS = 2; +export const STREAM_RELEASE_CATCHUP_DIVISOR = 12; +export const STREAM_HIDDEN_FLUSH_MS = 250; + +/** + * Characters to reveal on one frame. `backlogChars` is how much was waiting + * when the newest text arrived: dividing that by twelve clears any burst in + * about twelve frames, while the two-character floor keeps a slow model from + * reading as stalled. Callers without a backlog figure pass the pending count. + */ +export function streamReleaseCount(pendingChars: number, backlogChars = pendingChars): number { + if (pendingChars <= 0) return 0; + return Math.min( + pendingChars, + Math.max(STREAM_RELEASE_MIN_CHARS, Math.ceil(backlogChars / STREAM_RELEASE_CATCHUP_DIVISOR)), + ); +} + +/** Advance a released prefix toward its target by one frame's worth of text. */ +export function advanceStreamRelease(released: string, target: string, backlogChars?: number): string { + if (!target.startsWith(released)) { + // The target was replaced rather than extended: restart from its head. + return target.slice(0, streamReleaseCount(target.length, backlogChars ?? target.length)); + } + const pending = target.length - released.length; + if (pending <= 0) return target; + return target.slice(0, released.length + streamReleaseCount(pending, backlogChars ?? pending)); +} + +export type StreamFrameSnapshot = Readonly<{ + answer: string; + thinking: string; + meta: Meta; + /** True when this flush released everything that had arrived. */ + settled: boolean; +}>; + +export type StreamFrameScheduler = Readonly<{ + requestFrame: (callback: () => void) => number; + cancelFrame: (handle: number) => void; + requestTimeout: (callback: () => void, delayMs: number) => number; + cancelTimeout: (handle: number) => void; + hidden: () => boolean; +}>; + +export type StreamFrameBufferOptions = Readonly<{ + initialMeta: Meta; + flush: (snapshot: StreamFrameSnapshot) => void; + scheduler?: StreamFrameScheduler; +}>; + +export type StreamFrameBuffer = Readonly<{ + setAnswer: (fullText: string) => void; + setThinking: (fullText: string) => void; + setMeta: (next: Meta | ((current: Meta) => Meta)) => void; + /** Publish meta-only changes (timeline rows, activity) on the next frame. */ + touch: () => void; + /** Release everything received and flush synchronously. */ + settle: () => void; + /** Drop everything, including scheduled work, without flushing. */ + reset: (meta?: Meta) => void; + dispose: () => void; + /** Text released so far, for callers that persist partial output. */ + released: () => Readonly<{ answer: string; thinking: string }>; +}>; + +function pendingChars(released: string, target: string): number { + return target.startsWith(released) ? target.length - released.length : target.length; +} + +function browserScheduler(): StreamFrameScheduler { + return { + requestFrame: (callback) => window.requestAnimationFrame(callback), + cancelFrame: (handle) => window.cancelAnimationFrame(handle), + requestTimeout: (callback, delayMs) => window.setTimeout(callback, delayMs), + cancelTimeout: (handle) => window.clearTimeout(handle), + hidden: () => typeof document !== "undefined" && document.hidden, + }; +} + +export function createStreamFrameBuffer( + options: StreamFrameBufferOptions, +): StreamFrameBuffer { + const scheduler = options.scheduler ?? browserScheduler(); + let targetAnswer = ""; + let targetThinking = ""; + let releasedAnswer = ""; + let releasedThinking = ""; + let answerBacklog = 0; + let thinkingBacklog = 0; + let meta = options.initialMeta; + let dirty = false; + let disposed = false; + let frameHandle: number | null = null; + let timeoutHandle: number | null = null; + + const cancelScheduled = () => { + if (frameHandle !== null) { + scheduler.cancelFrame(frameHandle); + frameHandle = null; + } + if (timeoutHandle !== null) { + scheduler.cancelTimeout(timeoutHandle); + timeoutHandle = null; + } + }; + + const emit = (settled: boolean) => { + dirty = false; + options.flush({ + answer: releasedAnswer, + thinking: releasedThinking, + meta, + settled, + }); + }; + + const step = () => { + frameHandle = null; + timeoutHandle = null; + if (disposed) return; + if (scheduler.hidden()) { + releasedAnswer = targetAnswer; + releasedThinking = targetThinking; + } else { + releasedAnswer = advanceStreamRelease(releasedAnswer, targetAnswer, answerBacklog); + releasedThinking = advanceStreamRelease(releasedThinking, targetThinking, thinkingBacklog); + } + if (releasedAnswer === targetAnswer) answerBacklog = 0; + if (releasedThinking === targetThinking) thinkingBacklog = 0; + const caughtUp = releasedAnswer === targetAnswer && releasedThinking === targetThinking; + emit(caughtUp); + if (!caughtUp) schedule(); + }; + + const schedule = () => { + if (disposed || frameHandle !== null || timeoutHandle !== null) return; + if (scheduler.hidden()) { + timeoutHandle = scheduler.requestTimeout(step, STREAM_HIDDEN_FLUSH_MS); + } else { + frameHandle = scheduler.requestFrame(step); + } + }; + + return { + setAnswer(fullText) { + if (disposed || fullText === targetAnswer) return; + targetAnswer = fullText; + answerBacklog = Math.max(answerBacklog, pendingChars(releasedAnswer, targetAnswer)); + schedule(); + }, + setThinking(fullText) { + if (disposed || fullText === targetThinking) return; + targetThinking = fullText; + thinkingBacklog = Math.max(thinkingBacklog, pendingChars(releasedThinking, targetThinking)); + schedule(); + }, + setMeta(next) { + if (disposed) return; + meta = typeof next === "function" ? (next as (current: Meta) => Meta)(meta) : next; + dirty = true; + schedule(); + }, + touch() { + if (disposed) return; + dirty = true; + schedule(); + }, + settle() { + if (disposed) return; + cancelScheduled(); + releasedAnswer = targetAnswer; + releasedThinking = targetThinking; + answerBacklog = 0; + thinkingBacklog = 0; + emit(true); + }, + reset(nextMeta) { + cancelScheduled(); + targetAnswer = ""; + targetThinking = ""; + releasedAnswer = ""; + releasedThinking = ""; + answerBacklog = 0; + thinkingBacklog = 0; + dirty = false; + if (nextMeta !== undefined) meta = nextMeta; + }, + dispose() { + disposed = true; + cancelScheduled(); + }, + released() { + return { answer: releasedAnswer, thinking: releasedThinking }; + }, + }; +} diff --git a/frontend/tests/chat-markdown-split.test.ts b/frontend/tests/chat-markdown-split.test.ts new file mode 100644 index 00000000..fd8586e9 --- /dev/null +++ b/frontend/tests/chat-markdown-split.test.ts @@ -0,0 +1,74 @@ +import assert from "node:assert/strict"; +import { readFileSync } from "node:fs"; +import test from "node:test"; +import { createElement } from "react"; +import { renderToString } from "react-dom/server"; + +import { StreamingMarkdown } from "../src/components/chat-message-content.tsx"; +import { splitStableMarkdown } from "../src/lib/chat-markdown-split.ts"; + +const contentSource = readFileSync(new URL("../src/components/chat-message-content.tsx", import.meta.url), "utf8"); +const messageRowSource = readFileSync(new URL("../src/components/chat-message-row.tsx", import.meta.url), "utf8"); + +test("the stable prefix ends at the last completed paragraph and the tail keeps streaming", () => { + assert.deepEqual(splitStableMarkdown("只有一段还没写完"), { stable: "", tail: "只有一段还没写完" }); + assert.deepEqual(splitStableMarkdown("第一段。\n\n第二段还在"), { stable: "第一段。", tail: "第二段还在" }); + assert.deepEqual( + splitStableMarkdown("## 标题\n\n第一段。\n\n第二段。\n\n第三"), + { stable: "## 标题\n\n第一段。\n\n第二段。", tail: "第三" }, + ); +}); + +test("the cut never lands inside a fence, between list items, or inside a table", () => { + const fenced = "前言。\n\n```txt\n第一行\n\n第二行"; + assert.deepEqual(splitStableMarkdown(fenced), { stable: "前言。", tail: "```txt\n第一行\n\n第二行" }); + + const looseList = "- 甲\n\n- 乙\n\n- 丙还在"; + assert.deepEqual(splitStableMarkdown(looseList), { stable: "", tail: looseList }); + + const listThenParagraph = "- 甲\n- 乙\n\n总结一下"; + assert.deepEqual(splitStableMarkdown(listThenParagraph), { stable: "- 甲\n- 乙", tail: "总结一下" }); + + const orderedListContinues = "1. 甲\n\n2. 乙\n\n 缩进的补充"; + assert.deepEqual(splitStableMarkdown(orderedListContinues), { stable: "", tail: orderedListContinues }); + + const table = "说明。\n\n| 技法 | 状态 |\n| --- | --- |\n| 甲 | 已执行 |\n\n> 引用"; + assert.deepEqual( + splitStableMarkdown(table), + { stable: "说明。\n\n| 技法 | 状态 |\n| --- | --- |\n| 甲 | 已执行 |", tail: "> 引用" }, + ); +}); + +test("growing the tail keeps the prefix string identical so the memoised prefix is not re-parsed", () => { + const before = splitStableMarkdown("第一段。\n\n第二段。\n\n第三段正在"); + const after = splitStableMarkdown("第一段。\n\n第二段。\n\n第三段正在写，还没有换段"); + assert.equal(before.stable, after.stable); + assert.notEqual(before.tail, after.tail); + + // Only once a new paragraph completes does the prefix move forward. + const later = splitStableMarkdown("第一段。\n\n第二段。\n\n第三段写完了。\n\n第四"); + assert.equal(later.stable, "第一段。\n\n第二段。\n\n第三段写完了。"); +}); + +test("the streaming renderer parses the prefix and the tail as two separate documents", () => { + const calls: string[] = []; + const renderMarkdown = (text: string) => { + calls.push(text); + return createElement("p", null, text); + }; + renderToString(createElement(StreamingMarkdown, { + text: "第一段。\n\n第二段。\n\n第三段还在", + renderMarkdown, + })); + // Server rendering evaluates the tail in the parent and the memoised prefix as a child, + // so compare as a set: what matters is that neither call sees the whole answer. + assert.deepEqual([...calls].sort(), ["第一段。\n\n第二段。", "第三段还在"].sort()); + + // The prefix component is memoised on its text, so an unchanged prefix costs no parse. + assert.match(contentSource, /const StableMarkdownPrefix = memo\(function StableMarkdownPrefix/); + assert.match(contentSource, /splitStableMarkdown\(text\)/); + assert.match(contentSource, /streaming\s*\?\s*= 1); assert.equal(unsplit.unsplitListRenders, tokens.length); assert.ok(unsplit.settledRowRenders > split.settledRowRenders); }); + +test("frame coalescing renders the streaming row once per frame, not once per token", () => { + const messages: ChatMessage[] = [{ role: "user", text: "请继续说明这个月的安排。" }]; + const frames: Array<() => void> = []; + const scheduler: StreamFrameScheduler = { + requestFrame(callback) { + frames.push(callback); + return frames.length; + }, + cancelFrame() { frames.length = 0; }, + requestTimeout() { return 0; }, + cancelTimeout() {}, + hidden: () => false, + }; + + resetHomeStreamingRenderProbe(); + enableHomeStreamingRenderProbe(); + const buffer = createStreamFrameBuffer({ + initialMeta: null, + scheduler, + flush: (frame) => { + const streamingMessage = streamingChatMessageView(messages, true, frame.answer); + assert.ok(streamingMessage); + renderToString(createElement(StreamingMessageEntry, { message: streamingMessage })); + }, + }); + // 200 one-character tokens arrive four per frame across fifty frames. + let answer = ""; + for (let index = 0; index < 200; index += 1) { + answer += "字"; + buffer.setAnswer(answer); + if (index % 4 === 3) for (const callback of frames.splice(0)) callback(); + } + buffer.settle(); + const probe = homeStreamingRenderProbeSnapshot(); + disableHomeStreamingRenderProbe(); + + assert.ok(probe.streamingRowRenders <= 51, `rendered ${probe.streamingRowRenders} times for 200 tokens`); + assert.ok(probe.streamingRowRenders * 3 <= 200); +}); diff --git a/frontend/tests/stream-frame-buffer.test.ts b/frontend/tests/stream-frame-buffer.test.ts new file mode 100644 index 00000000..81fbf787 --- /dev/null +++ b/frontend/tests/stream-frame-buffer.test.ts @@ -0,0 +1,210 @@ +import assert from "node:assert/strict"; +import test from "node:test"; + +import { + STREAM_HIDDEN_FLUSH_MS, + STREAM_RELEASE_CATCHUP_DIVISOR, + STREAM_RELEASE_MIN_CHARS, + advanceStreamRelease, + createStreamFrameBuffer, + streamReleaseCount, + type StreamFrameScheduler, + type StreamFrameSnapshot, +} from "../src/lib/stream-frame-buffer.ts"; + +function fakeScheduler(hidden = () => false) { + const frames: Array<() => void> = []; + const timeouts: Array<{ callback: () => void; delayMs: number }> = []; + let handle = 0; + const scheduler: StreamFrameScheduler = { + requestFrame(callback) { + frames.push(callback); + handle += 1; + return handle; + }, + cancelFrame() { + frames.length = 0; + }, + requestTimeout(callback, delayMs) { + timeouts.push({ callback, delayMs }); + handle += 1; + return handle; + }, + cancelTimeout() { + timeouts.length = 0; + }, + hidden, + }; + return { + scheduler, + tick() { + const pending = frames.splice(0); + for (const callback of pending) callback(); + return pending.length; + }, + tickTimeouts() { + const pending = timeouts.splice(0); + for (const entry of pending) entry.callback(); + return pending; + }, + get scheduledFrames() { + return frames.length; + }, + }; +} + +test("release count is at least two characters and catches up a backlog within about twelve frames", () => { + assert.equal(streamReleaseCount(0), 0); + assert.equal(streamReleaseCount(1), 1); + assert.equal(streamReleaseCount(2), STREAM_RELEASE_MIN_CHARS); + assert.equal(streamReleaseCount(5), STREAM_RELEASE_MIN_CHARS); + assert.equal(streamReleaseCount(24), STREAM_RELEASE_MIN_CHARS); + assert.equal(streamReleaseCount(25), 3); + assert.equal(streamReleaseCount(1200), 1200 / STREAM_RELEASE_CATCHUP_DIVISOR); + + let released = ""; + const target = "字".repeat(3_000); + let frames = 0; + while (released !== target && frames < 100) { + released = advanceStreamRelease(released, target, target.length); + frames += 1; + } + // A 3,000-character backlog that arrived at once clears in twelve frames (~200ms). + assert.equal(frames, STREAM_RELEASE_CATCHUP_DIVISOR); + assert.equal(released, target); + + // Without the backlog figure the pace still floors at two characters per frame. + assert.equal(advanceStreamRelease("", "十二个字符十二个字符十二"), "十二"); +}); + +test("a replaced target that no longer extends the released prefix jumps instead of stalling", () => { + assert.equal(advanceStreamRelease("旧的回答", "新"), "新"); + // A replacement restarts at the paced rate from the new head rather than showing stale text. + assert.equal(advanceStreamRelease("abc", "abd"), "ab"); +}); + +test("many events collapse into one flush per frame and settle releases everything synchronously", () => { + const fake = fakeScheduler(); + const flushes: StreamFrameSnapshot[] = []; + const buffer = createStreamFrameBuffer({ + initialMeta: [], + scheduler: fake.scheduler, + flush: (snapshot) => flushes.push(snapshot), + }); + + // 200 one-character tokens arrive four per frame across fifty frames, the way a + // model streams Chinese text; each frame is allowed one commit. + let answer = ""; + let frameCount = 0; + for (let index = 0; index < 200; index += 1) { + answer += "字"; + buffer.setAnswer(answer); + buffer.setMeta((rows) => [...rows, `row-${index}`]); + if (index % 4 === 3) frameCount += fake.tick(); + } + assert.equal(frameCount, 50); + assert.equal(flushes.length, 50); + assert.ok(flushes.length * 3 <= 200, "at most one commit per frame, not per token"); + for (let index = 1; index < flushes.length; index += 1) { + assert.ok(flushes[index]!.answer.length >= flushes[index - 1]!.answer.length); + assert.ok(flushes[index]!.answer.length - flushes[index - 1]!.answer.length >= STREAM_RELEASE_MIN_CHARS); + } + assert.equal(flushes.at(-1)!.meta.length, 200); + assert.ok(flushes.at(-1)!.answer.length < 200, "pacing is still behind the network"); + + buffer.settle(); + assert.equal(flushes.at(-1)!.answer, answer); + assert.equal(flushes.at(-1)!.settled, true); + assert.equal(fake.scheduledFrames, 0); + assert.equal(buffer.released().answer, answer); +}); + +test("thinking text is paced separately from the answer and meta-only touches still flush", () => { + const fake = fakeScheduler(); + const flushes: StreamFrameSnapshot[] = []; + const buffer = createStreamFrameBuffer({ + initialMeta: null, + scheduler: fake.scheduler, + flush: (snapshot) => flushes.push(snapshot), + }); + buffer.setThinking("先看事业宫，再看大运。"); + fake.tick(); + assert.equal(flushes.length, 1); + assert.equal(flushes[0]!.answer, ""); + assert.ok(flushes[0]!.thinking.length >= STREAM_RELEASE_MIN_CHARS); + assert.equal(flushes[0]!.settled, false); + + buffer.settle(); + assert.equal(flushes.at(-1)!.thinking, "先看事业宫，再看大运。"); + + buffer.touch(); + fake.tick(); + assert.equal(flushes.length, 3); + assert.equal(flushes.at(-1)!.settled, true); +}); + +test("a burst that lands mid-stream is cleared within about twelve frames instead of trickling", () => { + const fake = fakeScheduler(); + const flushes: StreamFrameSnapshot[] = []; + const buffer = createStreamFrameBuffer({ + initialMeta: null, + scheduler: fake.scheduler, + flush: (snapshot) => flushes.push(snapshot), + }); + buffer.setAnswer("字".repeat(20)); + fake.tick(); + buffer.setAnswer("字".repeat(2_420)); + let frames = 0; + while (fake.scheduledFrames > 0 && frames < 100) { + fake.tick(); + frames += 1; + } + assert.equal(flushes.at(-1)!.answer.length, 2_420); + assert.ok(frames <= STREAM_RELEASE_CATCHUP_DIVISOR + 1, `took ${frames} frames`); +}); + +test("a hidden document falls back to a timeout and releases everything at once", () => { + const fake = fakeScheduler(() => true); + const flushes: StreamFrameSnapshot[] = []; + const buffer = createStreamFrameBuffer({ + initialMeta: null, + scheduler: fake.scheduler, + flush: (snapshot) => flushes.push(snapshot), + }); + buffer.setAnswer("字".repeat(500)); + assert.equal(fake.scheduledFrames, 0); + const fired = fake.tickTimeouts(); + assert.equal(fired.length, 1); + assert.equal(fired[0]!.delayMs, STREAM_HIDDEN_FLUSH_MS); + assert.equal(flushes.length, 1); + assert.equal(flushes[0]!.answer.length, 500); + assert.equal(flushes[0]!.settled, true); +}); + +test("reset drops received and released text plus scheduled work, and dispose silences the buffer", () => { + const fake = fakeScheduler(); + const flushes: StreamFrameSnapshot[] = []; + const buffer = createStreamFrameBuffer({ + initialMeta: 1, + scheduler: fake.scheduler, + flush: (snapshot) => flushes.push(snapshot), + }); + buffer.setAnswer("第一次尝试的正文"); + fake.tick(); + assert.equal(flushes.length, 1); + + buffer.reset(2); + assert.equal(fake.scheduledFrames, 0); + assert.deepEqual(buffer.released(), { answer: "", thinking: "" }); + buffer.touch(); + fake.tick(); + assert.equal(flushes.at(-1)!.answer, ""); + assert.equal(flushes.at(-1)!.meta, 2); + + buffer.dispose(); + buffer.setAnswer("不再发布"); + buffer.touch(); + assert.equal(fake.tick(), 0); + buffer.settle(); + assert.equal(flushes.at(-1)!.answer, ""); +});