Aari d1d869c06c
fix(frontend): keep streaming step text stable (#4510)
* fix frontend streaming step flicker

* fix frontend post-tool streaming text
2026-07-28 21:58:51 +08:00

862 lines
25 KiB
TypeScript

import type { AIMessage, Message } from "@langchain/langgraph-sdk";
interface GenericMessageGroup<T = string> {
type: T;
id: string | undefined;
messages: Message[];
}
interface HumanMessageGroup extends GenericMessageGroup<"human"> {}
interface AssistantProcessingGroup extends GenericMessageGroup<"assistant:processing"> {}
interface AssistantMessageGroup extends GenericMessageGroup<"assistant"> {}
interface AssistantPresentFilesGroup extends GenericMessageGroup<"assistant:present-files"> {}
interface AssistantClarificationGroup extends GenericMessageGroup<"assistant:clarification"> {}
interface AssistantSubagentGroup extends GenericMessageGroup<"assistant:subagent"> {}
export type MessageGroup =
| HumanMessageGroup
| AssistantProcessingGroup
| AssistantMessageGroup
| AssistantPresentFilesGroup
| AssistantClarificationGroup
| AssistantSubagentGroup;
const HIDDEN_CONTROL_MESSAGE_NAMES = new Set([
"summary",
"loop_warning",
"todo_reminder",
"todo_completion_reminder",
]);
export function getMessageGroups(
messages: Message[],
{ isCurrentTurnLoading = false }: { isCurrentTurnLoading?: boolean } = {},
): MessageGroup[] {
if (messages.length === 0) {
return [];
}
const groups: MessageGroup[] = [];
let currentTurnStartIndex = -1;
if (isCurrentTurnLoading) {
for (let index = messages.length - 1; index >= 0; index--) {
const message = messages[index];
if (message?.type === "human" && !isHiddenFromUIMessage(message)) {
currentTurnStartIndex = index;
break;
}
}
}
// Returns the last group if it can still accept tool messages
// (i.e. it's an in-flight processing group, not a terminal human/assistant group).
function lastOpenGroup() {
const last = groups[groups.length - 1];
if (
last &&
last.type !== "human" &&
last.type !== "assistant" &&
last.type !== "assistant:clarification"
) {
return last;
}
return null;
}
for (const [messageIndex, message] of messages.entries()) {
if (isHiddenFromUIMessage(message)) {
continue;
}
if (message.type === "human") {
groups.push({ id: message.id, type: "human", messages: [message] });
continue;
}
if (message.type === "tool") {
if (isClarificationToolMessage(message)) {
// Add to the preceding processing group to preserve tool-call association,
// then also open a standalone clarification group for prominent display.
lastOpenGroup()?.messages.push(message);
groups.push({
id: message.id,
type: "assistant:clarification",
messages: [message],
});
} else {
const open = lastOpenGroup();
if (open) {
open.messages.push(message);
} else {
// Fallback for orphan tool messages — LangGraph `messages-tuple` can
// emit tool-result events out of order or replay them from subagent
// state (e.g. bash subagent under LocalSandboxProvider with
// allow_host_bash). When that happens, the tool message arrives after
// a terminal group and lastOpenGroup() returns null. Previously we
// dropped the message with console.error, silently hiding the tool
// result from the UI. Attach to the most recent group instead so the
// user can still see what the agent did.
const lastGroup = groups[groups.length - 1];
if (lastGroup) {
lastGroup.messages.push(message);
} else {
// Leading orphan: `groups` is empty when this tool message
// arrives. Two paths reach here: (1) history pagination cuts by
// event seq, not turn boundaries, so the first loaded page begins
// mid-turn with a tool result whose AI tool-call sits on an
// unloaded older page (#4399); (2) the tool message is preceded
// only by hidden control messages. Open a processing group so it
// stays visible instead of being dropped with a per-render console
// error.
//
// Only case (1) self-heals — loading the older page re-groups the
// tool under its real turn. Case (2), and any truly orphaned tool
// with no AI antecedent, has no page to load: the group persists
// and renders as an empty ChainOfThought shell (convertToSteps
// emits steps only for `type === "ai"`). That empty shell is an
// accepted degradation — still a net win over dropping the result
// and firing console.error every render.
groups.push({
id: message.id,
type: "assistant:processing",
messages: [message],
});
}
}
}
continue;
}
if (message.type === "ai") {
// A message with answer content and no tool calls becomes its own
// assistant bubble below, which already renders the message's
// reasoning_content inside the bubble's <Reasoning> collapsible. Such a
// message must NOT also feed the processing group, or the ChainOfThought
// panel above the bubble paints the identical reasoning a second time
// (#3868). Intermediate reasoning (no content) and tool-calling steps
// still belong in the processing group.
// A content-only message is not necessarily the final answer while its
// turn is still streaming: providers can append tool-call chunks to the
// same message later. Keep that unresolved message in the processing
// group so its visible text does not jump from an assistant bubble into
// the steps panel when the tool call arrives (#4304).
const isUnresolvedAssistantText =
currentTurnStartIndex >= 0 &&
messageIndex > currentTurnStartIndex &&
hasContent(message) &&
!hasToolCalls(message);
const becomesAssistantBubble =
hasContent(message) &&
!hasToolCalls(message) &&
!isUnresolvedAssistantText;
if (hasPresentFiles(message)) {
groups.push({
id: message.id,
type: "assistant:present-files",
messages: [message],
});
} else if (hasSubagent(message)) {
groups.push({
id: message.id,
type: "assistant:subagent",
messages: [message],
});
} else if (
!becomesAssistantBubble &&
(hasReasoning(message) ||
hasToolCalls(message) ||
isUnresolvedAssistantText)
) {
const lastGroup = groups[groups.length - 1];
// Accumulate consecutive intermediate AI messages into one processing group.
if (lastGroup?.type !== "assistant:processing") {
groups.push({
id: message.id,
type: "assistant:processing",
messages: [message],
});
} else {
lastGroup.messages.push(message);
}
}
if (becomesAssistantBubble) {
groups.push({ id: message.id, type: "assistant", messages: [message] });
}
}
}
return groups;
}
export function getBranchableAssistantGroupIds(
groups: MessageGroup[],
isCurrentTurnLoading: boolean,
): Set<string> {
// Hidden messages were already removed by getMessageGroups, matching the
// backend's branch checkpoint visibility rules. Within each visible human
// turn, branching is exposed only when the final AI-bearing group is a
// terminal assistant text group. Processing, present-files, and subagent
// groups do not render assistant actions.
const branchableGroupIds = new Set<string>();
let lastAIGroup: MessageGroup | null = null;
const completeTurn = () => {
if (lastAIGroup?.type === "assistant" && lastAIGroup.id) {
branchableGroupIds.add(lastAIGroup.id);
}
lastAIGroup = null;
};
for (const group of groups) {
if (group.type === "human") {
completeTurn();
continue;
}
if (group.messages.some((message) => message.type === "ai")) {
lastAIGroup = group;
}
}
if (!isCurrentTurnLoading) {
completeTurn();
}
return branchableGroupIds;
}
export type EditableTurn = {
humanMessage: Message;
};
function isTerminalAssistantTextMessage(message: Message | undefined): boolean {
return (
message?.type === "ai" &&
Boolean(extractTextFromMessage(message).trim()) &&
!hasToolCalls(message)
);
}
export function getLatestEditableTurn(
groups: MessageGroup[],
isCurrentTurnLoading: boolean,
): EditableTurn | null {
if (isCurrentTurnLoading) {
return null;
}
let candidate: EditableTurn | null = null;
let currentHumanGroup: MessageGroup | null = null;
let currentTurnGroups: MessageGroup[] = [];
let lastAIGroup: MessageGroup | null = null;
const completeTurn = () => {
if (!currentHumanGroup) {
currentTurnGroups = [];
lastAIGroup = null;
return;
}
const humanMessage = currentHumanGroup?.messages.find(
(message) => message.type === "human" && message.id,
);
let assistantMessage: Message | undefined;
for (let i = (lastAIGroup?.messages.length ?? 0) - 1; i >= 0; i -= 1) {
const message = lastAIGroup?.messages[i];
if (message?.type === "ai" && message.id) {
assistantMessage = message;
break;
}
}
if (
currentHumanGroup &&
lastAIGroup?.type === "assistant" &&
humanMessage &&
isTerminalAssistantTextMessage(assistantMessage)
) {
candidate = {
humanMessage,
};
} else {
candidate = null;
}
currentHumanGroup = null;
currentTurnGroups = [];
lastAIGroup = null;
};
for (const group of groups) {
if (group.type === "human") {
completeTurn();
currentHumanGroup = group;
currentTurnGroups = [group];
continue;
}
if (currentHumanGroup) {
currentTurnGroups.push(group);
}
if (group.messages.some((message) => message.type === "ai")) {
lastAIGroup = group;
}
}
completeTurn();
return candidate;
}
export function groupMessages<T>(
messages: Message[],
mapper: (group: MessageGroup) => T,
): T[] {
return getMessageGroups(messages)
.map(mapper)
.filter((result) => result !== undefined && result !== null) as T[];
}
export function getAssistantTurnUsageMessages(groups: MessageGroup[]) {
const usageMessagesByGroupIndex: Array<Message[] | null> = Array.from(
{ length: groups.length },
() => null,
);
let turnStartIndex: number | null = null;
for (const [index, group] of groups.entries()) {
if (group.type === "human") {
turnStartIndex = null;
continue;
}
turnStartIndex ??= index;
const nextGroup = groups[index + 1];
const isTurnEnd = !nextGroup || nextGroup.type === "human";
if (!isTurnEnd) {
continue;
}
usageMessagesByGroupIndex[index] = groups
.slice(turnStartIndex, index + 1)
.flatMap((currentGroup) => currentGroup.messages)
.filter((message) => message.type === "ai");
turnStartIndex = null;
}
return usageMessagesByGroupIndex;
}
type MessageMetadataLookup = (
message: Message,
index: number,
) => { streamMetadata?: Record<string, unknown> } | undefined;
export type StreamingMessageLookup = {
ids: ReadonlySet<string>;
messages: ReadonlySet<Message>;
};
export function getStreamingMessageLookup(
messages: Message[],
isStreaming: boolean,
getMessagesMetadata?: MessageMetadataLookup,
): StreamingMessageLookup {
const streamingMessageIds = new Set<string>();
const streamingMessages = new Set<Message>();
if (!isStreaming) {
return {
ids: streamingMessageIds,
messages: streamingMessages,
};
}
messages.forEach((message, index) => {
if (!getMessagesMetadata?.(message, index)?.streamMetadata) {
return;
}
if (typeof message.id === "string" && message.id.length > 0) {
streamingMessageIds.add(message.id);
}
streamingMessages.add(message);
});
return {
ids: streamingMessageIds,
messages: streamingMessages,
};
}
export function isAssistantMessageGroupStreaming(
groupMessages: Message[],
streamingMessages: StreamingMessageLookup,
) {
return groupMessages.some((message) => {
if (message.type !== "ai") {
return false;
}
return (
(typeof message.id === "string" &&
message.id.length > 0 &&
streamingMessages.ids.has(message.id)) ||
streamingMessages.messages.has(message)
);
});
}
export function getAssistantTurnCopyData(
messages: Message[],
{ isStreaming = false }: { isStreaming?: boolean } = {},
) {
if (isStreaming) {
return null;
}
return (
[...messages]
.reverse()
.filter((message) => message.type === "ai")
.map((message) => {
const content = extractContentFromMessage(message);
return content ?? extractReasoningContentFromMessage(message) ?? "";
})
.find((content) => content.length > 0) ?? null
);
}
export function getMessageCopyData(message: Message) {
const content = extractContentFromMessage(message);
if (message.type === "human") {
return stripUploadedFilesTag(content);
}
if (content.length > 0) {
return content;
}
return extractReasoningContentFromMessage(message) ?? "";
}
export function extractTextFromMessage(message: Message) {
if (typeof message.content === "string") {
return (
splitInlineReasoningFromAIMessage(message)?.content ??
message.content.trim()
);
}
if (Array.isArray(message.content)) {
return message.content
.map((content) =>
typeof content === "string"
? content
: content.type === "text"
? content.text
: "",
)
.join("\n")
.trim();
}
return "";
}
const THINK_OPEN_TAG = "<think>";
const THINK_TAG_RE = /<think>\s*([\s\S]*?)\s*<\/think>/g;
interface InlineReasoningSplit {
content: string;
reasoning: string | null;
}
function splitInlineReasoning(content: string): InlineReasoningSplit {
const reasoningParts: string[] = [];
// First pass: strip every fully closed `<think>...</think>` pair and
// collect its body as reasoning.
let cleaned = content.replace(THINK_TAG_RE, (_, reasoning: string) => {
const normalized = reasoning.trim();
if (normalized) {
reasoningParts.push(normalized);
}
return "";
});
// Streaming-safe pass: a `<think>` opener whose `</think>` has not arrived
// yet means the rest of the chunk is reasoning in flight. Route it into the
// reasoning slot instead of letting it render as message content (the
// raw-HTML markdown pipeline would otherwise paint the inner text on
// screen until the closing tag lands).
//
// Skip when the opener sits right after a backtick — that is the model
// talking about `<think>` literally inside markdown inline code, not
// actually streaming reasoning.
const openTagIndex = cleaned.indexOf(THINK_OPEN_TAG);
if (openTagIndex !== -1 && cleaned[openTagIndex - 1] !== "`") {
const tail = cleaned.slice(openTagIndex + THINK_OPEN_TAG.length).trim();
if (tail) {
reasoningParts.push(tail);
}
cleaned = cleaned.slice(0, openTagIndex);
}
return {
content: cleaned.trim(),
reasoning: reasoningParts.length > 0 ? reasoningParts.join("\n\n") : null,
};
}
// The split is re-derived on every render: `hasContent`, `hasReasoning`,
// `extractContentFromMessage` and `extractReasoningContentFromMessage` all run
// over the whole message list on each stream chunk, so an unmemoized scan costs
// O(total content) per chunk — quadratic across a long run. Cache per message
// object, keyed by the exact content string it was derived from so a message
// whose `content` is reassigned recomputes instead of serving a stale split.
const inlineReasoningCache = new WeakMap<
object,
{ content: string; split: InlineReasoningSplit }
>();
function splitInlineReasoningFromAIMessage(message: Message) {
if (message.type !== "ai" || typeof message.content !== "string") {
return null;
}
const content = message.content;
const cached = inlineReasoningCache.get(message);
if (cached?.content === content) {
return cached.split;
}
const split = splitInlineReasoning(content);
inlineReasoningCache.set(message, { content, split });
return split;
}
export function extractContentFromMessage(message: Message) {
if (typeof message.content === "string") {
return (
splitInlineReasoningFromAIMessage(message)?.content ??
message.content.trim()
);
}
if (Array.isArray(message.content)) {
return message.content
.map((content) => {
if (typeof content === "string") {
return content;
}
switch (content.type) {
case "text":
return content.text;
case "image_url":
const imageURL = extractURLFromImageURLContent(content.image_url);
return `![image](${imageURL})`;
default:
return "";
}
})
.join("\n")
.trim();
}
return "";
}
export function extractReasoningContentFromMessage(message: Message) {
if (message.type !== "ai") {
return null;
}
if (
message.additional_kwargs &&
"reasoning_content" in message.additional_kwargs
) {
return message.additional_kwargs.reasoning_content as string | null;
}
if (Array.isArray(message.content)) {
const part = message.content[0];
if (part && typeof part === "object" && "thinking" in part) {
return part.thinking as string;
}
}
if (typeof message.content === "string") {
return splitInlineReasoningFromAIMessage(message)?.reasoning ?? null;
}
return null;
}
export function removeReasoningContentFromMessage(message: Message) {
if (message.type !== "ai" || !message.additional_kwargs) {
return;
}
delete message.additional_kwargs.reasoning_content;
}
export function extractURLFromImageURLContent(
content:
| string
| {
url: string;
},
) {
if (typeof content === "string") {
return content;
}
return content.url;
}
export function hasContent(message: Message) {
if (typeof message.content === "string") {
return (
(
splitInlineReasoningFromAIMessage(message)?.content ??
message.content.trim()
).length > 0
);
}
if (Array.isArray(message.content)) {
return message.content.length > 0;
}
return false;
}
export function hasReasoning(message: Message) {
if (message.type !== "ai") {
return false;
}
if (typeof message.additional_kwargs?.reasoning_content === "string") {
return true;
}
if (Array.isArray(message.content)) {
const part = message.content[0];
// Compatible with the Anthropic gateway
return (part as unknown as { type: "thinking" })?.type === "thinking";
}
if (typeof message.content === "string") {
return (
(splitInlineReasoningFromAIMessage(message)?.reasoning ?? null) !== null
);
}
return false;
}
export function hasToolCalls(message: Message) {
return (
message.type === "ai" && message.tool_calls && message.tool_calls.length > 0
);
}
export function hasPresentFiles(message: Message) {
return (
message.type === "ai" &&
message.tool_calls?.some((toolCall) => toolCall.name === "present_files")
);
}
export function isClarificationToolMessage(message: Message) {
return message.type === "tool" && message.name === "ask_clarification";
}
export function extractPresentFilesFromMessage(message: Message) {
if (message.type !== "ai" || !hasPresentFiles(message)) {
return [];
}
const files: string[] = [];
for (const toolCall of message.tool_calls ?? []) {
if (
toolCall.name === "present_files" &&
Array.isArray(toolCall.args.filepaths)
) {
files.push(...(toolCall.args.filepaths as string[]));
}
}
return files;
}
export function hasSubagent(message: AIMessage) {
for (const toolCall of message.tool_calls ?? []) {
if (toolCall.name === "task") {
return true;
}
}
return false;
}
export function findToolCallResult(toolCallId: string, messages: Message[]) {
for (const message of messages) {
if (message.type === "tool" && message.tool_call_id === toolCallId) {
const content = extractTextFromMessage(message);
if (content) {
return content;
}
}
}
return undefined;
}
export function isHiddenFromUIMessage(message: Message) {
if (message.additional_kwargs?.hide_from_ui === true) {
return true;
}
if (
typeof message.name === "string" &&
HIDDEN_CONTROL_MESSAGE_NAMES.has(message.name)
) {
return true;
}
// Only the human branch consults the text. Extracting it up front made every
// caller pay a full content scan for every AI message it was about to
// discard, and this predicate runs over the whole message list on each
// stream chunk (grouping, dedup, human-input state).
if (message.type !== "human") {
return false;
}
const content = extractTextFromMessage(message);
return (
content.includes("<slash_skill_activation>") &&
stripUploadedFilesTag(content).length === 0
);
}
/**
* Represents a file stored in message additional_kwargs.files.
* Used for optimistic UI (uploading state) and structured file metadata.
*/
export interface FileInMessage {
filename: string;
size: number; // bytes
path?: string; // virtual path, may not be set during upload
status?: "uploading" | "uploaded";
}
/**
* Strip backend-injected human context tags from message content.
* Kept under its historical name because callers use it for uploaded-file
* display cleanup.
*/
export function stripUploadedFilesTag(content: string): string {
return content
.replace(
/<(current_uploads|uploaded_files|slash_skill_activation)>[\s\S]*?<\/\1>/g,
"",
)
.trim();
}
/**
* Tag names that backend middlewares wrap around internal payloads before
* letting them ride along inside LangGraph message ``content``.
*
* These markers are *not* user copy — they come from:
*
* - ``UploadsMiddleware`` → ``<current_uploads>`` (``<uploaded_files>``
* before #4174; still emitted by IM channels and present in history)
* - ``SkillActivationMiddleware`` → ``<slash_skill_activation>``
* - ``DynamicContextMiddleware`` → ``<system-reminder>`` (carrying
* ``<memory>`` / ``<current_date>`` inside)
* - ``TodoListMiddleware`` / ``LoopDetectionMiddleware`` style reminders
* live in ``hide_from_ui`` HumanMessages, but their inner payload uses
* the same tag vocabulary.
*
* The primary export filter is {@link isHiddenFromUIMessage}. This list is
* the defence-in-depth strip for any message that — by middleware bug,
* provider quirk, or merge-conflict regression — slips through without
* its ``hide_from_ui`` flag set.
*/
export const INTERNAL_MARKER_TAGS = [
"current_uploads",
"uploaded_files",
"slash_skill_activation",
"system-reminder",
"memory",
"current_date",
] as const;
const INTERNAL_MARKER_RE = new RegExp(
`<(${INTERNAL_MARKER_TAGS.join("|")})>[\\s\\S]*?</\\1>`,
"g",
);
/**
* Strip every known backend-injected marker from message content.
*
* Intended for the chat export path where a marker leaking through is a
* privacy regression. UI render paths should keep using
* {@link stripUploadedFilesTag} — they receive ``hide_from_ui`` messages
* via a separate filter and the narrower function avoids stripping content
* a user might legitimately type into a meta-discussion (e.g. asking the
* model about its own ``<memory>`` system).
*/
export function stripInternalMarkers(content: string): string {
return content.replace(INTERNAL_MARKER_RE, "").trim();
}
// The upload context block renders sizes as human-readable strings
// (uploads_middleware.py::_format_file_entry emits "<n> KB" / "<n> MB",
// mirroring formatBytes). Convert them back to bytes so the parsed
// FileInMessage.size honours its bytes contract and chips re-render at the
// original magnitude instead of e.g. treating "177.6 KB" as 177 bytes.
function parseHumanReadableSize(raw: string): number {
const match = /([\d.]+)\s*(B|KB|MB|GB|TB)?/i.exec(raw.trim());
if (!match) return 0;
const value = parseFloat(match[1] ?? "");
if (!Number.isFinite(value)) return 0;
const multipliers: Record<string, number> = {
B: 1,
KB: 1024,
MB: 1024 ** 2,
GB: 1024 ** 3,
TB: 1024 ** 4,
};
const unit = (match[2] ?? "B").toUpperCase();
return Math.round(value * (multipliers[unit] ?? 1));
}
export function parseUploadedFiles(content: string): FileInMessage[] {
// Match the upload context block; the tag name depends on backend version
// (<current_uploads> since #4174, <uploaded_files> before / on IM paths).
const uploadedFilesRegex =
/<(current_uploads|uploaded_files)>([\s\S]*?)<\/\1>/;
// eslint-disable-next-line @typescript-eslint/prefer-regexp-exec
const match = content.match(uploadedFilesRegex);
if (!match) {
return [];
}
const uploadedFilesContent = match[2];
// Check if it's "No files have been uploaded yet."
if (uploadedFilesContent?.includes("No files have been uploaded yet.")) {
return [];
}
// Check if the backend reported no new files were uploaded in this message
if (uploadedFilesContent?.includes("(empty)")) {
return [];
}
// Parse file list
// Format: - filename (size)\n Path: /path/to/file
const fileRegex = /- ([^\n(]+)\s*\(([^)]+)\)\s*\n\s*Path:\s*([^\n]+)/g;
const files: FileInMessage[] = [];
let fileMatch;
while ((fileMatch = fileRegex.exec(uploadedFilesContent ?? "")) !== null) {
files.push({
filename: fileMatch[1].trim(),
size: parseHumanReadableSize(fileMatch[2]),
path: fileMatch[3].trim(),
});
}
return files;
}