root/Jyotisha
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

perf(chat): coalesce stream events per frame and pace text release

Browse Source 
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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01JUei7K13cYxLHE3Axe4A45
This commit is contained in:
Jesse_Chen
2026-09-02 04:29:01 +00:00
co-authored by Claude Fable 5.1
parent 02f06255c1
commit ad9dba5c79
10 changed files with 778 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/src/hooks/use-consultation-run.ts
+42 -37
View File
@@ -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<null>({
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") {