TYY-2005 36ee986d55
fix(frontend): restore thinking-only steps after reconnect (#7144)
* fix(frontend): restore thinking-only steps after reconnect

* docs(frontend): explain reconnect answer boundary

---------

Co-authored-by: TYY-2005 <TYY-2005@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
2026-10-20 09:57:28 +08:00

4075 lines
134 KiB
TypeScript

import type { AIMessage, Message, Run } from "@langchain/langgraph-sdk";
import type { ThreadsClient } from "@langchain/langgraph-sdk/client";
import { useStream } from "@langchain/langgraph-sdk/react";
import {
type QueryClient,
type InfiniteData,
type QueryFilters,
useInfiniteQuery,
useMutation,
useQuery,
useQueryClient,
} from "@tanstack/react-query";
import { useCallback, useEffect, useMemo, useRef, useState } from "react";
import { toast } from "sonner";
import type { PromptInputMessage } from "@/components/ai-elements/prompt-input";
import { getAPIClient } from "../api";
import { fetch } from "../api/fetcher";
import { getBackendBaseURL } from "../config";
import { useI18n } from "../i18n/hooks";
import { getMessageRunId } from "../messages/run-duration";
import {
extractContentFromMessage,
hasToolCalls,
isHiddenFromUIMessage,
} from "../messages/utils";
import type { FileInMessage } from "../messages/utils";
import { PROJECTS_QUERY_KEY } from "../projects/api";
import type { LocalSettings } from "../settings";
import { isSidecarThread, SIDECAR_METADATA_KEY } from "../sidecar/thread";
import { useSubtaskContext, useUpdateSubtask } from "../tasks/context";
import { taskEventToSubtaskUpdate } from "../tasks/lifecycle";
import { messageToStep } from "../tasks/steps";
import { promptInputFilePartToFile, uploadFiles } from "../uploads";
import { uuid } from "../utils/uuid";
import {
branchThreadFromTurn,
fetchThreadTokenUsage,
moveThreadToProject,
patchThreadMetadata,
searchThreadsByArchive,
type ThreadMetadataPatch,
} from "./api";
import {
dedupeMessagesByIdentity,
insertByTrustedSeq,
isValidMessageSeq,
MESSAGE_SEQ_KEY,
mergeMessages,
messageIdentity,
trustedMessageSeq,
} from "./message-order";
import {
hasRenderedThreadStateUpdate,
reduceThreadStateUpdates,
} from "./stream-state";
import {
buildThreadsSearchQueryOptions,
DEFAULT_THREAD_SEARCH_PARAMS,
filterThreadSearchResults,
type ThreadSearchParams,
} from "./thread-search-query";
import {
retainThreadTokenUsagePlaceholder,
threadTokenUsageQueryKey,
} from "./token-usage";
import type {
AgentThread,
AgentThreadState,
RunMessage,
ThreadTokenUsageResponse,
} from "./types";
import {
THREAD_PINNED_METADATA_KEY,
THREAD_PROJECT_METADATA_KEY,
} from "./utils";
export type ThreadStreamOptions = {
threadId?: string | null | undefined;
displayThreadId?: string | null | undefined;
/**
* Assistant identity sent to the Gateway for run admission and execution.
* Default-chat and sidecar callers use the lead agent; custom-agent pages
* pass their stable agent name so server-side capability checks see the same
* assistant that the runtime loads from the request context.
*/
assistantId?: string;
context: LocalSettings["context"];
isMock?: boolean;
onSend?: (threadId: string) => void;
onStart?: (threadId: string, runId: string) => void;
onFinish?: (state: AgentThreadState) => void;
};
type SendMessageOptions = {
additionalKwargs?: Record<string, unknown>;
additionalInputMessages?: Message[];
/**
* Thread IDs of conversations the user attached for this run. They ride in
* `context.conversation_references`, which the Gateway consumes at admission;
* the LangGraph SDK drops unknown top-level body fields, so the top-level
* request field is not reachable from here. Display metadata for the
* transcript travels separately in `additionalKwargs`.
*/
conversationReferences?: string[];
/**
* Invoked exactly once when the send is genuinely dispatched: after the
* in-flight guard and after any attachments upload, right before the run is
* submitted. It never fires for a dropped send or one whose attachments fail
* to prepare or upload, so callers can safely perform one-time cleanup (e.g.
* clearing quoted references) without losing state a retry still needs.
* The guarantee ends at dispatch: a run that fails afterwards does not
* reject `thread.submit` (the SDK reports it through the stream's
* `onError`), so the send resolves and the composer is cleared as a whole.
*/
onSent?: () => void;
};
type ThreadDeleteClient = {
threads: {
delete: (threadId: string) => Promise<unknown>;
search: (query: Record<string, unknown>) => Promise<AgentThread[]>;
};
};
type ThreadSidecarSearchClient = {
threads: {
search: (query: Record<string, unknown>) => Promise<AgentThread[]>;
};
};
type RegeneratePrepareResponse = {
input: Partial<AgentThreadState>;
checkpoint: {
checkpoint_ns: string;
checkpoint_id: string;
checkpoint_map: Record<string, unknown> | null;
};
metadata: Record<string, unknown>;
target_run_id: string;
};
type EditRegeneratePrepareResponse = RegeneratePrepareResponse & {
replacement_human_message_id: string;
source_message_ids: string[];
};
export type PendingPreparedReplayMask = {
kind: "regenerate" | "edit";
targetRunId: string;
supersededMessageIds: string[];
replacementHumanMessageId?: string;
};
export type ToolResultStatus = "pending" | "success" | "error";
function isErrorToolMessage(message: Message): boolean {
if (message.type !== "tool") {
return false;
}
if (message.status === "error") {
return true;
}
const metadata = message.additional_kwargs?.deerflow_tool_meta;
return (
typeof metadata === "object" &&
metadata !== null &&
"status" in metadata &&
metadata.status === "error"
);
}
export function getToolResultStatus(
messages: Message[],
toolName: string,
): ToolResultStatus {
let latestToolCallId: string | undefined;
for (const message of messages) {
if (message.type !== "ai") {
continue;
}
for (const toolCall of message.tool_calls ?? []) {
if (toolCall.name === toolName && toolCall.id) {
latestToolCallId = toolCall.id;
}
}
}
for (let index = messages.length - 1; index >= 0; index -= 1) {
const message = messages[index];
if (message?.type !== "tool") {
continue;
}
const matches = latestToolCallId
? message.tool_call_id === latestToolCallId
: message.name === toolName;
if (matches) {
return isErrorToolMessage(message) ? "error" : "success";
}
}
return "pending";
}
export function hasToolResult(messages: Message[], toolName: string): boolean {
return getToolResultStatus(messages, toolName) !== "pending";
}
/**
* `additional_kwargs` of this turn's visible human message. The optimistic
* display copy (before and after the upload) and the submitted message all
* build it here, so caller metadata such as quotes and conversation
* references stays on the bubble through the upload. Files staged out-of-band
* (e.g. a project document attached to this thread and carried in
* ``additionalKwargs.files``) ride alongside ``files`` instead of being
* overwritten by them.
*/
export function buildHumanMessageAdditionalKwargs(
additionalKwargs: Record<string, unknown> | undefined,
files: FileInMessage[],
): Record<string, unknown> {
const stagedFiles = Array.isArray(additionalKwargs?.files)
? (additionalKwargs.files as FileInMessage[])
: [];
const allFiles = [...stagedFiles, ...files];
return {
...additionalKwargs,
...(allFiles.length > 0 ? { files: allFiles } : {}),
};
}
export function buildThreadSubmitMessages({
text,
additionalKwargs,
additionalInputMessages = [],
filesForSubmit = [],
humanMessageId,
}: {
text: string;
additionalKwargs?: Record<string, unknown>;
additionalInputMessages?: Message[];
filesForSubmit?: FileInMessage[];
/**
* Client-generated id for the visible human message. The optimistic display
* copy and the actual submit share it, so the server echo (`<id>__user`)
* confirms the exact message the user already sees.
*/
humanMessageId?: string;
}): Message[] {
return [
...additionalInputMessages,
{
type: "human",
...(humanMessageId ? { id: humanMessageId } : {}),
content: [
{
type: "text",
text,
},
],
additional_kwargs: buildHumanMessageAdditionalKwargs(
additionalKwargs,
filesForSubmit,
),
} as Message,
];
}
/**
* Run context sent with `thread.submit`. Both submit paths (send, and the
* regenerate/edit replay) build it here so the client half of the Gateway
* contract stays in one place: conversation references travel only as a plain
* `string[]` under `context.conversation_references`, only when the caller
* attached them, and never from local settings. A stray key in settings is
* dropped rather than forwarded, so a stale value can never grant access.
* `client_timezone` carries the browser's zone (scheduled-task default only).
*/
export function buildRunContext({
settings,
threadId,
extraContext,
conversationReferences,
}: {
settings: LocalSettings["context"];
threadId: string;
extraContext?: Record<string, unknown>;
conversationReferences?: string[];
}): Record<string, unknown> {
const ownedSettings = Object.fromEntries(
Object.entries(settings).filter(
([key]) => key !== "conversation_references",
),
);
return {
...extraContext,
...ownedSettings,
...(conversationReferences?.length
? { conversation_references: [...conversationReferences] }
: {}),
thinking_enabled: settings.mode !== "flash",
is_plan_mode: settings.mode === "pro" || settings.mode === "ultra",
subagent_enabled: settings.mode === "ultra",
reasoning_effort:
settings.reasoning_effort ??
(settings.mode === "ultra"
? "high"
: settings.mode === "pro"
? "medium"
: settings.mode === "thinking"
? "low"
: undefined),
thread_id: threadId,
...clientTimezoneContext(),
};
}
/**
* The browser's IANA zone as `context.client_timezone`. The Gateway reads it
* only to offer a default zone when a chat creates a scheduled task; it never
* reaches the model's context. Omitted when the browser cannot tell.
*/
function clientTimezoneContext(): { client_timezone?: string } {
try {
const timeZone = Intl.DateTimeFormat().resolvedOptions().timeZone;
return typeof timeZone === "string" && timeZone.length > 0
? { client_timezone: timeZone }
: {};
} catch {
return {};
}
}
// Stable identity for "no optimistic messages" so the merged-messages memo
// below is not invalidated by a fresh empty array on every render.
const EMPTY_MESSAGES: Message[] = [];
const EMPTY_RUN_MESSAGES: RunMessage[] = [];
const EMPTY_MESSAGE_IDENTITIES: readonly string[] = [];
const EMPTY_MESSAGE_IDENTITIES_SET: ReadonlySet<string> = new Set<string>();
const ACTIVE_RUN_STATUSES = new Set(["pending", "running"]);
const ACTIVE_RUN_REJOIN_RETRY_DELAYS_MS = [1_000, 2_000] as const;
const MAX_ACTIVE_RUN_REJOIN_ATTEMPTS =
ACTIVE_RUN_REJOIN_RETRY_DELAYS_MS.length + 1;
type ActiveRunRejoinState = {
attempts: number;
inFlight: boolean;
retryTimer: ReturnType<typeof setTimeout> | null;
runId: string | null;
settled: boolean;
threadId: string | null;
};
function createActiveRunRejoinState(
threadId: string | null = null,
runId: string | null = null,
): ActiveRunRejoinState {
return {
attempts: 0,
inFlight: false,
retryTimer: null,
runId,
settled: false,
threadId,
};
}
function readReconnectRun(threadId: string): string | null {
try {
return window.sessionStorage.getItem(`lg:stream:${threadId}`);
} catch {
return null;
}
}
function rememberReconnectRun(threadId: string, runId: string): void {
try {
window.sessionStorage.setItem(`lg:stream:${threadId}`, runId);
} catch {
// The stream can still be joined, but SDK stop cannot cancel it without
// the tab-local run pointer.
}
}
function clearReconnectRun(threadId: string, runId: string): void {
try {
const key = `lg:stream:${threadId}`;
if (window.sessionStorage.getItem(key) === runId) {
window.sessionStorage.removeItem(key);
}
} catch {
// Storage access is best-effort and must never block stream cleanup.
}
}
/**
* The turn this client submitted, recorded at dispatch time. The visible human
* input gets one client-generated identity shared by the optimistic display
* copy and the submitted message, so the turn anchor is a *known* identity
* instead of a guess derived from the pre-submit baseline. `humanIdentity` is
* the normalized identity (`message:<id>`) of that human, or null when the
* turn has no visible human (hidden human-input reply, regenerate replay) —
* such turns must never borrow an older visible human as their anchor.
*/
export type LocalTurnAnchor = {
threadId: string;
humanIdentity: string | null;
baselineIdentities: ReadonlySet<string>;
/** Canonical REST-history identities already loaded when this turn began. */
preSubmitHistoryIdentities: ReadonlySet<string>;
/** Transient-bridge identities already established before this turn began. */
preSubmitBridgeIdentities: ReadonlySet<string>;
/**
* Highest authoritative feed position known before submit. Older pages that
* arrive later may still be confirmed as pre-submit history through this
* boundary; messages from later external turns may not.
*/
preSubmitMaxSeq?: number;
};
function isNonEmptyString(value: string | undefined): value is string {
return typeof value === "string" && value.length > 0;
}
const EMPTY_THREAD_VALUES: AgentThreadState = {
title: "",
messages: [],
artifacts: [],
todos: [],
};
const SUMMARIZATION_MIDDLEWARE_UPDATE_KEYS = new Set([
"SummarizationMiddleware.before_model",
"DeerFlowSummarizationMiddleware.before_model",
]);
function maxMessageSeq(messages: Message[]): number | undefined {
let maxSeq: number | undefined;
for (const message of messages) {
const seq = trustedMessageSeq(message);
if (seq !== undefined && (maxSeq === undefined || seq > maxSeq)) {
maxSeq = seq;
}
}
return maxSeq;
}
function getConfirmedPreSubmitHistoryIdentities(
visibleHistory: Message[],
localTurnAnchor: LocalTurnAnchor | null,
): Set<string> {
if (localTurnAnchor === null) {
return new Set();
}
const confirmed = new Set([
...localTurnAnchor.preSubmitHistoryIdentities,
...localTurnAnchor.preSubmitBridgeIdentities,
]);
const maxSeq = localTurnAnchor.preSubmitMaxSeq;
if (maxSeq === undefined) {
return confirmed;
}
for (const message of visibleHistory) {
const identity = messageIdentity(message);
const seq = trustedMessageSeq(message);
if (identity !== undefined && seq !== undefined && seq <= maxSeq) {
confirmed.add(identity);
}
}
return confirmed;
}
function findMessageRunIdByIdentity(
messages: Message[],
identity: string,
): string | undefined {
for (const message of messages) {
if (messageIdentity(message) !== identity) {
continue;
}
const runId = getMessageRunId(message);
if (runId) {
return runId;
}
}
return undefined;
}
function dedupeRunMessagesByIdentity(messages: RunMessage[]): RunMessage[] {
const lastIndexByIdentity = new Map<string, number>();
messages.forEach((message, index) => {
const identity = messageIdentity(message.content);
if (identity) {
lastIndexByIdentity.set(`${message.run_id}:${identity}`, index);
}
});
return messages.filter((message, index) => {
const identity = messageIdentity(message.content);
if (!identity) {
return true;
}
return lastIndexByIdentity.get(`${message.run_id}:${identity}`) === index;
});
}
export function removeSetItems<T>(
values: ReadonlySet<T>,
itemsToRemove: Iterable<T>,
) {
const next = new Set(values);
for (const item of itemsToRemove) {
next.delete(item);
}
return next;
}
export function buildVisibleHistoryMessages(
messageRows: RunMessage[],
supersededRunIds: ReadonlySet<string>,
) {
const visibleRows = messageRows.filter(
(message) => !supersededRunIds.has(message.run_id),
);
// Content and position converge separately for a repeated identity: the
// newest visible row supplies the content, but the position stays the
// earliest trusted feed row — mirroring the backend `get_message_seqs`
// earliest-seq-wins rule, so a re-persisted update cannot push the message
// towards the tail. Hidden control copies never contribute a visible
// position (they only serve as a fallback when no visible row carries the
// identity).
const earliestSeqByIdentity = new Map<string, number>();
const earliestVisibleSeqByIdentity = new Map<string, number>();
for (const row of visibleRows) {
const identity = messageIdentity(row.content);
if (!identity || !isValidMessageSeq(row.seq)) {
continue;
}
const known = earliestSeqByIdentity.get(identity);
if (known === undefined || row.seq < known) {
earliestSeqByIdentity.set(identity, row.seq);
}
if (!isHiddenFromUIMessage(row.content)) {
const knownVisible = earliestVisibleSeqByIdentity.get(identity);
if (knownVisible === undefined || row.seq < knownVisible) {
earliestVisibleSeqByIdentity.set(identity, row.seq);
}
}
}
const deduped = dedupeMessagesByIdentity([
// Carry the owning run_id onto the content message so historical subtask
// cards can fetch their persisted step history on expand (#3779). run_id
// lives on the RunMessage wrapper and would otherwise be dropped here.
// seq rides along for the same reason: it is the thread-global position
// this feed is ordered by, and merging needs it on the message itself to
// place a checkpoint copy that falls outside the loaded window (#4666).
...visibleRows.map((message) => ({
...message.content,
run_id: message.run_id,
additional_kwargs: {
...message.content.additional_kwargs,
[MESSAGE_SEQ_KEY]: message.seq,
},
})),
]);
return deduped.map((message) => {
const identity = messageIdentity(message);
if (!identity) {
return message;
}
const earliestSeq =
earliestVisibleSeqByIdentity.get(identity) ??
earliestSeqByIdentity.get(identity);
if (
earliestSeq === undefined ||
message.additional_kwargs?.[MESSAGE_SEQ_KEY] === earliestSeq
) {
return message;
}
return {
...message,
additional_kwargs: {
...message.additional_kwargs,
[MESSAGE_SEQ_KEY]: earliestSeq,
},
} as Message;
});
}
export type ThreadMessagesPageResponse = {
data: RunMessage[];
has_more: boolean;
next_before_seq: number | null;
};
/**
* Validate the sequence fields that history reconciliation and pagination use
* as runtime identities. The static RunMessage type cannot protect this JSON
* boundary from version skew or malformed responses.
*/
export function parseThreadMessagesPageResponse(
value: unknown,
): ThreadMessagesPageResponse {
if (typeof value !== "object" || value === null) {
throw new Error("Thread history returned an invalid response.");
}
const data = Reflect.get(value, "data");
const hasMore = Reflect.get(value, "has_more");
const nextBeforeSeq = Reflect.get(value, "next_before_seq");
if (!Array.isArray(data) || typeof hasMore !== "boolean") {
throw new Error("Thread history returned an invalid response.");
}
const seenSeqs = new Set<number>();
for (const row of data) {
const seq =
typeof row === "object" && row !== null
? Reflect.get(row, "seq")
: undefined;
if (!isValidMessageSeq(seq)) {
throw new Error("Thread history returned a row with an invalid seq.");
}
if (seenSeqs.has(seq)) {
throw new Error("Thread history returned duplicate seq values.");
}
seenSeqs.add(seq);
}
if (
(hasMore && !isValidMessageSeq(nextBeforeSeq)) ||
(!hasMore && nextBeforeSeq !== null)
) {
throw new Error(
"Thread history returned an invalid next_before_seq cursor.",
);
}
return value as ThreadMessagesPageResponse;
}
export function getThreadHistoryNextPageParam(
lastPage: ThreadMessagesPageResponse,
): number | undefined {
if (!lastPage.has_more) {
return undefined;
}
if (lastPage.next_before_seq === null) {
console.warn(
"Thread history returned has_more without next_before_seq; pagination cannot continue.",
);
return undefined;
}
return lastPage.next_before_seq;
}
export const threadHistoryQueryKey = (threadId: string) =>
["thread-messages", threadId] as const;
export function buildThreadMessagesPageUrl(
baseUrl: string,
threadId: string,
beforeSeq?: number,
limit?: number,
) {
const normalizedBaseUrl = baseUrl.replace(/\/$/, "");
const path = `/api/threads/${encodeURIComponent(threadId)}/messages/page`;
const url = new URL(
`${normalizedBaseUrl}${path}`,
typeof window !== "undefined" ? window.location.origin : "http://localhost",
);
if (beforeSeq !== undefined) {
url.searchParams.set("before_seq", String(beforeSeq));
}
if (limit !== undefined) url.searchParams.set("limit", String(limit));
return normalizedBaseUrl ? url.toString() : `${url.pathname}${url.search}`;
}
export function flattenThreadHistoryPages(
pages: ThreadMessagesPageResponse[],
): RunMessage[] {
return dedupeRunMessagesByIdentity(
pages
.slice()
.reverse()
.flatMap((page) => page.data),
);
}
/** Read persisted history independently of the loaded UI and compacted state. */
export async function fetchThreadExportMessages(threadId: string) {
const pages: ThreadMessagesPageResponse[] = [];
let beforeSeq: number | undefined;
do {
const response = await fetch(
buildThreadMessagesPageUrl(getBackendBaseURL(), threadId, beforeSeq, 200),
);
if (!response.ok) {
throw new Error("Failed to load conversation for export.");
}
const page = parseThreadMessagesPageResponse(await response.json());
const next = getThreadHistoryNextPageParam(page);
if (
next !== undefined &&
(page.data.length === 0 ||
next !== page.data[0]?.seq ||
(beforeSeq !== undefined && next >= beforeSeq))
) {
throw new Error("Thread export history cursor did not advance.");
}
pages.push(page);
beforeSeq = next;
} while (beforeSeq !== undefined);
return mergeMessages(
buildVisibleHistoryMessages(
flattenThreadHistoryPages(pages),
new Set<string>(),
),
[],
[],
);
}
/**
* Preserve rows that this client has already loaded while newest-first cursor
* pages move forward during a long run.
*
* A background refetch recalculates every loaded page from the refreshed first
* page. When older pages have not all been loaded yet, that can displace rows
* which were visible a moment ago even though they still exist on the server.
* Thread-global seq is the authoritative order; refreshed copies win without
* moving their established position.
*/
export function reconcileThreadHistoryRows(
previousRows: RunMessage[],
currentRows: RunMessage[],
isAuthoritativeComplete: boolean,
): RunMessage[] {
const sourceRows = isAuthoritativeComplete
? currentRows
: [...previousRows, ...currentRows];
if (sourceRows.some((row) => !isValidMessageSeq(row.seq))) {
console.error(
"Thread history reconciliation received an invalid sequence value.",
);
// Never skip an invalid row or feed it into Map/Array.sort: either choice
// can silently lose or misorder messages. A failed refresh keeps the last
// known-good snapshot; an invalid first snapshot degrades to server order.
return previousRows.length > 0 ? previousRows : currentRows;
}
const rowsBySeq = new Map<number, RunMessage>();
for (const row of sourceRows) {
rowsBySeq.set(row.seq, row);
}
const sortedRows = [...rowsBySeq.values()].sort(
(left, right) => left.seq - right.seq,
);
const reconciled = dedupeRunMessagesByIdentity(sortedRows);
// Deduping collapses a re-persisted identity to its newest copy, but per
// the backend `get_message_seqs` earliest-seq-wins rule the message keeps
// the position it first occupied (see buildVisibleHistoryMessages).
// Re-anchor the surviving row to the earliest seq the identity held in
// this window, otherwise a re-persisted update pushes the message towards
// the tail. Prefer visible rows so a hidden control copy cannot move its
// visible twin; hidden rows provide a fallback only when no visible copy exists.
const earliestSeqByRunIdentity = new Map<string, number>();
const earliestVisibleSeqByRunIdentity = new Map<string, number>();
for (const row of sortedRows) {
const identity = messageIdentity(row.content);
if (!identity || !isValidMessageSeq(row.seq)) {
continue;
}
const key = `${row.run_id}:${identity}`;
const known = earliestSeqByRunIdentity.get(key);
if (known === undefined || row.seq < known) {
earliestSeqByRunIdentity.set(key, row.seq);
}
if (!isHiddenFromUIMessage(row.content)) {
const knownVisible = earliestVisibleSeqByRunIdentity.get(key);
if (knownVisible === undefined || row.seq < knownVisible) {
earliestVisibleSeqByRunIdentity.set(key, row.seq);
}
}
}
const anchored = reconciled.map((row) => {
const identity = messageIdentity(row.content);
if (!identity) {
return row;
}
const key = `${row.run_id}:${identity}`;
const earliestSeq =
earliestVisibleSeqByRunIdentity.get(key) ??
earliestSeqByRunIdentity.get(key);
if (earliestSeq === undefined || row.seq === earliestSeq) {
return row;
}
return { ...row, seq: earliestSeq };
});
// Re-anchoring can change row order. Sort before comparing with the retained
// snapshot so unchanged reconciliations can still reuse the previous array.
anchored.sort((left, right) => left.seq - right.seq);
if (
anchored.length === previousRows.length &&
anchored.every((row, index) => {
const previous = previousRows[index];
if (!previous) {
return false;
}
// Re-anchoring rebuilds the row object on every pass; treat a row whose
// run_id, seq and content reference all match as unchanged so the
// retained-history state can stay referentially stable.
return (
row === previous ||
(row.run_id === previous.run_id &&
row.seq === previous.seq &&
row.content === previous.content)
);
})
) {
return previousRows;
}
return anchored;
}
// mergeMessages now lives in ./message-order (pure, unit-testable ordering
// module); it is imported above and re-exported here so existing consumers of
// this module keep working unchanged.
export { mergeMessages };
/**
* Collect live run ids that were not part of the pre-submit checkpoint.
* An empty result is safe because restoreLocalTurnMessageOrder independently
* anchors the current turn from the pending human's run_id when interrupt/stop
* has already flushed the live steps into canonical history.
*/
export function getCurrentTurnRunIds(
messages: Message[],
baselineMessageIdentities: ReadonlySet<string> | null,
confirmedHistoryIdentities: ReadonlySet<string> = EMPTY_MESSAGE_IDENTITIES_SET,
): Set<string> {
const runIds = new Set<string>();
if (baselineMessageIdentities === null) {
return runIds;
}
for (const message of messages) {
if (
(message.type !== "ai" && message.type !== "tool") ||
isHiddenFromUIMessage(message)
) {
continue;
}
const identity = messageIdentity(message);
const runId = getMessageRunId(message);
if (
runId &&
(!identity ||
(!baselineMessageIdentities.has(identity) &&
!confirmedHistoryIdentities.has(identity)))
) {
runIds.add(runId);
}
}
return runIds;
}
/**
* Keep messages from a locally submitted turn behind that turn's user input.
* LangGraph `messages-tuple` events can publish the first AI/tool steps before
* canonical history contains the user message. Those steps are not part of the
* pre-submit baseline, so move only that visible pending segment behind the
* latest new human message. Conversely, a baseline or history-confirmed message
* from an established turn can be woven after that human before a live
* checkpoint tail; move those established messages back before the input. The
* caller keeps the baseline after stream completion because the SDK may retain
* its transient event order until the next submit.
*/
export function restoreLocalTurnMessageOrder(
messages: Message[],
baselineMessageIdentities: ReadonlySet<string>,
confirmedHistoryIdentities: ReadonlySet<string> = EMPTY_MESSAGE_IDENTITIES_SET,
currentTurnRunIds: ReadonlySet<string> = EMPTY_MESSAGE_IDENTITIES_SET,
anchorHumanIdentity?: string | null,
canonicalHistoryIdentities: ReadonlySet<string> = confirmedHistoryIdentities,
): Message[] {
// When the caller recorded the exact human identity this turn submitted
// (LocalTurnAnchor), only that message may anchor the repair. `null` means
// the turn has no visible human at all (hidden human-input reply,
// regenerate replay): no human may be borrowed from history. An identity
// that has not reached the render snapshot yet means the display is a frame
// behind — keep the established order instead of re-anchoring on an older
// history-only human (absence from the checkpoint baseline is not proof
// that a message belongs to this turn).
if (anchorHumanIdentity === null) {
return messages;
}
let pendingHumanIndex = -1;
if (anchorHumanIdentity !== undefined) {
for (let index = messages.length - 1; index >= 0; index -= 1) {
const message = messages[index]!;
if (
message.type === "human" &&
!isHiddenFromUIMessage(message) &&
messageIdentity(message) === anchorHumanIdentity
) {
pendingHumanIndex = index;
break;
}
}
} else {
// Compat path for callers without a local-turn anchor. Context compaction
// can omit an older human from the checkpoint while the REST history page
// still supplies it, so anchor on the LATEST visible human that is not in
// the baseline; an old history-only turn then cannot claim the current
// stream. A freshly submitted human may not have a server id yet, and
// identity-less messages are necessarily absent from the baseline.
for (let index = messages.length - 1; index >= 0; index -= 1) {
const message = messages[index]!;
const identity = messageIdentity(message);
if (
message.type === "human" &&
!isHiddenFromUIMessage(message) &&
(identity === undefined || !baselineMessageIdentities.has(identity))
) {
pendingHumanIndex = index;
break;
}
}
}
if (pendingHumanIndex < 0) {
return messages;
}
// The fixed confirmed set decides which suffix messages may move back across
// this turn's human. The wider canonical set has a different job in the
// prefix: a REST-history message is not speculative current-turn output just
// because the latest-page window advanced after submit.
const isConfirmedHistoryMessage = (identity: string | undefined) =>
identity !== undefined && confirmedHistoryIdentities.has(identity);
const isCanonicalHistoryMessage = (identity: string | undefined) =>
identity !== undefined && canonicalHistoryIdentities.has(identity);
// Steps of the CURRENT run must never be treated as displaced history: after
// an interrupt/stop the current turn's already-executed steps are persisted
// into canonical history, but they still belong AFTER the new human input.
// The pending human message itself carries the current run_id, so it is the
// most reliable anchor even when the live checkpoint no longer holds the
// current turn's steps (stop/interrupt can flush them to history).
const effectiveCurrentTurnRunIds = new Set(currentTurnRunIds);
const pendingHumanRunId = getMessageRunId(messages[pendingHumanIndex]!);
if (pendingHumanRunId) {
effectiveCurrentTurnRunIds.add(pendingHumanRunId);
}
const isCurrentTurnStep = (message: Message) => {
const runId = getMessageRunId(message);
return runId !== undefined && effectiveCurrentTurnRunIds.has(runId);
};
const stablePrefix: Message[] = [];
const earlyPendingSteps: Message[] = [];
for (const message of messages.slice(0, pendingHumanIndex)) {
const identity = messageIdentity(message);
const isVisiblePendingStep =
(message.type === "ai" || message.type === "tool") &&
!isHiddenFromUIMessage(message) &&
identity !== undefined &&
!baselineMessageIdentities.has(identity) &&
((!isCanonicalHistoryMessage(identity) &&
!isConfirmedHistoryMessage(identity)) ||
isCurrentTurnStep(message));
if (isVisiblePendingStep) {
earlyPendingSteps.push(message);
} else {
stablePrefix.push(message);
}
}
const displacedMessages: Message[] = [];
const stableSuffix: Message[] = [];
for (const message of messages.slice(pendingHumanIndex + 1)) {
const identity = messageIdentity(message);
const wasPresentBeforeSubmit =
identity !== undefined && baselineMessageIdentities.has(identity);
const wasConfirmedInPreviousHistory =
(message.type === "ai" || message.type === "tool") &&
!isHiddenFromUIMessage(message) &&
isConfirmedHistoryMessage(identity) &&
!isCurrentTurnStep(message);
if (wasPresentBeforeSubmit || wasConfirmedInPreviousHistory) {
displacedMessages.push(message);
} else {
stableSuffix.push(message);
}
}
if (earlyPendingSteps.length === 0 && displacedMessages.length === 0) {
return messages;
}
return [
...stablePrefix,
...displacedMessages,
messages[pendingHumanIndex]!,
...earlyPendingSteps,
...stableSuffix,
];
}
/**
* Reconnect/reload counterpart of {@link restoreLocalTurnMessageOrder}.
*
* After a mid-run page reload the local-turn baseline is empty, so the local
* restore never runs — yet the same ordering race still applies: replayed
* `messages-tuple` steps can reach the merged list before the turn's human
* message (the retained replay buffer may even have dropped it), and the
* live-only human is then woven in before the next shared history anchor,
* leaving steps of the SAME run above the user message they belong to.
*
* Canonical history is seq-sorted, so a visible AI/tool step sitting above
* the last visible human while another message of the same run sits below it
* (a "same-run sandwich") is provably misplaced. Run_id-less steps are
* live-only (history rows always carry run_id) and are attributable to the
* reconnected run only when a run_id-less step also follows the human.
*
* Only steps after the last terminal assistant answer (visible content, no
* tool calls) in the segment are candidates: such an answer completes the
* turn that owns it, so everything up to it belongs to a finished turn and
* must keep its position even when run_ids are absent or uniform across
* turns (branch-seeded history, mocked feeds). A still-streaming text step
* can look like a terminal answer before its tool call arrives (#4304); it
* then stays above the human until canonical history heals the order — an
* accepted transient, far safer than pulling a completed turn's answer
* below the next user message. A resent turn after an interrupted run and
* pagination orphans from older turns (#4399) fail the checks as well and
* are left untouched. When an interrupted turn and the next turn share one
* synthetic run_id, the previous human anchor plus the missing terminal
* answer makes step ownership ambiguous, so that layout is also preserved.
*/
export function restoreReconnectedTurnMessageOrder(
messages: Message[],
): Message[] {
let humanIndex = -1;
for (let index = messages.length - 1; index >= 0; index--) {
const message = messages[index];
if (message?.type === "human" && !isHiddenFromUIMessage(message)) {
humanIndex = index;
break;
}
}
if (humanIndex <= 0) {
return messages;
}
// Only the segment since the previous visible human can belong to the
// active turn; older turns are anchored by their own human message.
let segmentStart = 0;
for (let index = humanIndex - 1; index >= 0; index--) {
const message = messages[index];
if (message?.type === "human" && !isHiddenFromUIMessage(message)) {
segmentStart = index + 1;
break;
}
}
if (segmentStart >= humanIndex) {
return messages;
}
// A terminal assistant answer completes the turn that owns it. Only steps
// after the last such boundary can belong to the active turn.
let candidateStart = segmentStart;
for (let index = segmentStart; index < humanIndex; index++) {
const message = messages[index]!;
if (
message.type === "ai" &&
!isHiddenFromUIMessage(message) &&
// Thinking-only blocks (Anthropic) are not completed answers: require
// extractable text/image content to close the turn.
extractContentFromMessage(message).length > 0 &&
!hasToolCalls(message)
) {
candidateStart = index + 1;
}
}
const previousHumanRunId =
segmentStart > 0 ? getMessageRunId(messages[segmentStart - 1]!) : undefined;
const hasUniformRunIdSincePreviousHuman =
previousHumanRunId !== undefined &&
messages
.slice(segmentStart - 1)
.every(
(message) =>
isHiddenFromUIMessage(message) ||
getMessageRunId(message) === previousHumanRunId,
);
if (candidateStart === segmentStart && hasUniformRunIdSincePreviousHuman) {
return messages;
}
const runIdsAfter = new Set<string>();
let hasLiveOnlyStepAfter = false;
for (const message of messages.slice(humanIndex + 1)) {
const runId = getMessageRunId(message);
if (runId) {
runIdsAfter.add(runId);
} else if (
(message.type === "ai" || message.type === "tool") &&
!isHiddenFromUIMessage(message)
) {
hasLiveOnlyStepAfter = true;
}
}
const misplacedSteps: Message[] = [];
const stablePrefix = messages.slice(0, candidateStart);
for (const message of messages.slice(candidateStart, humanIndex)) {
const isVisibleStep =
(message.type === "ai" || message.type === "tool") &&
!isHiddenFromUIMessage(message);
const runId = getMessageRunId(message);
if (
isVisibleStep &&
((runId !== undefined && runIdsAfter.has(runId)) ||
(runId === undefined && hasLiveOnlyStepAfter))
) {
misplacedSteps.push(message);
} else {
stablePrefix.push(message);
}
}
if (misplacedSteps.length === 0) {
return messages;
}
return [
...stablePrefix,
messages[humanIndex]!,
...misplacedSteps,
...messages.slice(humanIndex + 1),
];
}
/**
* Keep a run-scoped ledger of every visible message that reached a committed
* UI frame. Live checkpoint windows can roll forward between two
* summarization events; replacing this ledger with only the newest window
* would make the intervening steps impossible to rescue at the next
* RemoveMessage(ALL).
*
* The newest visible copy wins by identity without moving its established
* position. Explicitly superseded messages are removed so regeneration cannot
* revive an answer that the UI intentionally hid.
*/
export function mergeRenderedMessageLedger(
previouslyRenderedMessages: Message[],
visibleMessages: Message[],
supersededMessageIds: ReadonlySet<string> = new Set<string>(),
): Message[] {
const isEligible = (message: Message) =>
messageIdentity(message) !== undefined &&
(!message.id || !supersededMessageIds.has(message.id));
const retainedPrevious =
supersededMessageIds.size === 0
? previouslyRenderedMessages.filter(
(message) => messageIdentity(message) !== undefined,
)
: previouslyRenderedMessages.filter(isEligible);
const eligibleVisibleMessages = visibleMessages.filter(isEligible);
if (retainedPrevious.length === 0) {
return eligibleVisibleMessages;
}
return mergeMessages(retainedPrevious, eligibleVisibleMessages, []);
}
/**
* Derive the live turns that context summarization is about to drop and that
* therefore need a short-lived visual bridge until run-event history catches up.
*
* Summarization emits `RemoveMessage(ALL)` + a hidden summary + the retained
* tail. Everything in the current live thread that is absent from the retained
* visible window is being removed; we keep those (minus the summary control
* messages already tracked) so the UI can still show the full conversation
* (#3825). Comparing identities instead of slicing at the first retained
* message also handles a protected early input followed by a recent tail.
*/
export function computeSummarizationTransientMessages(
currentMessages: Message[],
summarizationMessages: Message[],
summarizedMessageIds: ReadonlySet<string>,
previouslyRenderedMessages: Message[] = EMPTY_MESSAGES,
): Message[] {
const retainedVisibleIdentities = new Set(
summarizationMessages
.filter((message) => message.type !== "remove")
.filter((message) => !isHiddenFromUIMessage(message))
.map(messageIdentity)
.filter(isNonEmptyString),
);
// Updates can outrun React while the SDK applies RemoveMessage(ALL). In that
// case currentMessages may already be the retained post-compaction window
// even though the previous committed UI frame still showed the removed
// processing steps. Use that frame as the chronological base, then overlay
// fresher live copies. This rescues only messages the user actually saw and
// preserves the unloaded-history-gap protection in the bridge resolver.
const captureMessages =
previouslyRenderedMessages.length > 0
? mergeMessages(previouslyRenderedMessages, currentMessages, [])
: currentMessages;
const moved: Message[] = [];
for (const message of captureMessages) {
const identity = messageIdentity(message);
if (identity && retainedVisibleIdentities.has(identity)) {
continue;
}
if (!summarizedMessageIds.has(message.id ?? "")) {
moved.push(message);
}
}
return moved;
}
/**
* Overlay messages rescued from context summarization on top of the
* (possibly stale) visible history so the merged view never drops them.
*
* Background (#3825): after summarization the backend removes every live
* message (`RemoveMessage(ALL)`) while canonical run events can still be
* waiting for the journal flush/refetch lifecycle. Reading the captured turns
* from a synchronous transient buffer keeps the merge correct during that gap.
*
* Canonical history is cursor-paginated from newest to oldest. A rescued turn
* can therefore be older than the first row in the currently loaded page even
* though both came from the same pre-compression checkpoint. ``bridgeOrder``
* retains identities that canonical history has already confirmed so missing
* rescued turns can be inserted next to an overlapping anchor instead of being
* blindly appended after the newest page. Canonical copies always win.
*/
export function resolveTransientHistoryBridge(
visibleHistory: Message[],
transientMessages: Message[],
bridgeOrder: readonly string[] = transientMessages
.map(messageIdentity)
.filter(isNonEmptyString),
previouslyRenderedOrder: readonly string[] = EMPTY_MESSAGE_IDENTITIES,
): Message[] {
if (transientMessages.length === 0) {
return visibleHistory;
}
const presentIdentities = new Set(
visibleHistory.map(messageIdentity).filter(isNonEmptyString),
);
const missing = transientMessages.filter((message) => {
const identity = messageIdentity(message);
// Identity-less messages are intentionally skipped: without a stable
// identity they cannot be matched against history to drain or dedupe, so
// overlaying them would risk a permanent duplicate. Canonical history will
// surface them after the run journal is flushed and the page refetches.
return identity !== undefined && !presentIdentities.has(identity);
});
if (missing.length === 0) {
return visibleHistory;
}
// Trusted seq outranks identity anchors — the same position priority
// mergeMessages applies. A rescued message whose seq is known lands exactly
// where the feed places it, even when no bridge identity overlaps the
// loaded window; anchor weaving below remains the fallback for entries
// without a trustworthy position.
const seqPositioned: Message[] = [];
const unpositioned: Message[] = [];
for (const message of missing) {
if (trustedMessageSeq(message) !== undefined) {
seqPositioned.push(message);
} else {
unpositioned.push(message);
}
}
// Place rescued seq rows first so weaving can anchor unsequenced
// neighbors to them without a later insertion reversing their order.
const positionedHistory = insertByTrustedSeq(visibleHistory, seqPositioned);
const anchorIdentities = new Set(
positionedHistory.map(messageIdentity).filter(isNonEmptyString),
);
const missingByIdentity = new Map(
unpositioned.flatMap((message) => {
const identity = messageIdentity(message);
return identity ? [[identity, message] as const] : [];
}),
);
// This mirrors mergeMessages' identity-anchor weaving shape, but transient
// messages never replace canonical copies and identity-less entries are
// intentionally excluded to avoid permanent duplicates.
const beforeAnchor = new Map<string, Message[]>();
const emittedMissingIdentities = new Set<string>();
const previouslyRenderedIndex = new Map(
previouslyRenderedOrder.map((identity, index) => [identity, index]),
);
let pending: Message[] = [];
let lastAnchorIdentity: string | undefined;
for (const identity of bridgeOrder) {
if (anchorIdentities.has(identity)) {
if (pending.length > 0) {
// A rescued seq anchor preserves its captured prefix even if React
// has not rendered it yet. Only a leading prefix anchored to loaded
// history needs proof that it does not span an unloaded cursor gap.
if (
lastAnchorIdentity !== undefined ||
!presentIdentities.has(identity)
) {
beforeAnchor.set(identity, [
...(beforeAnchor.get(identity) ?? []),
...pending,
]);
} else {
const anchorRenderIndex = previouslyRenderedIndex.get(identity);
if (anchorRenderIndex !== undefined) {
const safeRenderedPrefix = pending
.filter((message) => {
const pendingIdentity = messageIdentity(message);
const renderIndex = pendingIdentity
? previouslyRenderedIndex.get(pendingIdentity)
: undefined;
return (
renderIndex !== undefined && renderIndex < anchorRenderIndex
);
})
.sort((left, right) => {
const leftIndex =
previouslyRenderedIndex.get(messageIdentity(left) ?? "") ??
Number.MAX_SAFE_INTEGER;
const rightIndex =
previouslyRenderedIndex.get(messageIdentity(right) ?? "") ??
Number.MAX_SAFE_INTEGER;
return leftIndex - rightIndex;
});
if (safeRenderedPrefix.length > 0) {
beforeAnchor.set(identity, safeRenderedPrefix);
}
}
}
}
pending = [];
lastAnchorIdentity = identity;
continue;
}
const message = missingByIdentity.get(identity);
if (message && !emittedMissingIdentities.has(identity)) {
pending.push(message);
emittedMissingIdentities.add(identity);
}
}
// No bridge identity overlaps a positioned row. This is the original
// persistence-gap case: loaded history is older and the rescued live turns
// belong after it.
if (!lastAnchorIdentity) {
return [...positionedHistory, ...unpositioned];
}
// A candidate added before its ordering snapshot (or carrying an identity
// absent from that snapshot) cannot be anchored. Keep it in capture order at
// the trailing edge of the anchored bridge rather than dropping it.
for (const message of unpositioned) {
const identity = messageIdentity(message);
if (identity && !emittedMissingIdentities.has(identity)) {
pending.push(message);
emittedMissingIdentities.add(identity);
}
}
const resolved: Message[] = [];
for (const message of positionedHistory) {
const identity = messageIdentity(message);
if (identity) {
resolved.push(...(beforeAnchor.get(identity) ?? []));
}
resolved.push(message);
if (identity === lastAnchorIdentity) {
resolved.push(...pending);
}
}
return resolved;
}
export function mergeTransientHistoryBridge(
currentBridge: Message[],
capturedMessages: Message[],
): Message[] {
const merged = dedupeMessagesByIdentity(currentBridge);
const indexByIdentity = new Map<string, number>();
merged.forEach((message, index) => {
const identity = messageIdentity(message);
if (identity) {
indexByIdentity.set(identity, index);
}
});
for (const captured of dedupeMessagesByIdentity(capturedMessages)) {
const identity = messageIdentity(captured);
const existingIndex = identity ? indexByIdentity.get(identity) : undefined;
if (existingIndex === undefined) {
if (identity) {
indexByIdentity.set(identity, merged.length);
}
merged.push(captured);
continue;
}
const existing = merged[existingIndex];
if (
existing &&
(!isHiddenFromUIMessage(captured) || isHiddenFromUIMessage(existing))
) {
// Refresh the buffered snapshot without moving its first-known
// chronological position — and without dropping a trusted position the
// earlier copy already carried. Repeated compression can recapture
// protected prefix messages before a newer tail; a refresh without seq
// must not reset the position the bridge resolver relies on.
const existingSeq = trustedMessageSeq(existing);
const capturedSeq = trustedMessageSeq(captured);
const keptSeq =
existingSeq !== undefined &&
(capturedSeq === undefined || existingSeq < capturedSeq)
? existingSeq
: capturedSeq;
merged[existingIndex] =
keptSeq !== undefined && capturedSeq !== keptSeq
? ({
...captured,
additional_kwargs: {
...captured.additional_kwargs,
[MESSAGE_SEQ_KEY]: keptSeq,
},
} as Message)
: captured;
}
}
return merged;
}
/**
* Preserve the complete checkpoint-relative identity order independently from
* bridge candidates. Confirmed candidates are pruned from the render buffer,
* but their identities remain useful as non-rendering pagination anchors.
*/
export function mergeTransientHistoryBridgeOrder(
currentOrder: readonly string[],
capturedMessages: Message[],
): readonly string[] {
const capturedOrder = dedupeMessagesByIdentity(capturedMessages)
.map(messageIdentity)
.filter(isNonEmptyString);
// Clone lazily and return the input when nothing is appended: this runs per
// render while the bridge is active, and a fresh array would invalidate the
// coalesced render memo on every chunk (#4409 Phase 1).
let merged: string[] | null = null;
const seen = new Set(currentOrder);
for (const identity of capturedOrder) {
if (!seen.has(identity)) {
seen.add(identity);
(merged ??= [...currentOrder]).push(identity);
}
}
return merged ?? currentOrder;
}
export function resolveThreadTransientHistoryBridge(
visibleHistory: Message[],
transientMessages: Message[],
bridgeThreadId: string | null,
currentThreadId: string | null | undefined,
bridgeOrder?: readonly string[],
previouslyRenderedOrder?: readonly string[],
): Message[] {
if (!bridgeThreadId || bridgeThreadId !== currentThreadId) {
return visibleHistory;
}
return resolveTransientHistoryBridge(
visibleHistory,
transientMessages,
bridgeOrder,
previouslyRenderedOrder,
);
}
/**
* Drop transient-buffer entries that canonical history has already
* absorbed. This keeps the buffer a transient bridge across the async gap
* rather than a second long-lived source of truth — otherwise a stale copy
* could resurrect a message that history later filtered out (e.g. a superseded
* or regenerated run).
*/
export function pruneConfirmedTransientMessages(
transientMessages: Message[],
visibleHistory: Message[],
): Message[] {
if (transientMessages.length === 0) {
return transientMessages;
}
const confirmedIdentities = new Set(
visibleHistory.map(messageIdentity).filter(isNonEmptyString),
);
return transientMessages.filter((message) => {
const identity = messageIdentity(message);
return !identity || !confirmedIdentities.has(identity);
});
}
function getMessagesAfterBaseline(
messages: Message[],
baselineMessageIds: ReadonlySet<string>,
): Message[] {
return messages.filter((message) => {
const id = messageIdentity(message);
return !id || !baselineMessageIds.has(id);
});
}
/**
* Human-message baseline for a prepared replay (regenerate / edit).
*
* A replay masks the turn it supersedes, so those messages leave the live
* message list the moment the mask is applied. Baselining on the pre-mask count
* means the replacement only ever restores the count instead of exceeding it,
* and the optimistic copy is never recognised as confirmed. That matters for
* the first turn of a thread, where the runtime re-keys the replacement message
* so identity comparison cannot stand in for the count either.
*/
export function countHumanMessagesExcludingSuperseded(
messages: Message[],
supersededMessageIds: readonly string[],
): number {
const superseded = new Set(supersededMessageIds);
return messages.filter(
(message) =>
message.type === "human" && (!message.id || !superseded.has(message.id)),
).length;
}
export function getVisibleOptimisticMessages(
optimisticMessages: Message[],
previousHumanMessageCount: number,
currentHumanMessageCount: number,
): Message[] {
if (
optimisticMessages.some((message) => message.type === "human") &&
currentHumanMessageCount > previousHumanMessageCount
) {
return [];
}
return optimisticMessages;
}
export function areOptimisticMessagesConfirmed(
optimisticMessages: Message[],
persistedMessages: Message[],
): boolean {
const optimisticIdentities = optimisticMessages
.map(messageIdentity)
.filter(isNonEmptyString);
if (optimisticIdentities.length === 0) {
return false;
}
const persistedIdentities = new Set(
persistedMessages.map(messageIdentity).filter(isNonEmptyString),
);
return optimisticIdentities.every((identity) =>
persistedIdentities.has(identity),
);
}
export function getSummarizationMiddlewareMessages(
data: unknown,
): Message[] | undefined {
if (typeof data !== "object" || data === null) {
return undefined;
}
for (const [key, update] of Object.entries(data)) {
if (!SUMMARIZATION_MIDDLEWARE_UPDATE_KEYS.has(key)) {
continue;
}
if (typeof update !== "object" || update === null) {
continue;
}
const messages = Reflect.get(update, "messages");
if (Array.isArray(messages)) {
return [...messages] as Message[];
}
}
return undefined;
}
export const STREAM_RENDER_COALESCE_MS = 80;
export type CoalesceDecision =
| { action: "flush-now" }
| { action: "schedule"; delayMs: number }
| { action: "wait" };
/**
* Decide how an incoming stream update reaches the rendered snapshot: flush
* immediately once a full interval has elapsed (leading edge), otherwise
* schedule exactly one trailing flush for the interval remainder. Unlike a
* debounce, the delay never extends past the interval, so a dense stream can
* never starve rendering.
*/
export function decideCoalesce(
nowMs: number,
lastFlushMs: number,
intervalMs: number,
hasPendingTimer: boolean,
): CoalesceDecision {
if (nowMs - lastFlushMs >= intervalMs) {
return { action: "flush-now" };
}
if (hasPendingTimer) {
return { action: "wait" };
}
return { action: "schedule", delayMs: intervalMs - (nowMs - lastFlushMs) };
}
/**
* While a run is streaming, expose the messages array as a snapshot that
* updates at most once per interval instead of once per SSE chunk, so the
* merge/group/render pipeline runs per frame budget rather than per token
* (#4409 Phase 1). When the stream is idle the latest array passes straight
* through, keeping non-stream updates immediate.
*/
function sameMessageArray(a: Message[], b: Message[]): boolean {
return (
a === b ||
(a.length === b.length && a.every((message, index) => message === b[index]))
);
}
export function useCoalescedStreamMessages(
messages: Message[],
isStreaming: boolean,
intervalMs: number = STREAM_RENDER_COALESCE_MS,
): Message[] {
// `null` means "no snapshot belongs to the current stream": the live array is
// returned until the leading-edge flush lands, so a snapshot left over from an
// earlier stream can never be painted. This hook outlives thread switches (the
// chat page deliberately avoids re-mounting, see its `onStart` comment), so a
// retained snapshot would otherwise be another thread's messages.
const [snapshot, setSnapshot] = useState<Message[] | null>(null);
const latestRef = useRef(messages);
latestRef.current = messages;
// Monotonic clock: a wall-clock step (NTP, sleep/wake) between two reads
// would otherwise be added to the remaining interval and stall the flush for
// the length of the jump. -Infinity means "never flushed", so the first
// update of a stream always takes the leading edge.
const lastFlushRef = useRef(Number.NEGATIVE_INFINITY);
const timerRef = useRef<ReturnType<typeof setTimeout> | null>(null);
// Every publication goes through the shallow-equality guard: inputs whose
// identity churns without content change (the SDK getter mints fresh arrays)
// must not re-trigger renders, or this effect would setState-loop.
const publish = useCallback(() => {
setSnapshot((previous) =>
previous !== null && sameMessageArray(previous, latestRef.current)
? previous
: latestRef.current,
);
}, []);
const clearPendingFlush = useCallback(() => {
if (timerRef.current !== null) {
clearTimeout(timerRef.current);
timerRef.current = null;
}
}, []);
useEffect(() => {
if (!isStreaming) {
clearPendingFlush();
// Drop the flush baseline so the leading edge is per stream rather than
// per hook instance: a run starting within one interval of the previous
// one must not have its first frame deferred. Dropping the snapshot with
// it costs one render per stream end, versus one per idle message change
// if the snapshot were instead kept in sync while nothing reads it.
lastFlushRef.current = Number.NEGATIVE_INFINITY;
setSnapshot((previous) => (previous === null ? previous : null));
return;
}
const now = performance.now();
const decision = decideCoalesce(
now,
lastFlushRef.current,
intervalMs,
timerRef.current !== null,
);
if (decision.action === "flush-now") {
// A trailing timer can still be armed here: timers fire late under
// main-thread load, which is exactly when a chunk overtakes one. Leaving
// it would publish a second time and slip the next interval forward.
clearPendingFlush();
lastFlushRef.current = now;
publish();
} else if (decision.action === "schedule") {
timerRef.current = setTimeout(() => {
timerRef.current = null;
// Read the clock again: timers fire late under load, and the next
// interval must start from the real flush.
lastFlushRef.current = performance.now();
publish();
}, decision.delayMs);
}
}, [messages, isStreaming, intervalMs, publish, clearPendingFlush]);
useEffect(
() => () => {
if (timerRef.current !== null) {
clearTimeout(timerRef.current);
}
},
[],
);
return isStreaming && snapshot !== null ? snapshot : messages;
}
export function upsertThreadInSearchCache(
queryClient: QueryClient,
thread: AgentThread,
) {
queryClient.setQueriesData(
{
queryKey: ["threads", "search"],
exact: false,
},
(oldData: Array<AgentThread> | undefined) => {
if (!oldData) {
return [thread];
}
const existingIndex = oldData.findIndex(
(t) => t.thread_id === thread.thread_id,
);
if (existingIndex === -1) {
return [thread, ...oldData];
}
return oldData.map((t, index) => {
if (index !== existingIndex) {
return t;
}
return {
...thread,
...t,
metadata: {
...(thread.metadata ?? {}),
...(t.metadata ?? {}),
},
values: {
...thread.values,
...t.values,
},
};
});
},
);
}
export function upsertThreadInInfiniteCache(
queryClient: QueryClient,
thread: AgentThread,
) {
// Run-created snapshots do not carry archive metadata. Let the server
// decide membership instead of injecting a running chat into both views.
const hasArchiveFilter = ({ queryKey }: { queryKey: readonly unknown[] }) =>
typeof (queryKey[2] as InfiniteThreadsParams | undefined)?.archived ===
"boolean";
void queryClient.invalidateQueries({
queryKey: INFINITE_THREADS_QUERY_KEY_PREFIX,
predicate: hasArchiveFilter,
});
queryClient.setQueriesData(
{
queryKey: INFINITE_THREADS_QUERY_KEY_PREFIX,
exact: false,
predicate: (query) => !hasArchiveFilter(query),
},
(oldData: InfiniteThreadsData | undefined) => {
if (!oldData) {
return oldData;
}
const merged = oldData.pages.map((page) => ({
...page,
threads: page.threads.map((t) =>
t.thread_id === thread.thread_id
? {
...thread,
...t,
metadata: {
...(thread.metadata ?? {}),
...(t.metadata ?? {}),
},
values: {
...thread.values,
...t.values,
},
}
: t,
),
}));
const exists = merged.some((page) =>
page.threads.some((t) => t.thread_id === thread.thread_id),
);
if (exists) {
return { ...oldData, pages: merged };
}
const firstPage = merged[0] ?? {
threads: [],
hiddenCount: 0,
hasMore: false,
};
const restPages = merged.slice(1);
return {
...oldData,
pages: [
{ ...firstPage, threads: [thread, ...firstPage.threads] },
...restPages,
],
};
},
);
}
export function invalidateStoppedThreadCaches(
queryClient: QueryClient,
threadId: string | null | undefined,
isMock = false,
) {
void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
void queryClient.invalidateQueries({
queryKey: INFINITE_THREADS_QUERY_KEY_PREFIX,
});
// A finished run updates the title/recency the project page thread list
// shows ([...PROJECTS_QUERY_KEY, "threads", id, ...]).
void queryClient.invalidateQueries({
queryKey: [...PROJECTS_QUERY_KEY, "threads"],
});
if (!threadId || isMock) {
return;
}
void queryClient.invalidateQueries({ queryKey: ["thread", threadId] });
void queryClient.invalidateQueries({
queryKey: threadHistoryQueryKey(threadId),
});
void queryClient.invalidateQueries({
queryKey: ["thread", "metadata", threadId, isMock],
});
void queryClient.invalidateQueries({
queryKey: threadTokenUsageQueryKey(threadId),
});
}
export const STOP_THREAD_FINALIZATION_REFETCH_DELAY_MS = 1500;
function scheduleStoppedThreadFinalizationRefetch(
queryClient: QueryClient,
threadId: string | null | undefined,
isMock = false,
) {
if (isMock) {
return;
}
globalThis.setTimeout(() => {
invalidateStoppedThreadCaches(queryClient, threadId, isMock);
}, STOP_THREAD_FINALIZATION_REFETCH_DELAY_MS);
}
export async function stopThreadAndInvalidateCaches(
queryClient: QueryClient,
stop: () => Promise<void> | void,
threadId: string | null | undefined,
isMock = false,
) {
try {
await stop();
} finally {
invalidateStoppedThreadCaches(queryClient, threadId, isMock);
scheduleStoppedThreadFinalizationRefetch(queryClient, threadId, isMock);
}
}
function getStreamErrorMessage(error: unknown): string {
if (typeof error === "string" && error.trim()) {
return error;
}
if (error instanceof Error && error.message.trim()) {
return error.message;
}
if (typeof error === "object" && error !== null) {
const message = Reflect.get(error, "message");
if (typeof message === "string" && message.trim()) {
return message;
}
const nestedError = Reflect.get(error, "error");
if (nestedError instanceof Error && nestedError.message.trim()) {
return nestedError.message;
}
if (typeof nestedError === "string" && nestedError.trim()) {
return nestedError;
}
}
return "Request failed.";
}
async function readResponseErrorMessage(
response: Response,
fallback = "Request failed.",
) {
try {
const data = await response.json();
if (typeof data?.detail === "string" && data.detail.trim()) {
return data.detail;
}
} catch {
// Use the fallback below when the response body is not JSON.
}
return response.statusText || fallback;
}
function getHttpStatus(error: unknown): number | undefined {
if (typeof error !== "object" || error === null) {
return undefined;
}
const status = Reflect.get(error, "status");
if (typeof status === "number") {
return status;
}
const response = Reflect.get(error, "response");
if (typeof response === "object" && response !== null) {
const responseStatus = Reflect.get(response, "status");
if (typeof responseStatus === "number") {
return responseStatus;
}
}
return undefined;
}
function isThreadMissingError(error: unknown): boolean {
const status = getHttpStatus(error);
// Treat 403 like 404 here to avoid disclosing whether an inaccessible thread
// exists; callers redirect stale/inaccessible URLs back to a blank chat.
return status === 403 || status === 404;
}
export function useThreadStream({
threadId,
displayThreadId,
assistantId = "lead_agent",
context,
isMock,
onSend,
onStart,
onFinish,
}: ThreadStreamOptions) {
const { t } = useI18n();
const currentViewThreadId = displayThreadId ?? threadId ?? null;
const currentViewThreadIdRef = useRef(currentViewThreadId);
currentViewThreadIdRef.current = currentViewThreadId;
// Optimistic messages shown before the server stream responds.
const [optimisticMessages, setOptimisticMessages] = useState<Message[]>([]);
const [optimisticThreadId, setOptimisticThreadId] = useState<string | null>(
null,
);
const [liveMessagesThreadId, setLiveMessagesThreadId] = useState<
string | null
>(null);
const [pendingSupersededRunIds, setPendingSupersededRunIds] = useState<
ReadonlySet<string>
>(() => new Set());
const [pendingSupersededMessageIds, setPendingSupersededMessageIds] =
useState<ReadonlySet<string>>(() => new Set());
const [isUploading, setIsUploading] = useState(false);
// Track the thread ID that is currently streaming to handle thread changes during streaming
const [onStreamThreadId, setOnStreamThreadId] = useState(() => threadId);
// Ref to track current thread ID across async callbacks without causing re-renders,
// and to allow access to the current thread id in onUpdateEvent
const threadIdRef = useRef<string | null>(threadId ?? null);
const startedRef = useRef(false);
const pendingUsageBaselineMessageIdsRef = useRef<Set<string>>(new Set());
const pendingPreparedReplayRef = useRef<PendingPreparedReplayMask | null>(
null,
);
const listeners = useRef({
onSend,
onStart,
onFinish,
});
const {
messages: history,
hasMore: hasMoreHistory,
loadMore: loadMoreHistory,
loading: isHistoryLoading,
} = useThreadHistory(onStreamThreadId ?? "", {
enabled: !isMock,
pendingSupersededRunIds,
});
const runsQuery = useThreadRuns(onStreamThreadId ?? undefined, {
enabled: !isMock,
});
const activeRunId = useMemo(
() =>
runsQuery.data?.find((run) => ACTIVE_RUN_STATUSES.has(String(run.status)))
?.run_id,
[runsQuery.data],
);
const activeRunRejoinRef = useRef<ActiveRunRejoinState>(
createActiveRunRejoinState(),
);
const [activeRunRejoinRetry, setActiveRunRejoinRetry] = useState(0);
// Runs reads can lag behind SDK completion, including the initial read.
// Keep completed IDs across recovery-state resets so stale "running" data
// cannot restart a submitted or natively reconnected stream.
const completedRunIdsRef = useRef(new Set<string>());
// onCreated fires only for runs this hook submitted. A failed submit stays
// with the submit flow; only SDK reconnects are handed to recovery.
const submittedRunIdRef = useRef<string | null>(null);
// Keep listeners ref updated with latest callbacks
useEffect(() => {
listeners.current = { onSend, onStart, onFinish };
}, [onSend, onStart, onFinish]);
useEffect(() => {
const normalizedThreadId = threadId ?? null;
if (!normalizedThreadId) {
// Reset when the UI moves back to a brand new unsaved thread.
startedRef.current = false;
setOnStreamThreadId(normalizedThreadId);
} else {
setOnStreamThreadId(normalizedThreadId);
}
threadIdRef.current = normalizedThreadId;
}, [threadId]);
const handleStreamStart = useCallback((_threadId: string, _runId: string) => {
threadIdRef.current = _threadId;
setOptimisticThreadId((currentOptimisticThreadId) => {
const currentView = currentViewThreadIdRef.current;
if (
currentOptimisticThreadId &&
(currentOptimisticThreadId === currentView ||
currentOptimisticThreadId === _threadId)
) {
return _threadId;
}
return currentOptimisticThreadId;
});
setLiveMessagesThreadId((currentLiveMessagesThreadId) => {
const currentView = currentViewThreadIdRef.current;
if (
currentLiveMessagesThreadId &&
(currentLiveMessagesThreadId === currentView ||
currentLiveMessagesThreadId === _threadId)
) {
return _threadId;
}
return currentLiveMessagesThreadId;
});
if (!startedRef.current) {
listeners.current.onStart?.(_threadId, _runId);
startedRef.current = true;
}
setOnStreamThreadId(_threadId);
}, []);
const queryClient = useQueryClient();
const { tasksRef, setTasks } = useSubtaskContext();
const updateSubtask = useUpdateSubtask();
const scheduleActiveRunRejoinRetry = useCallback(
(run: Pick<Run, "thread_id" | "run_id"> | undefined) => {
const rejoin = activeRunRejoinRef.current;
// SDK 1.6.0 LGP joinStream errors include { thread_id, run_id }; history
// errors omit it. Re-verify this callback contract when upgrading the SDK.
if (
run &&
!rejoin.inFlight &&
run.run_id !== submittedRunIdRef.current &&
readReconnectRun(run.thread_id) === run.run_id
) {
// The SDK's same-tab reconnect tries once and keeps its pointer on
// error, so the recovery effect would keep deferring to a stream that
// no longer exists. Release the pointer and let recovery take over.
clearReconnectRun(run.thread_id, run.run_id);
setActiveRunRejoinRetry((current) => current + 1);
return;
}
if (
!rejoin.inFlight ||
!rejoin.threadId ||
!rejoin.runId ||
run?.thread_id !== rejoin.threadId ||
run?.run_id !== rejoin.runId
) {
return;
}
rejoin.inFlight = false;
clearReconnectRun(rejoin.threadId, rejoin.runId);
const retryDelay = ACTIVE_RUN_REJOIN_RETRY_DELAYS_MS[rejoin.attempts - 1];
if (retryDelay === undefined) {
return;
}
rejoin.retryTimer = setTimeout(() => {
rejoin.retryTimer = null;
setActiveRunRejoinRetry((current) => current + 1);
}, retryDelay);
},
[],
);
const settleActiveRunRejoin = useCallback(() => {
const rejoin = activeRunRejoinRef.current;
if (!rejoin.inFlight) {
return;
}
rejoin.inFlight = false;
rejoin.settled = true;
if (rejoin.retryTimer !== null) {
clearTimeout(rejoin.retryTimer);
rejoin.retryTimer = null;
}
}, []);
const clearPreparedReplayMasks = useCallback(
(replay: PendingPreparedReplayMask | null) => {
if (!replay) {
return;
}
setPendingSupersededRunIds((current) =>
removeSetItems(current, [replay.targetRunId]),
);
setPendingSupersededMessageIds((current) =>
removeSetItems(current, replay.supersededMessageIds),
);
if (pendingPreparedReplayRef.current === replay) {
pendingPreparedReplayRef.current = null;
}
},
[],
);
const thread = useStream<AgentThreadState>({
client: getAPIClient(isMock),
assistantId,
threadId: onStreamThreadId,
reconnectOnMount: true,
fetchStateHistory: { limit: 1 },
// Batch stream updates received in the same macrotask without adding a
// fixed debounce interval that could delay a continuously active stream.
// Coalesce same-tick stream events into one React notification. Only the
// boolean tier is safe: the SDK's numeric tier is a trailing debounce that
// starves UI updates while chunks keep arriving faster than the window.
// Keep explicit: SDK types claim @default true, but runtime uses throttle ?? false.
throttle: true,
onCreated(meta) {
submittedRunIdRef.current = meta.run_id;
handleStreamStart(meta.thread_id, meta.run_id);
const now = new Date().toISOString();
upsertThreadInSearchCache(queryClient, {
thread_id: meta.thread_id,
created_at: now,
updated_at: now,
metadata: context.agent_name ? { agent_name: context.agent_name } : {},
status: "busy",
values: {
title: t.pages.newChat,
messages: [],
artifacts: [],
},
interrupts: {},
});
upsertThreadInInfiniteCache(queryClient, {
thread_id: meta.thread_id,
created_at: now,
updated_at: now,
metadata: context.agent_name ? { agent_name: context.agent_name } : {},
status: "busy",
values: {
title: t.pages.newChat,
messages: [],
artifacts: [],
},
interrupts: {},
});
if (context.agent_name && !isMock) {
void getAPIClient()
.threads.update(meta.thread_id, {
metadata: { agent_name: context.agent_name },
})
.catch(() => ({}));
}
},
onUpdateEvent(data, { mutate }) {
if (hasRenderedThreadStateUpdate(data)) {
mutate((previous) => reduceThreadStateUpdates(previous, data) ?? {});
}
const _messages = getSummarizationMiddlewareMessages(data);
if (_messages && _messages.length >= 2) {
for (const m of _messages) {
// Backward-compat shim: pre-PR2 threads may still carry a synthetic
// HumanMessage(name="summary") from the old summarization path. New
// threads keep the summary in ThreadState.summary_text instead.
if (m.name === "summary" && m.type === "human") {
summarizedRef.current?.add(m.id ?? "");
}
}
const transientMessages = computeSummarizationTransientMessages(
messagesRef.current,
_messages,
summarizedRef.current ?? new Set<string>(),
renderedMessageSnapshotRef.current.threadId === threadIdRef.current
? renderedMessageSnapshotRef.current.messages
: EMPTY_MESSAGES,
);
transientHistoryOrderRef.current = mergeTransientHistoryBridgeOrder(
transientHistoryOrderRef.current,
transientMessages,
);
transientHistoryBridgeRef.current = mergeTransientHistoryBridge(
transientHistoryBridgeRef.current,
transientMessages,
);
transientHistoryThreadIdRef.current = threadIdRef.current;
messagesRef.current = [];
}
const updates: Array<Partial<AgentThreadState> | null> = Object.values(
data || {},
);
for (const update of updates) {
if (update && "title" in update && update.title) {
const currentThreadId = threadIdRef.current;
if (currentThreadId) {
setThreadTitleInCaches(queryClient, currentThreadId, update.title);
}
}
}
},
onCustomEvent(event: unknown) {
// Narrow `event.type` once; taskEventToSubtaskUpdate already validated the
// task_* events, so the per-branch re-narrowing below reads this single
// source of truth instead of re-checking the object shape each time.
const eventType =
typeof event === "object" && event !== null && "type" in event
? (event as { type: unknown }).type
: undefined;
if (eventType === "stream_replay_gap") {
setOptimisticMessages([]);
setOptimisticThreadId(null);
setLiveMessagesThreadId(null);
setPendingSupersededRunIds(new Set());
setPendingSupersededMessageIds(new Set());
messagesRef.current = [];
transientHistoryBridgeRef.current = [];
transientHistoryOrderRef.current = [];
transientHistoryThreadIdRef.current = null;
summarizedRef.current = new Set<string>();
pendingUsageBaselineMessageIdsRef.current = new Set();
localTurnAnchorRef.current = null;
tasksRef.current = {};
setTasks({});
invalidateStoppedThreadCaches(queryClient, threadIdRef.current, isMock);
toast.warning(t.conversation.streamReplayGap);
return;
}
const taskUpdate = taskEventToSubtaskUpdate(event);
if (taskUpdate) {
updateSubtask(taskUpdate);
}
if (eventType === "task_running") {
const e = event as {
type: "task_running";
task_id: string;
message: AIMessage;
message_index?: number;
};
// Accumulate the full step history instead of overwriting (#3779): keep
// latestMessage for the collapsed-header tool-call hint, and append the
// normalized step (assistant turn or tool output) to the timeline.
updateSubtask({
id: e.task_id,
latestMessage: e.message,
steps: [messageToStep(e.message, e.message_index ?? 0)],
});
return;
}
if (eventType === "llm_retry") {
const e = event as { type: "llm_retry"; message?: unknown };
if (typeof e.message === "string" && e.message.trim()) {
toast(e.message);
}
}
},
onError(error, run) {
scheduleActiveRunRejoinRetry(run);
setOptimisticMessages([]);
setOptimisticThreadId(null);
setLiveMessagesThreadId(null);
pendingPreparedReplayRef.current = null;
setPendingSupersededRunIds(new Set());
setPendingSupersededMessageIds(new Set());
toast.error(getStreamErrorMessage(error));
pendingUsageBaselineMessageIdsRef.current = new Set(
messagesRef.current
.map(messageIdentity)
.filter((id): id is string => Boolean(id)),
);
if (threadIdRef.current && !isMock) {
void queryClient.invalidateQueries({
queryKey: threadHistoryQueryKey(threadIdRef.current),
});
void queryClient.invalidateQueries({
queryKey: threadTokenUsageQueryKey(threadIdRef.current),
});
}
},
onFinish(state, run) {
if (run) {
completedRunIdsRef.current.add(run.run_id);
}
settleActiveRunRejoin();
listeners.current.onFinish?.(state.values);
pendingPreparedReplayRef.current = null;
pendingUsageBaselineMessageIdsRef.current = new Set(
messagesRef.current
.map(messageIdentity)
.filter((id): id is string => Boolean(id)),
);
invalidateStoppedThreadCaches(queryClient, threadIdRef.current, isMock);
},
});
const { isLoading: isThreadLoading, joinStream } = thread;
// reconnectOnMount only knows the run id stored in this tab's
// sessionStorage. A reopened browser or a new tab has no pointer, so recover
// the newest active run from the server and join its resumable SSE stream.
useEffect(() => {
const resolvedThreadId = onStreamThreadId ?? null;
const resolvedRunId = activeRunId ?? null;
let rejoin = activeRunRejoinRef.current;
if (
rejoin.threadId !== resolvedThreadId ||
rejoin.runId !== resolvedRunId
) {
if (rejoin.retryTimer !== null) {
clearTimeout(rejoin.retryTimer);
}
if (rejoin.attempts > 0 && rejoin.threadId && rejoin.runId) {
clearReconnectRun(rejoin.threadId, rejoin.runId);
}
rejoin = createActiveRunRejoinState(resolvedThreadId, resolvedRunId);
activeRunRejoinRef.current = rejoin;
}
if (
!resolvedThreadId ||
!resolvedRunId ||
completedRunIdsRef.current.has(resolvedRunId) ||
rejoin.inFlight ||
rejoin.retryTimer !== null ||
rejoin.settled ||
rejoin.attempts >= MAX_ACTIVE_RUN_REJOIN_ATTEMPTS ||
isThreadLoading
) {
return;
}
// A matching pointer means the SDK's native same-tab reconnect owns this
// run. Do not create a second SSE consumer; if that reconnect fails,
// scheduleActiveRunRejoinRetry releases the pointer.
if (readReconnectRun(resolvedThreadId) === resolvedRunId) {
return;
}
rejoin.attempts += 1;
rejoin.inFlight = true;
rememberReconnectRun(resolvedThreadId, resolvedRunId);
void joinStream(resolvedRunId);
}, [
activeRunId,
activeRunRejoinRetry,
isThreadLoading,
joinStream,
onStreamThreadId,
]);
useEffect(
() => () => {
const rejoin = activeRunRejoinRef.current;
if (rejoin.threadId !== (onStreamThreadId ?? null)) {
return;
}
if (rejoin.retryTimer !== null) {
clearTimeout(rejoin.retryTimer);
}
if (rejoin.attempts > 0 && rejoin.threadId && rejoin.runId) {
clearReconnectRun(rejoin.threadId, rejoin.runId);
}
activeRunRejoinRef.current = createActiveRunRejoinState();
},
[onStreamThreadId],
);
const stopThread = useCallback(async () => {
const stoppedThreadId =
threadIdRef.current ?? displayThreadId ?? threadId ?? null;
const pendingReplay = pendingPreparedReplayRef.current;
await stopThreadAndInvalidateCaches(
queryClient,
() => thread.stop(),
stoppedThreadId,
isMock,
);
if (pendingReplay) {
setOptimisticMessages([]);
setOptimisticThreadId(null);
clearPreparedReplayMasks(pendingReplay);
}
}, [
clearPreparedReplayMasks,
displayThreadId,
isMock,
queryClient,
thread,
threadId,
]);
const hasVisibleStreamState =
Boolean(threadId) || liveMessagesThreadId === currentViewThreadId;
const persistedMessages = useMemo(() => {
if (!hasVisibleStreamState) {
return EMPTY_MESSAGES;
}
const filtered = thread.messages.filter(
(message) => !message.id || !pendingSupersededMessageIds.has(message.id),
);
// The SDK getter mints a fresh [] on every read while the stream has no
// values; normalize to a stable identity so downstream effects keyed on
// this array cannot re-fire (and setState-loop) on idle renders.
return filtered.length === 0 ? EMPTY_MESSAGES : filtered;
}, [hasVisibleStreamState, pendingSupersededMessageIds, thread.messages]);
const visibleHistory = useMemo(
() => (threadId ? history : []),
[history, threadId],
);
// Render-facing coalesced snapshot. Optimistic-input confirmation and the
// turn anchor observe THIS snapshot — the same frames the user sees — so a
// human echo landing in the per-chunk SDK array one coalesce interval early
// can no longer withdraw the local input before the snapshot shows it.
// Refs, summarization capture, and token-usage tracking keep consuming the
// per-chunk `persistedMessages` array above, unchanged.
const renderMessages = useCoalescedStreamMessages(
persistedMessages,
thread.isLoading,
);
const humanMessageCount = renderMessages.filter(
(m) => m.type === "human",
).length;
const latestMessageCountsRef = useRef({ humanMessageCount });
const sendInFlightRef = useRef(false);
const messagesRef = useRef<Message[]>([]);
// Non-null only after a turn submitted by this mounted client. Keep it after
// finish/stop/error because the SDK can retain its transient event order in
// the settled frame. The next local submit replaces it and a thread switch or
// replay gap clears it. An empty set is meaningful for a new thread and must
// not be confused with a reconnect that has no local turn anchor.
const localTurnAnchorRef = useRef<LocalTurnAnchor | null>(null);
// Current-stream lifecycle bridge for messages removed from the checkpoint
// tail before the canonical run-event page refetch observes the journal
// flush. It is never appended into useThreadHistory's persisted pages.
const transientHistoryBridgeRef = useRef<Message[]>([]);
// Full identity order of each captured checkpoint. Confirmed bridge entries
// are pruned from the message buffer, but remain here as non-rendering
// anchors so an older rescue can be placed before a newest-first page.
const transientHistoryOrderRef = useRef<readonly string[]>([]);
const transientHistoryThreadIdRef = useRef<string | null>(null);
// The run-scoped committed display ledger supplies message objects when a
// compaction replacement outruns React or several rolling checkpoint windows
// pass between compactions. The merged order separately anchors those
// objects against canonical history.
const renderedMessageSnapshotRef = useRef<{
threadId: string | null;
messages: Message[];
order: readonly string[];
}>({
threadId: null,
messages: EMPTY_MESSAGES,
order: EMPTY_MESSAGE_IDENTITIES,
});
const summarizedRef = useRef<Set<string>>(null);
// Track human message count before sending to prevent clearing optimistic
// messages before the server's human message arrives (e.g. when AI messages
// from "messages-tuple" events arrive before the input human message from
// "values" events).
const prevHumanMsgCountRef = useRef(humanMessageCount);
latestMessageCountsRef.current = { humanMessageCount };
summarizedRef.current ??= new Set<string>();
// Reset thread-local pending UI state when switching between threads so
// optimistic messages and in-flight guards do not leak across chat views.
useEffect(() => {
startedRef.current = false;
sendInFlightRef.current = false;
messagesRef.current = [];
transientHistoryBridgeRef.current = [];
transientHistoryOrderRef.current = [];
transientHistoryThreadIdRef.current = null;
renderedMessageSnapshotRef.current = {
threadId: null,
messages: EMPTY_MESSAGES,
order: EMPTY_MESSAGE_IDENTITIES,
};
summarizedRef.current = new Set<string>();
pendingUsageBaselineMessageIdsRef.current = new Set();
pendingPreparedReplayRef.current = null;
setPendingSupersededRunIds(new Set());
setPendingSupersededMessageIds(new Set());
prevHumanMsgCountRef.current =
latestMessageCountsRef.current.humanMessageCount;
}, [threadId]);
// Confirming a new thread only assigns its SDK id; the displayed
// conversation and its submitted human anchor have not changed.
useEffect(() => {
localTurnAnchorRef.current = null;
}, [currentViewThreadId]);
// Release entries individually once canonical history confirms their stable
// identities. Keep unconfirmed entries across failure/refetch within this
// page lifecycle so a temporary persistence gap cannot hide a turn.
useEffect(() => {
transientHistoryBridgeRef.current = pruneConfirmedTransientMessages(
transientHistoryBridgeRef.current,
visibleHistory,
);
if (transientHistoryBridgeRef.current.length === 0) {
transientHistoryOrderRef.current = [];
transientHistoryThreadIdRef.current = null;
}
}, [visibleHistory]);
useEffect(() => {
if (optimisticThreadId && optimisticThreadId !== currentViewThreadId) {
setOptimisticMessages([]);
setOptimisticThreadId(null);
}
if (liveMessagesThreadId && liveMessagesThreadId !== currentViewThreadId) {
setLiveMessagesThreadId(null);
}
}, [currentViewThreadId, liveMessagesThreadId, optimisticThreadId]);
// When streaming starts without a baseline (e.g. reconnection, run started
// from another client, or page reload mid-stream), snapshot the current
// messages so only *new* messages are treated as "pending" for token usage.
useEffect(() => {
if (
thread.isLoading &&
pendingUsageBaselineMessageIdsRef.current.size === 0
) {
pendingUsageBaselineMessageIdsRef.current = new Set(
persistedMessages
.map(messageIdentity)
.filter((id): id is string => Boolean(id)),
);
}
}, [persistedMessages, thread.isLoading]);
// Clear optimistic when server messages arrive.
// For messages with a human optimistic message, wait until the server's
// human message has arrived in the RENDER SNAPSHOT — identity match first,
// human-count growth of the rendered frames as fallback for runtime-re-keyed
// first turns — never in the unthrottled per-chunk array, which would
// withdraw the local input one coalesce interval before the user can see
// its confirmed copy.
const optimisticMessageCount = optimisticMessages.length;
const hasHumanOptimistic = optimisticMessages.some((m) => m.type === "human");
useEffect(() => {
if (optimisticMessageCount === 0) return;
const newHumanMsgArrived = humanMessageCount > prevHumanMsgCountRef.current;
if (!hasHumanOptimistic || newHumanMsgArrived) {
setOptimisticMessages([]);
setOptimisticThreadId(null);
}
}, [hasHumanOptimistic, humanMessageCount, optimisticMessageCount]);
useEffect(() => {
if (
optimisticMessageCount > 0 &&
areOptimisticMessagesConfirmed(optimisticMessages, renderMessages)
) {
setOptimisticMessages([]);
setOptimisticThreadId(null);
}
}, [optimisticMessageCount, optimisticMessages, renderMessages]);
const sendMessage = useCallback(
async (
threadId: string,
message: PromptInputMessage,
extraContext?: Record<string, unknown>,
options?: SendMessageOptions,
) => {
if (sendInFlightRef.current) {
return;
}
sendInFlightRef.current = true;
const text = message.text.trim();
// Capture the current human message count before showing optimistic
// messages so we can wait for the server's copy of the user input.
prevHumanMsgCountRef.current = humanMessageCount;
pendingUsageBaselineMessageIdsRef.current = new Set(
persistedMessages
.map(messageIdentity)
.filter((id): id is string => Boolean(id)),
);
// One client-generated id for this turn's human input: the optimistic
// display copy and the submitted message share it, so the render
// snapshot confirms the exact identity it already shows instead of the
// ordering repair guessing from the baseline (a compaction-trimmed
// checkpoint must never promote an older history-only human into this
// turn's anchor).
const hideFromUI = options?.additionalKwargs?.hide_from_ui === true;
const humanMessageId = `local-human-${uuid()}`;
localTurnAnchorRef.current = {
threadId,
humanIdentity: hideFromUI ? null : `message:${humanMessageId}`,
baselineIdentities: new Set(pendingUsageBaselineMessageIdsRef.current),
preSubmitHistoryIdentities: new Set(
visibleHistory.map(messageIdentity).filter(isNonEmptyString),
),
preSubmitBridgeIdentities: new Set(
transientHistoryThreadIdRef.current === threadId
? transientHistoryBridgeRef.current
.map(messageIdentity)
.filter(isNonEmptyString)
: EMPTY_MESSAGE_IDENTITIES,
),
preSubmitMaxSeq: maxMessageSeq([
...visibleHistory,
...persistedMessages,
]),
};
// Build optimistic files list with uploading status
const optimisticFiles: FileInMessage[] = (message.files ?? []).map(
(f) => ({
filename: f.filename ?? "",
size: 0,
status: "uploading" as const,
}),
);
const newOptimistic: Message[] = [];
if (!hideFromUI) {
newOptimistic.push({
type: "human",
id: humanMessageId,
content: text ? [{ type: "text", text }] : "",
additional_kwargs: buildHumanMessageAdditionalKwargs(
options?.additionalKwargs,
optimisticFiles,
),
});
}
if (optimisticFiles.length > 0 && !hideFromUI) {
// Mock AI message while files are being uploaded
newOptimistic.push({
type: "ai",
id: `opt-ai-${Date.now()}`,
content: t.uploads.uploadingFiles,
additional_kwargs: { element: "task" },
});
}
setOptimisticThreadId(threadId);
setLiveMessagesThreadId(threadId);
setOptimisticMessages(newOptimistic);
listeners.current.onSend?.(threadId);
// Shared by the optimistic bubble and the submit, so the files the
// user sees are exactly the files that are sent.
let uploadedFiles: FileInMessage[] = [];
try {
// Upload files first if any
if (message.files && message.files.length > 0) {
setIsUploading(true);
try {
const filePromises = message.files.map((fileUIPart) =>
promptInputFilePartToFile(fileUIPart),
);
const conversionResults = await Promise.all(filePromises);
const files = conversionResults.filter(
(file): file is File => file !== null,
);
const failedConversions = conversionResults.length - files.length;
if (failedConversions > 0) {
throw new Error(
`Failed to prepare ${failedConversions} attachment(s) for upload. Please retry.`,
);
}
if (!threadId) {
throw new Error("Thread is not ready for file upload.");
}
if (files.length > 0) {
const uploadResponse = await uploadFiles(threadId, files);
uploadedFiles = uploadResponse.files.map((info) => ({
filename: info.filename,
size: info.size,
path: info.virtual_path,
status: "uploaded" as const,
}));
// Update optimistic human message with uploaded status + paths
setOptimisticMessages((messages) => {
if (messages.length > 1 && messages[0]) {
const humanMessage: Message = messages[0];
return [
{
...humanMessage,
additional_kwargs: buildHumanMessageAdditionalKwargs(
options?.additionalKwargs,
uploadedFiles,
),
},
...messages.slice(1),
];
}
return messages;
});
}
} catch (error) {
const errorMessage =
error instanceof Error
? error.message
: "Failed to upload files.";
toast.error(errorMessage);
setOptimisticMessages([]);
setOptimisticThreadId(null);
setLiveMessagesThreadId(null);
throw error;
} finally {
setIsUploading(false);
}
}
// Attachments are uploaded, so the send is genuinely dispatched and
// callers can drop one-time state. A failed upload rejected above
// without reaching here, leaving that state for a retry.
options?.onSent?.();
await thread.submit(
{
messages: buildThreadSubmitMessages({
text,
additionalKwargs: options?.additionalKwargs,
additionalInputMessages: options?.additionalInputMessages,
filesForSubmit: uploadedFiles,
humanMessageId,
}),
},
{
threadId: threadId,
// No streamSubgraphs: subtask progress arrives via root-namespace
// custom events, while subgraph frames would leak a delegated
// subagent's values/messages into the thread view (#4399).
streamResumable: true,
config: {
recursion_limit: 1000,
},
context: buildRunContext({
settings: context,
threadId,
extraContext,
conversationReferences: options?.conversationReferences,
}),
},
);
void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
void queryClient.invalidateQueries({
queryKey: INFINITE_THREADS_QUERY_KEY_PREFIX,
});
} catch (error) {
setOptimisticMessages([]);
setOptimisticThreadId(null);
setLiveMessagesThreadId(null);
setIsUploading(false);
localTurnAnchorRef.current = null;
throw error;
} finally {
sendInFlightRef.current = false;
}
},
[
thread,
t.uploads.uploadingFiles,
context,
queryClient,
humanMessageCount,
persistedMessages,
visibleHistory,
],
);
const submitPreparedReplay = useCallback(
async <TPrepared extends RegeneratePrepareResponse>({
threadId,
prepare,
getSupersededMessageIds,
getOptimisticMessages,
}: {
threadId: string;
prepare: () => Promise<TPrepared>;
getSupersededMessageIds: (prepared: TPrepared) => string[];
getOptimisticMessages?: (prepared: TPrepared) => Message[];
}) => {
if (sendInFlightRef.current || !threadId) {
return false;
}
sendInFlightRef.current = true;
prevHumanMsgCountRef.current = humanMessageCount;
pendingUsageBaselineMessageIdsRef.current = new Set(
persistedMessages
.map(messageIdentity)
.filter((id): id is string => Boolean(id)),
);
localTurnAnchorRef.current = {
threadId,
// Replay turns submit no new visible human; an edit replay adopts the
// prepare response's replacement identity once it lands below.
humanIdentity: null,
baselineIdentities: new Set(pendingUsageBaselineMessageIdsRef.current),
preSubmitHistoryIdentities: new Set(
visibleHistory.map(messageIdentity).filter(isNonEmptyString),
),
preSubmitBridgeIdentities: new Set(
transientHistoryThreadIdRef.current === threadId
? transientHistoryBridgeRef.current
.map(messageIdentity)
.filter(isNonEmptyString)
: EMPTY_MESSAGE_IDENTITIES,
),
preSubmitMaxSeq: maxMessageSeq([
...visibleHistory,
...persistedMessages,
]),
};
setLiveMessagesThreadId(threadId);
listeners.current.onSend?.(threadId);
let preparedSupersededRunId: string | null = null;
let preparedSupersededMessageIds: string[] = [];
try {
const prepared = await prepare();
preparedSupersededRunId = prepared.target_run_id;
preparedSupersededMessageIds = getSupersededMessageIds(prepared);
prevHumanMsgCountRef.current = countHumanMessagesExcludingSuperseded(
persistedMessages,
preparedSupersededMessageIds,
);
const replacementHumanMessageId =
"replacement_human_message_id" in prepared &&
typeof prepared.replacement_human_message_id === "string"
? prepared.replacement_human_message_id
: undefined;
const replayAnchor = localTurnAnchorRef.current;
if (replayAnchor?.threadId === threadId && replacementHumanMessageId) {
// The edit replay reuses the server-prepared replacement identity;
// supersede semantics stay with the prepare response.
localTurnAnchorRef.current = {
...replayAnchor,
humanIdentity: `message:${replacementHumanMessageId}`,
};
}
const pendingReplay: PendingPreparedReplayMask = {
kind: replacementHumanMessageId ? "edit" : "regenerate",
targetRunId: prepared.target_run_id,
supersededMessageIds: preparedSupersededMessageIds,
replacementHumanMessageId,
};
pendingPreparedReplayRef.current = pendingReplay;
setPendingSupersededRunIds((current) => {
const next = new Set(current);
next.add(prepared.target_run_id);
return next;
});
setPendingSupersededMessageIds((current) => {
const next = new Set(current);
for (const id of preparedSupersededMessageIds) {
next.add(id);
}
return next;
});
const nextOptimisticMessages = getOptimisticMessages?.(prepared) ?? [];
if (nextOptimisticMessages.length > 0) {
setOptimisticThreadId(threadId);
setOptimisticMessages(nextOptimisticMessages);
}
await thread.submit(prepared.input, {
threadId,
checkpoint: prepared.checkpoint,
metadata: prepared.metadata,
// No streamSubgraphs — same contract as the main submit path (#4399).
streamResumable: true,
config: {
recursion_limit: 1000,
},
// Replaying a turn never carries conversation references: the grant
// is per send, so a regenerate or edit runs without them unless the
// user attaches them again.
context: buildRunContext({ settings: context, threadId }),
});
void queryClient.invalidateQueries({ queryKey: ["thread", threadId] });
void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
void queryClient.invalidateQueries({
queryKey: INFINITE_THREADS_QUERY_KEY_PREFIX,
});
void queryClient.invalidateQueries({
queryKey: threadHistoryQueryKey(threadId),
});
void queryClient.invalidateQueries({
queryKey: threadTokenUsageQueryKey(threadId),
});
return true;
} catch (error) {
setOptimisticMessages([]);
setOptimisticThreadId(null);
setLiveMessagesThreadId(null);
localTurnAnchorRef.current = null;
if (preparedSupersededRunId) {
const supersededRunId = preparedSupersededRunId;
pendingPreparedReplayRef.current = null;
setPendingSupersededRunIds((current) =>
removeSetItems(current, [supersededRunId]),
);
setPendingSupersededMessageIds((current) =>
removeSetItems(current, preparedSupersededMessageIds),
);
}
toast.error(getStreamErrorMessage(error));
return false;
} finally {
sendInFlightRef.current = false;
}
},
[
context,
humanMessageCount,
persistedMessages,
queryClient,
thread,
visibleHistory,
],
);
const regenerateMessage = useCallback(
async (
threadId: string,
messageId: string,
supersededMessageIds: string[] = [messageId],
) => {
if (!messageId) {
return false;
}
return submitPreparedReplay({
threadId,
prepare: async () => {
const response = await fetch(
`${getBackendBaseURL()}/api/threads/${encodeURIComponent(
threadId,
)}/runs/regenerate/prepare`,
{
method: "POST",
headers: {
"Content-Type": "application/json",
},
credentials: "include",
body: JSON.stringify({ message_id: messageId }),
},
);
if (!response.ok) {
throw new Error(await readResponseErrorMessage(response));
}
return (await response.json()) as RegeneratePrepareResponse;
},
getSupersededMessageIds: () => supersededMessageIds,
});
},
[submitPreparedReplay],
);
const editAndRegenerateMessage = useCallback(
async (
threadId: string,
humanMessageId: string,
replacementText: string,
additionalKwargs?: Record<string, unknown>,
) => {
if (!humanMessageId) {
return false;
}
return submitPreparedReplay<EditRegeneratePrepareResponse>({
threadId,
prepare: async () => {
const response = await fetch(
`${getBackendBaseURL()}/api/threads/${encodeURIComponent(
threadId,
)}/runs/edit-regenerate/prepare`,
{
method: "POST",
headers: {
"Content-Type": "application/json",
},
credentials: "include",
body: JSON.stringify({
human_message_id: humanMessageId,
replacement_text: replacementText,
}),
},
);
if (!response.ok) {
throw new Error(await readResponseErrorMessage(response));
}
const prepared =
(await response.json()) as EditRegeneratePrepareResponse;
if (!additionalKwargs || !Array.isArray(prepared.input.messages)) {
return prepared;
}
const messages = [...prepared.input.messages];
for (let index = messages.length - 1; index >= 0; index -= 1) {
const message = messages[index];
if (message?.type !== "human") continue;
messages[index] = {
...message,
additional_kwargs: {
...message.additional_kwargs,
...additionalKwargs,
},
};
break;
}
return { ...prepared, input: { ...prepared.input, messages } };
},
getSupersededMessageIds: (prepared) => prepared.source_message_ids,
getOptimisticMessages: (prepared) => prepared.input.messages ?? [],
});
},
[submitPreparedReplay],
);
// Cache the latest thread messages in a ref to compare against incoming history messages for deduplication,
// and to allow access to the full message list in onUpdateEvent without causing re-renders.
if (persistedMessages.length >= messagesRef.current.length) {
messagesRef.current = persistedMessages;
}
const rawVisibleOptimisticMessages = getVisibleOptimisticMessages(
optimisticThreadId === currentViewThreadId ? optimisticMessages : [],
prevHumanMsgCountRef.current,
humanMessageCount,
);
const visibleOptimisticMessages =
rawVisibleOptimisticMessages.length === 0
? EMPTY_MESSAGES
: rawVisibleOptimisticMessages;
const transientHistoryOrder =
transientHistoryBridgeRef.current.length > 0 &&
transientHistoryThreadIdRef.current === threadId
? mergeTransientHistoryBridgeOrder(
transientHistoryOrderRef.current,
persistedMessages,
)
: transientHistoryOrderRef.current;
const previouslyRenderedOrder =
renderedMessageSnapshotRef.current.threadId === threadId
? renderedMessageSnapshotRef.current.order
: EMPTY_MESSAGE_IDENTITIES;
// Commit the extended non-rendering order skeleton after React commits this
// render. The local value above keeps this render correctly anchored without
// mutating a ref during render.
useEffect(() => {
if (
transientHistoryBridgeRef.current.length > 0 &&
transientHistoryThreadIdRef.current === threadId
) {
transientHistoryOrderRef.current = mergeTransientHistoryBridgeOrder(
transientHistoryOrderRef.current,
persistedMessages,
);
}
}, [persistedMessages, threadId]);
// The transient-bridge refs mutate in lockstep with stream/history updates
// already captured by these deps, and resolveTransientHistoryBridge is
// idempotent for entries canonical history has absorbed, so memoizing on the
// coalesced snapshot cannot pin a stale bridge.
const mergedMessages = useMemo(() => {
const effectiveHistory = resolveThreadTransientHistoryBridge(
visibleHistory,
transientHistoryBridgeRef.current,
transientHistoryThreadIdRef.current,
threadId,
transientHistoryOrder,
previouslyRenderedOrder,
);
const merged = mergeMessages(
effectiveHistory,
renderMessages,
visibleOptimisticMessages,
);
const localTurnAnchor =
localTurnAnchorRef.current?.threadId === currentViewThreadId
? localTurnAnchorRef.current
: null;
const canonicalHistoryIdentities = new Set(
visibleHistory.map(messageIdentity).filter(isNonEmptyString),
);
// Only established history known to predate this local submit may be moved
// across its human anchor. The fixed identity snapshots cover messages
// already loaded from REST and pre-existing transient-bridge rescue; the
// authoritative seq boundary also admits older pages that finish loading
// after submit. Post-submit rescue and later external turns stay outside.
const confirmedHistoryIdentities = getConfirmedPreSubmitHistoryIdentities(
visibleHistory,
localTurnAnchor,
);
// The current turn's run(s): visible ai/tool steps that appear in the live
// checkpoint but are neither part of the pre-submit baseline nor already
// canonical REST history. These are output from the in-flight submit and
// must never be moved before their human.
const currentTurnRunIds = getCurrentTurnRunIds(
renderMessages,
localTurnAnchor ? localTurnAnchor.baselineIdentities : null,
canonicalHistoryIdentities,
);
if (localTurnAnchor?.humanIdentity) {
// The surviving merged copy of the submitted human can be the run_id-less
// optimistic one; recover the run from any rendered or canonical copy so
// an interrupt-flushed current-run step is still recognised as ours.
const anchorRunId =
findMessageRunIdByIdentity(
renderMessages,
localTurnAnchor.humanIdentity,
) ??
findMessageRunIdByIdentity(
effectiveHistory,
localTurnAnchor.humanIdentity,
);
if (anchorRunId) {
currentTurnRunIds.add(anchorRunId);
}
}
return localTurnAnchor === null
? restoreReconnectedTurnMessageOrder(merged)
: restoreLocalTurnMessageOrder(
merged,
localTurnAnchor.baselineIdentities,
confirmedHistoryIdentities,
currentTurnRunIds,
localTurnAnchor.humanIdentity,
canonicalHistoryIdentities,
);
}, [
currentViewThreadId,
previouslyRenderedOrder,
renderMessages,
threadId,
transientHistoryOrder,
visibleHistory,
visibleOptimisticMessages,
]);
useEffect(() => {
// The committed render ledger excludes hidden control copies and the
// still-unconfirmed optimistic ones (keyed by identity, since the local
// input now shares its id with the submit instead of an `opt-` prefix):
// a failed send must never pin a message the server never saw.
const pendingOptimisticIdentities = new Set(
(optimisticThreadId === currentViewThreadId
? optimisticMessages
: EMPTY_MESSAGES
)
.map(messageIdentity)
.filter(isNonEmptyString),
);
const visibleMergedMessages = mergedMessages.filter((message) => {
if (isHiddenFromUIMessage(message)) {
return false;
}
const identity = messageIdentity(message);
return (
identity === undefined || !pendingOptimisticIdentities.has(identity)
);
});
const previousLedger =
thread.isLoading &&
renderedMessageSnapshotRef.current.threadId === threadId
? renderedMessageSnapshotRef.current.messages
: EMPTY_MESSAGES;
const renderedMessageLedger = mergeRenderedMessageLedger(
previousLedger,
visibleMergedMessages,
pendingSupersededMessageIds,
);
renderedMessageSnapshotRef.current = {
threadId: threadId ?? null,
messages: renderedMessageLedger,
order: renderedMessageLedger
.map(messageIdentity)
.filter(isNonEmptyString),
};
}, [
mergedMessages,
optimisticMessages,
optimisticThreadId,
currentViewThreadId,
pendingSupersededMessageIds,
thread.isLoading,
threadId,
]);
const pendingUsageMessages = thread.isLoading
? getMessagesAfterBaseline(
persistedMessages,
pendingUsageBaselineMessageIdsRef.current,
)
: [];
// Merge history, live stream, and optimistic messages for display
// History messages may overlap with thread.messages; thread.messages take precedence
const mergedThread = {
...thread,
stop: stopThread,
values: hasVisibleStreamState ? thread.values : EMPTY_THREAD_VALUES,
messages: mergedMessages,
} as typeof thread;
return {
thread: mergedThread,
pendingUsageMessages,
sendMessage,
regenerateMessage,
editAndRegenerateMessage,
isUploading,
isHistoryLoading,
hasMoreHistory,
loadMoreHistory,
} as const;
}
type ThreadHistoryOptions = {
enabled?: boolean;
pendingSupersededRunIds?: ReadonlySet<string>;
};
export const THREAD_HISTORY_QUERY_POLICY = {
refetchOnWindowFocus: false,
staleTime: 5 * 60 * 1_000,
} as const;
export function useThreadHistory(
threadId: string,
{ enabled = true, pendingSupersededRunIds }: ThreadHistoryOptions = {},
) {
const historyQuery = useInfiniteQuery<
ThreadMessagesPageResponse,
Error,
InfiniteData<ThreadMessagesPageResponse>,
ReturnType<typeof threadHistoryQueryKey>,
number | null
>({
...THREAD_HISTORY_QUERY_POLICY,
queryKey: threadHistoryQueryKey(threadId),
enabled: enabled && Boolean(threadId),
initialPageParam: null,
queryFn: async ({ pageParam, signal }) => {
const url = buildThreadMessagesPageUrl(
getBackendBaseURL(),
threadId,
pageParam ?? undefined,
);
const response = await fetch(url, {
method: "GET",
headers: {
"Content-Type": "application/json",
},
credentials: "include",
signal,
});
if (!response.ok) {
throw new Error(
await readResponseErrorMessage(
response,
"Failed to load thread history.",
),
);
}
return parseThreadMessagesPageResponse(await response.json());
},
getNextPageParam: getThreadHistoryNextPageParam,
});
const currentMessageRows = useMemo(
() => flattenThreadHistoryPages(historyQuery.data?.pages ?? []),
[historyQuery.data?.pages],
);
const [retainedHistory, setRetainedHistory] = useState<{
threadId: string;
rows: RunMessage[];
}>({ threadId, rows: EMPTY_RUN_MESSAGES });
const previousRows =
retainedHistory.threadId === threadId
? retainedHistory.rows
: EMPTY_RUN_MESSAGES;
const pages = historyQuery.data?.pages ?? [];
const isAuthoritativeComplete =
historyQuery.isSuccess &&
!historyQuery.isFetching &&
pages.length > 0 &&
pages.at(-1)?.has_more === false;
const messageRows = useMemo(
() =>
reconcileThreadHistoryRows(
previousRows,
currentMessageRows,
isAuthoritativeComplete,
),
[currentMessageRows, isAuthoritativeComplete, previousRows],
);
useEffect(() => {
setRetainedHistory((current) => {
if (current.threadId === threadId && current.rows === messageRows) {
return current;
}
return { threadId, rows: messageRows };
});
}, [messageRows, threadId]);
const messages = useMemo(() => {
return buildVisibleHistoryMessages(
messageRows,
pendingSupersededRunIds ?? new Set<string>(),
);
}, [messageRows, pendingSupersededRunIds]);
useEffect(() => {
if (historyQuery.error) {
console.error(historyQuery.error);
toast.error("Failed to load thread history.");
}
}, [historyQuery.error]);
return {
messages,
loading: historyQuery.isLoading || historyQuery.isFetchingNextPage,
hasMore: Boolean(historyQuery.hasNextPage),
loadMore: historyQuery.fetchNextPage,
};
}
export function useThreads(
params: ThreadSearchParams = DEFAULT_THREAD_SEARCH_PARAMS,
) {
const apiClient = getAPIClient();
return useQuery<AgentThread[]>({
...buildThreadsSearchQueryOptions(apiClient, params),
});
}
export const INFINITE_THREADS_PAGE_SIZE = 50;
export const INFINITE_THREADS_QUERY_KEY_PREFIX = [
"threads",
"searchInfinite",
] as const;
type InfiniteThreadsParams = Omit<
NonNullable<Parameters<ThreadsClient["search"]>[0]>,
"limit" | "offset"
> & { archived?: boolean };
type InfiniteThreadsSearchClient = {
threads: {
search: ThreadsClient["search"];
};
};
/**
* One loaded page of an infinite thread list. `threads` holds the rows the
* list shows; `hiddenCount` counts the server rows the page consumed but the
* client filtered out (sidecar threads). Both live in the page data itself so
* cache writes and structural sharing, which rebuild pages, keep them.
*/
export type InfiniteThreadsPage = {
threads: AgentThread[];
hiddenCount: number;
/** False once the server returned fewer rows than requested. */
hasMore: boolean;
};
export type InfiniteThreadsData = InfiniteData<InfiniteThreadsPage, number>;
export async function fetchInfiniteThreadsPage(
apiClient: InfiniteThreadsSearchClient,
params: InfiniteThreadsParams,
pageParam: number,
pageSize: number = INFINITE_THREADS_PAGE_SIZE,
): Promise<InfiniteThreadsPage> {
const threads: AgentThread[] = [];
let offset = pageParam;
let hasMore = true;
while (threads.length < pageSize) {
const currentLimit = pageSize - threads.length;
const response =
params.archived === undefined
? ((await apiClient.threads.search<AgentThreadState>({
...params,
limit: currentLimit,
offset,
})) as AgentThread[])
: await searchThreadsByArchive({
...params,
archived: params.archived,
metadata: params.metadata ?? undefined,
limit: currentLimit,
offset,
});
threads.push(...filterThreadSearchResults(response, params));
offset += response.length;
if (response.length < currentLimit) {
hasMore = false;
break;
}
}
return {
threads,
hiddenCount: offset - pageParam - threads.length,
hasMore,
};
}
export function getInfiniteThreadsNextPageParam(
lastPage: InfiniteThreadsPage,
allPages: InfiniteThreadsPage[],
): number | undefined {
if (!lastPage.hasMore) {
return undefined;
}
// Count visible rows from the cache instead of freezing the fetch-time
// offset: a local delete or insert changes the server list the same way.
return allPages.reduce(
(offset, page) => offset + page.threads.length + page.hiddenCount,
0,
);
}
/** List components read plain thread pages; pagination state stays cached. */
function selectInfiniteThreads(
data: InfiniteThreadsData,
): InfiniteData<AgentThread[], number> {
return {
pages: data.pages.map((page) => page.threads),
pageParams: data.pageParams,
};
}
function updateInfiniteThreadsPages(
oldData: InfiniteThreadsData | undefined,
update: (threads: AgentThread[]) => AgentThread[],
): InfiniteThreadsData | undefined {
if (!oldData) {
return oldData;
}
return {
...oldData,
pages: oldData.pages.map((page) => ({
...page,
threads: update(page.threads),
})),
};
}
export function mapInfiniteThreadsCache(
oldData: InfiniteThreadsData | undefined,
mapper: (thread: AgentThread) => AgentThread,
): InfiniteThreadsData | undefined {
return updateInfiniteThreadsPages(oldData, (threads) => threads.map(mapper));
}
export function filterInfiniteThreadsCache(
oldData: InfiniteThreadsData | undefined,
predicate: (thread: AgentThread) => boolean,
): InfiniteThreadsData | undefined {
return updateInfiniteThreadsPages(oldData, (threads) =>
threads.filter(predicate),
);
}
function mergeThreadMetadata(
thread: AgentThread,
metadata: ThreadMetadataPatch,
): AgentThread {
return {
...thread,
metadata: {
...(thread.metadata ?? {}),
...metadata,
},
};
}
function mergeThreadTitle(thread: AgentThread, title: string): AgentThread {
return {
...thread,
values: {
...thread.values,
title,
},
};
}
function getThreadSnapshotCacheFilters(threadId: string) {
return {
search: {
queryKey: ["threads", "search"],
exact: false,
},
infiniteSearch: {
queryKey: INFINITE_THREADS_QUERY_KEY_PREFIX,
exact: false,
},
metadata: {
queryKey: ["thread", "metadata", threadId],
exact: false,
},
} as const;
}
function setThreadInCaches(
queryClient: QueryClient,
threadId: string,
mapper: (thread: AgentThread) => AgentThread,
): void {
const filters = getThreadSnapshotCacheFilters(threadId);
queryClient.setQueriesData(
filters.search,
(oldData: Array<AgentThread> | undefined) => {
if (!oldData) {
return oldData;
}
return oldData.map((thread) =>
thread.thread_id === threadId ? mapper(thread) : thread,
);
},
);
queryClient.setQueriesData(
filters.infiniteSearch,
(oldData: InfiniteThreadsData | undefined) =>
mapInfiniteThreadsCache(oldData, (thread) =>
thread.thread_id === threadId ? mapper(thread) : thread,
),
);
queryClient.setQueriesData(
filters.metadata,
(oldData: AgentThread | null | undefined) =>
oldData ? mapper(oldData) : oldData,
);
}
export function setThreadMetadataInCaches(
queryClient: QueryClient,
threadId: string,
metadata: ThreadMetadataPatch,
) {
setThreadInCaches(queryClient, threadId, (thread) =>
mergeThreadMetadata(thread, metadata),
);
}
export function setThreadTitleInCaches(
queryClient: QueryClient,
threadId: string,
title: string,
): void {
setThreadInCaches(queryClient, threadId, (thread) =>
mergeThreadTitle(thread, title),
);
}
export function useInfiniteThreads(
params: InfiniteThreadsParams = {
sortBy: "updated_at",
sortOrder: "desc",
select: ["thread_id", "updated_at", "values", "metadata"],
},
{ enabled = true }: { enabled?: boolean } = {},
) {
const apiClient = getAPIClient();
return useInfiniteQuery<
InfiniteThreadsPage,
Error,
InfiniteData<AgentThread[], number>,
readonly unknown[],
number
>({
queryKey: [...INFINITE_THREADS_QUERY_KEY_PREFIX, params],
initialPageParam: 0,
queryFn: async ({ pageParam }) =>
fetchInfiniteThreadsPage(
apiClient,
params,
pageParam,
INFINITE_THREADS_PAGE_SIZE,
),
getNextPageParam: (lastPage, allPages) =>
getInfiniteThreadsNextPageParam(lastPage, allPages),
select: selectInfiniteThreads,
refetchOnWindowFocus: false,
enabled,
});
}
export function useThreadRuns(
threadId?: string,
{ enabled = true }: { enabled?: boolean } = {},
) {
const apiClient = getAPIClient();
return useQuery<Run[]>({
queryKey: ["thread", threadId],
queryFn: async () => {
if (!threadId) {
return [];
}
const response = await apiClient.runs.list(threadId);
return response;
},
enabled: enabled && Boolean(threadId),
refetchOnWindowFocus: false,
retry: false,
});
}
export function useThreadMetadata(
threadId?: string | null,
{
enabled = true,
isMock = false,
}: { enabled?: boolean; isMock?: boolean } = {},
) {
const apiClient = getAPIClient(isMock);
return useQuery<AgentThread | null>({
queryKey: ["thread", "metadata", threadId, isMock],
queryFn: async () => {
if (!threadId) {
return null;
}
try {
const response = await apiClient.threads.get(threadId);
return response as AgentThread;
} catch (error) {
if (isThreadMissingError(error)) {
return null;
}
throw error;
}
},
enabled: enabled && Boolean(threadId),
retry: false,
refetchOnWindowFocus: false,
});
}
export function useThreadTokenUsage(
threadId?: string | null,
{ enabled = true }: { enabled?: boolean } = {},
) {
return useQuery<ThreadTokenUsageResponse | null>({
queryKey: threadTokenUsageQueryKey(threadId),
queryFn: async () => {
if (!threadId) {
return null;
}
return fetchThreadTokenUsage(threadId);
},
enabled: enabled && Boolean(threadId),
retry: false,
refetchOnWindowFocus: false,
// Keep same-thread data visible during refetches without carrying usage
// from the previous route into a newly selected thread.
placeholderData: (previous) =>
retainThreadTokenUsagePlaceholder(previous, threadId),
});
}
export function useBranchThread() {
const queryClient = useQueryClient();
return useMutation({
mutationFn: async ({
threadId,
messageId,
messageIds,
title,
}: {
threadId: string;
messageId: string;
messageIds?: string[];
title?: string;
}) => branchThreadFromTurn(threadId, { messageId, messageIds, title }),
onSuccess(response, { threadId }) {
void queryClient.invalidateQueries({
queryKey: ["thread", "metadata", response.thread_id],
});
void queryClient.invalidateQueries({
queryKey: ["thread", "metadata", threadId],
});
void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
void queryClient.invalidateQueries({
queryKey: INFINITE_THREADS_QUERY_KEY_PREFIX,
});
},
});
}
export function usePinThread() {
const queryClient = useQueryClient();
return useMutation({
mutationFn: async ({
threadId,
pinned,
}: {
threadId: string;
pinned: boolean;
}) =>
patchThreadMetadata(threadId, {
[THREAD_PINNED_METADATA_KEY]: pinned,
}),
onSuccess(_response, { threadId, pinned }) {
setThreadMetadataInCaches(queryClient, threadId, {
[THREAD_PINNED_METADATA_KEY]: pinned,
});
},
onSettled() {
void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
void queryClient.invalidateQueries({
queryKey: INFINITE_THREADS_QUERY_KEY_PREFIX,
});
// Pin changes the ordering the project page thread list shows
// ([...PROJECTS_QUERY_KEY, "threads", id, ...]); without this, cached
// pages keep the old order and pagination can duplicate or skip
// entries across the refetch boundary.
void queryClient.invalidateQueries({
queryKey: [...PROJECTS_QUERY_KEY, "threads"],
});
},
});
}
export function useMoveThreadToProject(options?: {
onError?: (
error: Error,
variables: { threadId: string; projectId: string | null },
) => void;
}) {
const queryClient = useQueryClient();
return useMutation({
mutationFn: async ({
threadId,
projectId,
}: {
threadId: string;
projectId: string | null;
}) => moveThreadToProject(threadId, projectId),
// Hook-level error handler: survives the caller's dropdown unmounting,
// unlike a per-mutate `onError` passed from inside a closing menu.
onError: options?.onError,
async onSuccess(_response, { threadId, projectId }) {
// An older GET must not overwrite the confirmed affiliation. Match all
// metadata variants, including an initial read with no cached snapshot.
await queryClient.cancelQueries({
queryKey: ["thread", "metadata", threadId],
});
setThreadMetadataInCaches(queryClient, threadId, {
[THREAD_PROJECT_METADATA_KEY]: projectId,
});
await queryClient.invalidateQueries({
queryKey: ["thread", "metadata", threadId],
});
},
onSettled() {
void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
void queryClient.invalidateQueries({
queryKey: INFINITE_THREADS_QUERY_KEY_PREFIX,
});
// Moving a thread changes membership of project thread lists
// ([...PROJECTS_QUERY_KEY, "threads", id, ...]).
void queryClient.invalidateQueries({
queryKey: [...PROJECTS_QUERY_KEY, "threads"],
});
},
});
}
export function useRunDetail(threadId: string, runId: string) {
const apiClient = getAPIClient();
return useQuery<Run>({
queryKey: ["thread", threadId, "run", runId],
queryFn: async () => {
const response = await apiClient.runs.get(threadId, runId);
return response;
},
refetchOnWindowFocus: false,
});
}
async function deleteLocalThreadData(threadId: string) {
const response = await fetch(
`${getBackendBaseURL()}/api/threads/${encodeURIComponent(threadId)}`,
{
method: "DELETE",
},
);
// A 404 means the thread is already gone — the desired end state. The prior
// `apiClient.threads.delete` call hits the same gateway handler (nginx
// rewrites /api/langgraph/threads/* to /api/threads/*) and removes the
// thread_meta row, so this second delete's ownership guard 404s. Treat it as
// success to keep the delete idempotent.
if (!response.ok && response.status !== 404) {
const error = await response
.json()
.catch(() => ({ detail: "Failed to delete local thread data." }));
throw new Error(error.detail ?? "Failed to delete local thread data.");
}
}
async function deleteThreadEverywhere(
apiClient: ThreadDeleteClient,
threadId: string,
) {
try {
await apiClient.threads.delete(threadId);
} catch (error) {
// A previous attempt may have deleted the remote thread before local
// cleanup failed. Only 404 is success here; authorization failures are not.
if (getHttpStatus(error) !== 404) throw error;
}
await deleteLocalThreadData(threadId);
}
export async function findSidecarThreadIdsForParent(
apiClient: ThreadSidecarSearchClient,
parentThreadId: string,
) {
const threadIds: string[] = [];
const limit = 100;
let offset = 0;
while (true) {
const response = await apiClient.threads.search({
metadata: {
[SIDECAR_METADATA_KEY]: true,
parent_thread_id: parentThreadId,
},
limit,
offset,
sortBy: "updated_at",
sortOrder: "desc",
select: ["thread_id", "metadata"],
});
for (const thread of response) {
if (
isSidecarThread(thread) &&
thread.metadata?.parent_thread_id === parentThreadId
) {
threadIds.push(thread.thread_id);
}
}
if (response.length < limit) {
break;
}
offset += response.length;
}
return threadIds;
}
async function deleteSidecarThreadsForParent(
apiClient: ThreadDeleteClient,
parentThreadId: string,
) {
let sidecarThreadIds: string[];
try {
sidecarThreadIds = await findSidecarThreadIdsForParent(
apiClient,
parentThreadId,
);
} catch (err) {
console.warn(
`Failed to look up sidecar threads for parent ${parentThreadId}; skipping cascade cleanup. Orphaned sidecar threads may remain.`,
err,
);
return [];
}
const results = await Promise.allSettled(
sidecarThreadIds.map((threadId) =>
deleteThreadEverywhere(apiClient, threadId),
),
);
const failedDeletions = results
.map((result, index) =>
result.status === "rejected"
? { threadId: sidecarThreadIds[index], reason: result.reason }
: null,
)
.filter((entry): entry is { threadId: string; reason: unknown } =>
Boolean(entry),
);
if (failedDeletions.length > 0) {
console.warn(
`Failed to delete ${failedDeletions.length} sidecar thread(s) for parent ${parentThreadId}; orphaned sidecar threads may remain.`,
failedDeletions,
);
}
return sidecarThreadIds.filter((_, index) => {
return results[index]?.status === "fulfilled";
});
}
export function useDeleteThread() {
const queryClient = useQueryClient();
const apiClient = getAPIClient() as ThreadDeleteClient;
return useMutation({
mutationFn: async ({
threadId,
onDeleted,
}: {
threadId: string;
onDeleted?: () => void;
}) => {
const deletedSidecarThreadIds = await deleteSidecarThreadsForParent(
apiClient,
threadId,
);
await deleteThreadEverywhere(apiClient, threadId);
onDeleted?.();
return deletedSidecarThreadIds;
},
onSuccess(deletedSidecarThreadIds, { threadId }) {
const deletedThreadIds = new Set([threadId, ...deletedSidecarThreadIds]);
queryClient.setQueriesData(
{
queryKey: ["threads", "search"],
exact: false,
},
(oldData: Array<AgentThread> | undefined) => {
if (oldData == null) {
return oldData;
}
return oldData.filter((t) => !deletedThreadIds.has(t.thread_id));
},
);
queryClient.setQueriesData(
{
queryKey: INFINITE_THREADS_QUERY_KEY_PREFIX,
exact: false,
},
(oldData: InfiniteThreadsData | undefined) =>
filterInfiniteThreadsCache(
oldData,
(t) => !deletedThreadIds.has(t.thread_id),
),
);
},
onSettled() {
void queryClient.invalidateQueries({ queryKey: ["threads", "search"] });
void queryClient.invalidateQueries({
queryKey: INFINITE_THREADS_QUERY_KEY_PREFIX,
});
// Deleting a thread changes membership of project thread lists
// ([...PROJECTS_QUERY_KEY, "threads", id, ...]).
void queryClient.invalidateQueries({
queryKey: [...PROJECTS_QUERY_KEY, "threads"],
});
},
});
}
export function useRenameThread() {
const queryClient = useQueryClient();
const apiClient = getAPIClient();
return useMutation({
mutationFn: async ({
threadId,
title,
}: {
threadId: string;
title: string;
}) => {
await apiClient.threads.updateState(threadId, {
values: { title },
});
},
async onSuccess(_, { threadId, title }) {
const filters: QueryFilters[] = Object.values(
getThreadSnapshotCacheFilters(threadId),
);
// Prevent pre-rename snapshot requests from restoring the stale title.
await Promise.all(
filters.map((filter) => queryClient.cancelQueries(filter)),
);
setThreadTitleInCaches(queryClient, threadId, title);
for (const filter of filters) {
void queryClient.invalidateQueries(filter);
}
// The project page thread list is REST-shaped, not covered by
// setThreadTitleInCaches; invalidate it so renamed titles refresh.
void queryClient.invalidateQueries({
queryKey: [...PROJECTS_QUERY_KEY, "threads"],
});
},
});
}