mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
fix(frontend): preserve message order during long runs (#4513)
* fix(frontend): preserve message order during long runs * test(frontend): fix history pagination regression mock * fix(frontend): validate thread history sequences
This commit is contained in:
parent
6456c35675
commit
2654bc60da
@ -68,7 +68,7 @@ The frontend is a stateful chat application. Users create **threads** (conversat
|
|||||||
1. Optional composer helpers such as `core/input-polish` can rewrite the local draft before submission, and `core/voice-input` can transcribe browser microphone input into that same local draft; confirmed user input then flows to thread hooks (`core/threads/hooks.ts`) → LangGraph SDK streaming
|
1. Optional composer helpers such as `core/input-polish` can rewrite the local draft before submission, and `core/voice-input` can transcribe browser microphone input into that same local draft; confirmed user input then flows to thread hooks (`core/threads/hooks.ts`) → LangGraph SDK streaming
|
||||||
2. Stream events update thread state (messages, artifacts, todos, goal)
|
2. Stream events update thread state (messages, artifacts, todos, goal)
|
||||||
File-tool artifact auto-open work must run in an effect with timer cleanup; never schedule timers while rendering streamed `write_file` or `str_replace` updates.
|
File-tool artifact auto-open work must run in an effect with timer cleanup; never schedule timers while rendering streamed `write_file` or `str_replace` updates.
|
||||||
3. `useThreadHistory` loads persisted conversation pages from `GET /api/threads/{id}/messages/page`, preserving the backend's thread-global event `seq`; rendering overlays checkpoint/live copies at their matching canonical identities (a summarized checkpoint may contain a protected early input plus a recent tail), suppresses checkpoint/transient prefixes whose canonical position is still behind an unloaded cursor page instead of collapsing that unknown gap before a recent anchor, then adds optimistic messages without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded.
|
3. `useThreadHistory` loads persisted conversation pages from `GET /api/threads/{id}/messages/page`, preserving the backend's thread-global event `seq`; rendering overlays checkpoint/live copies at their matching canonical identities (a summarized checkpoint may contain a protected early input plus a recent tail). Context-compaction rescue diffs every retained visible identity rather than slicing at the first anchor, and keeps a run-scoped ledger of committed visible messages so replacement updates and repeated rolling checkpoint windows cannot erase an already displayed step. The resolver suppresses checkpoint/transient prefixes whose canonical position is still behind an unloaded cursor page instead of collapsing that unknown gap before a recent anchor, then adds optimistic messages without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded.
|
||||||
4. Stop actions call the LangGraph SDK stream stop path; `core/threads/hooks.ts` invalidates current-thread, thread-history, token-usage, and sidebar/search caches immediately and schedules one follow-up refetch because SDK stop may finish via abort + fire-and-forget cancel before backend title finalization commits
|
4. Stop actions call the LangGraph SDK stream stop path; `core/threads/hooks.ts` invalidates current-thread, thread-history, token-usage, and sidebar/search caches immediately and schedules one follow-up refetch because SDK stop may finish via abort + fire-and-forget cancel before backend title finalization commits
|
||||||
5. TanStack Query manages server state; localStorage stores user settings
|
5. TanStack Query manages server state; localStorage stores user settings
|
||||||
6. Components subscribe to thread state and render updates
|
6. Components subscribe to thread state and render updates
|
||||||
|
|||||||
@ -161,6 +161,8 @@ export function buildThreadSubmitMessages({
|
|||||||
// Stable identity for "no optimistic messages" so the merged-messages memo
|
// Stable identity for "no optimistic messages" so the merged-messages memo
|
||||||
// below is not invalidated by a fresh empty array on every render.
|
// below is not invalidated by a fresh empty array on every render.
|
||||||
const EMPTY_MESSAGES: Message[] = [];
|
const EMPTY_MESSAGES: Message[] = [];
|
||||||
|
const EMPTY_RUN_MESSAGES: RunMessage[] = [];
|
||||||
|
const EMPTY_MESSAGE_IDENTITIES: readonly string[] = [];
|
||||||
|
|
||||||
const EMPTY_THREAD_VALUES: AgentThreadState = {
|
const EMPTY_THREAD_VALUES: AgentThreadState = {
|
||||||
title: "",
|
title: "",
|
||||||
@ -302,6 +304,56 @@ export type ThreadMessagesPageResponse = {
|
|||||||
next_before_seq: number | null;
|
next_before_seq: number | null;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
function isValidThreadMessageSeq(value: unknown): value is number {
|
||||||
|
return typeof value === "number" && Number.isSafeInteger(value) && value >= 1;
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 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 (!isValidThreadMessageSeq(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 && !isValidThreadMessageSeq(nextBeforeSeq)) ||
|
||||||
|
(!hasMore && nextBeforeSeq !== null)
|
||||||
|
) {
|
||||||
|
throw new Error(
|
||||||
|
"Thread history returned an invalid next_before_seq cursor.",
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
return value as ThreadMessagesPageResponse;
|
||||||
|
}
|
||||||
|
|
||||||
export function getThreadHistoryNextPageParam(
|
export function getThreadHistoryNextPageParam(
|
||||||
lastPage: ThreadMessagesPageResponse,
|
lastPage: ThreadMessagesPageResponse,
|
||||||
): number | undefined {
|
): number | undefined {
|
||||||
@ -348,6 +400,51 @@ export function flattenThreadHistoryPages(
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 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) => !isValidThreadMessageSeq(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 reconciled = dedupeRunMessagesByIdentity(
|
||||||
|
[...rowsBySeq.values()].sort((left, right) => left.seq - right.seq),
|
||||||
|
);
|
||||||
|
if (
|
||||||
|
reconciled.length === previousRows.length &&
|
||||||
|
reconciled.every((row, index) => row === previousRows[index])
|
||||||
|
) {
|
||||||
|
return previousRows;
|
||||||
|
}
|
||||||
|
return reconciled;
|
||||||
|
}
|
||||||
|
|
||||||
export function mergeMessages(
|
export function mergeMessages(
|
||||||
historyMessages: Message[],
|
historyMessages: Message[],
|
||||||
threadMessages: Message[],
|
threadMessages: Message[],
|
||||||
@ -481,33 +578,78 @@ export function mergeMessages(
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 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
|
* 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.
|
* therefore need a short-lived visual bridge until run-event history catches up.
|
||||||
*
|
*
|
||||||
* Summarization emits `RemoveMessage(ALL)` + a hidden summary + the retained
|
* Summarization emits `RemoveMessage(ALL)` + a hidden summary + the retained
|
||||||
* tail. Everything in the current live thread before the first retained visible
|
* tail. Everything in the current live thread that is absent from the retained
|
||||||
* message is being removed; we keep those (minus the summary control messages
|
* visible window is being removed; we keep those (minus the summary control
|
||||||
* already tracked) so the UI can still show the full conversation (#3825).
|
* 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(
|
export function computeSummarizationTransientMessages(
|
||||||
currentMessages: Message[],
|
currentMessages: Message[],
|
||||||
summarizationMessages: Message[],
|
summarizationMessages: Message[],
|
||||||
summarizedMessageIds: ReadonlySet<string>,
|
summarizedMessageIds: ReadonlySet<string>,
|
||||||
|
previouslyRenderedMessages: Message[] = EMPTY_MESSAGES,
|
||||||
): Message[] {
|
): Message[] {
|
||||||
const firstRetainedVisibleIdentity = summarizationMessages
|
const retainedVisibleIdentities = new Set(
|
||||||
.filter((message) => message.type !== "remove")
|
summarizationMessages
|
||||||
.filter((message) => !isHiddenFromUIMessage(message))
|
.filter((message) => message.type !== "remove")
|
||||||
.map(messageIdentity)
|
.filter((message) => !isHiddenFromUIMessage(message))
|
||||||
.find(isNonEmptyString);
|
.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[] = [];
|
const moved: Message[] = [];
|
||||||
for (const message of currentMessages) {
|
for (const message of captureMessages) {
|
||||||
if (
|
const identity = messageIdentity(message);
|
||||||
firstRetainedVisibleIdentity &&
|
if (identity && retainedVisibleIdentities.has(identity)) {
|
||||||
messageIdentity(message) === firstRetainedVisibleIdentity
|
continue;
|
||||||
) {
|
|
||||||
break;
|
|
||||||
}
|
}
|
||||||
if (!summarizedMessageIds.has(message.id ?? "")) {
|
if (!summarizedMessageIds.has(message.id ?? "")) {
|
||||||
moved.push(message);
|
moved.push(message);
|
||||||
@ -538,6 +680,7 @@ export function resolveTransientHistoryBridge(
|
|||||||
bridgeOrder: readonly string[] = transientMessages
|
bridgeOrder: readonly string[] = transientMessages
|
||||||
.map(messageIdentity)
|
.map(messageIdentity)
|
||||||
.filter(isNonEmptyString),
|
.filter(isNonEmptyString),
|
||||||
|
previouslyRenderedOrder: readonly string[] = EMPTY_MESSAGE_IDENTITIES,
|
||||||
): Message[] {
|
): Message[] {
|
||||||
if (transientMessages.length === 0) {
|
if (transientMessages.length === 0) {
|
||||||
return visibleHistory;
|
return visibleHistory;
|
||||||
@ -568,20 +711,53 @@ export function resolveTransientHistoryBridge(
|
|||||||
// intentionally excluded to avoid permanent duplicates.
|
// intentionally excluded to avoid permanent duplicates.
|
||||||
const beforeAnchor = new Map<string, Message[]>();
|
const beforeAnchor = new Map<string, Message[]>();
|
||||||
const emittedMissingIdentities = new Set<string>();
|
const emittedMissingIdentities = new Set<string>();
|
||||||
|
const previouslyRenderedIndex = new Map(
|
||||||
|
previouslyRenderedOrder.map((identity, index) => [identity, index]),
|
||||||
|
);
|
||||||
let pending: Message[] = [];
|
let pending: Message[] = [];
|
||||||
let lastAnchorIdentity: string | undefined;
|
let lastAnchorIdentity: string | undefined;
|
||||||
let hasCanonicalAnchor = false;
|
let hasCanonicalAnchor = false;
|
||||||
|
|
||||||
for (const identity of bridgeOrder) {
|
for (const identity of bridgeOrder) {
|
||||||
if (presentIdentities.has(identity)) {
|
if (presentIdentities.has(identity)) {
|
||||||
if (pending.length > 0 && hasCanonicalAnchor) {
|
if (pending.length > 0) {
|
||||||
beforeAnchor.set(identity, [
|
if (hasCanonicalAnchor) {
|
||||||
...(beforeAnchor.get(identity) ?? []),
|
beforeAnchor.set(identity, [
|
||||||
...pending,
|
...(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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
// The prefix before the first loaded anchor has no trustworthy position:
|
// The prefix before the first loaded anchor has no trustworthy position:
|
||||||
// cursor pages containing its intervening history may not be loaded yet.
|
// cursor pages containing its intervening history may not be loaded yet.
|
||||||
|
// The sole exception is a prefix whose exact relative position was
|
||||||
|
// already committed to the previous UI frame.
|
||||||
pending = [];
|
pending = [];
|
||||||
hasCanonicalAnchor = true;
|
hasCanonicalAnchor = true;
|
||||||
lastAnchorIdentity = identity;
|
lastAnchorIdentity = identity;
|
||||||
@ -696,6 +872,7 @@ export function resolveThreadTransientHistoryBridge(
|
|||||||
bridgeThreadId: string | null,
|
bridgeThreadId: string | null,
|
||||||
currentThreadId: string | null | undefined,
|
currentThreadId: string | null | undefined,
|
||||||
bridgeOrder?: readonly string[],
|
bridgeOrder?: readonly string[],
|
||||||
|
previouslyRenderedOrder?: readonly string[],
|
||||||
): Message[] {
|
): Message[] {
|
||||||
if (!bridgeThreadId || bridgeThreadId !== currentThreadId) {
|
if (!bridgeThreadId || bridgeThreadId !== currentThreadId) {
|
||||||
return visibleHistory;
|
return visibleHistory;
|
||||||
@ -704,6 +881,7 @@ export function resolveThreadTransientHistoryBridge(
|
|||||||
visibleHistory,
|
visibleHistory,
|
||||||
transientMessages,
|
transientMessages,
|
||||||
bridgeOrder,
|
bridgeOrder,
|
||||||
|
previouslyRenderedOrder,
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -1324,6 +1502,9 @@ export function useThreadStream({
|
|||||||
messagesRef.current,
|
messagesRef.current,
|
||||||
_messages,
|
_messages,
|
||||||
summarizedRef.current ?? new Set<string>(),
|
summarizedRef.current ?? new Set<string>(),
|
||||||
|
renderedMessageSnapshotRef.current.threadId === threadIdRef.current
|
||||||
|
? renderedMessageSnapshotRef.current.messages
|
||||||
|
: EMPTY_MESSAGES,
|
||||||
);
|
);
|
||||||
transientHistoryOrderRef.current = mergeTransientHistoryBridgeOrder(
|
transientHistoryOrderRef.current = mergeTransientHistoryBridgeOrder(
|
||||||
transientHistoryOrderRef.current,
|
transientHistoryOrderRef.current,
|
||||||
@ -1535,6 +1716,19 @@ export function useThreadStream({
|
|||||||
// anchors so an older rescue can be placed before a newest-first page.
|
// anchors so an older rescue can be placed before a newest-first page.
|
||||||
const transientHistoryOrderRef = useRef<readonly string[]>([]);
|
const transientHistoryOrderRef = useRef<readonly string[]>([]);
|
||||||
const transientHistoryThreadIdRef = useRef<string | null>(null);
|
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);
|
const summarizedRef = useRef<Set<string>>(null);
|
||||||
// Track human message count before sending to prevent clearing optimistic
|
// Track human message count before sending to prevent clearing optimistic
|
||||||
// messages before the server's human message arrives (e.g. when AI messages
|
// messages before the server's human message arrives (e.g. when AI messages
|
||||||
@ -1554,6 +1748,11 @@ export function useThreadStream({
|
|||||||
transientHistoryBridgeRef.current = [];
|
transientHistoryBridgeRef.current = [];
|
||||||
transientHistoryOrderRef.current = [];
|
transientHistoryOrderRef.current = [];
|
||||||
transientHistoryThreadIdRef.current = null;
|
transientHistoryThreadIdRef.current = null;
|
||||||
|
renderedMessageSnapshotRef.current = {
|
||||||
|
threadId: null,
|
||||||
|
messages: EMPTY_MESSAGES,
|
||||||
|
order: EMPTY_MESSAGE_IDENTITIES,
|
||||||
|
};
|
||||||
summarizedRef.current = new Set<string>();
|
summarizedRef.current = new Set<string>();
|
||||||
pendingUsageBaselineMessageIdsRef.current = new Set();
|
pendingUsageBaselineMessageIdsRef.current = new Set();
|
||||||
pendingPreparedReplayRef.current = null;
|
pendingPreparedReplayRef.current = null;
|
||||||
@ -2069,6 +2268,10 @@ export function useThreadStream({
|
|||||||
persistedMessages,
|
persistedMessages,
|
||||||
)
|
)
|
||||||
: transientHistoryOrderRef.current;
|
: 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
|
// Commit the extended non-rendering order skeleton after React commits this
|
||||||
// render. The local value above keeps this render correctly anchored without
|
// render. The local value above keeps this render correctly anchored without
|
||||||
@ -2096,6 +2299,7 @@ export function useThreadStream({
|
|||||||
transientHistoryThreadIdRef.current,
|
transientHistoryThreadIdRef.current,
|
||||||
threadId,
|
threadId,
|
||||||
transientHistoryOrder,
|
transientHistoryOrder,
|
||||||
|
previouslyRenderedOrder,
|
||||||
);
|
);
|
||||||
return mergeMessages(
|
return mergeMessages(
|
||||||
effectiveHistory,
|
effectiveHistory,
|
||||||
@ -2103,12 +2307,36 @@ export function useThreadStream({
|
|||||||
visibleOptimisticMessages,
|
visibleOptimisticMessages,
|
||||||
);
|
);
|
||||||
}, [
|
}, [
|
||||||
|
previouslyRenderedOrder,
|
||||||
renderMessages,
|
renderMessages,
|
||||||
threadId,
|
threadId,
|
||||||
transientHistoryOrder,
|
transientHistoryOrder,
|
||||||
visibleHistory,
|
visibleHistory,
|
||||||
visibleOptimisticMessages,
|
visibleOptimisticMessages,
|
||||||
]);
|
]);
|
||||||
|
useEffect(() => {
|
||||||
|
const visibleMergedMessages = mergedMessages.filter(
|
||||||
|
(message) =>
|
||||||
|
!isHiddenFromUIMessage(message) && !message.id?.startsWith("opt-"),
|
||||||
|
);
|
||||||
|
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, pendingSupersededMessageIds, thread.isLoading, threadId]);
|
||||||
const pendingUsageMessages = thread.isLoading
|
const pendingUsageMessages = thread.isLoading
|
||||||
? getMessagesAfterBaseline(
|
? getMessagesAfterBaseline(
|
||||||
persistedMessages,
|
persistedMessages,
|
||||||
@ -2179,15 +2407,47 @@ export function useThreadHistory(
|
|||||||
),
|
),
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
return (await response.json()) as ThreadMessagesPageResponse;
|
return parseThreadMessagesPageResponse(await response.json());
|
||||||
},
|
},
|
||||||
getNextPageParam: getThreadHistoryNextPageParam,
|
getNextPageParam: getThreadHistoryNextPageParam,
|
||||||
});
|
});
|
||||||
|
|
||||||
const messageRows = useMemo(
|
const currentMessageRows = useMemo(
|
||||||
() => flattenThreadHistoryPages(historyQuery.data?.pages ?? []),
|
() => flattenThreadHistoryPages(historyQuery.data?.pages ?? []),
|
||||||
[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(() => {
|
const messages = useMemo(() => {
|
||||||
return buildVisibleHistoryMessages(
|
return buildVisibleHistoryMessages(
|
||||||
|
|||||||
@ -53,7 +53,7 @@ export interface AgentThread extends Thread<AgentThreadState> {
|
|||||||
|
|
||||||
export interface RunMessage {
|
export interface RunMessage {
|
||||||
run_id: string;
|
run_id: string;
|
||||||
seq?: number;
|
seq: number;
|
||||||
content: Message;
|
content: Message;
|
||||||
metadata: {
|
metadata: {
|
||||||
caller: string;
|
caller: string;
|
||||||
|
|||||||
@ -94,6 +94,162 @@ test.describe("Thread history", () => {
|
|||||||
).toBeVisible({ timeout: 15_000 });
|
).toBeVisible({ timeout: 15_000 });
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("keeps rendered messages ordered when the latest history page advances", async ({
|
||||||
|
page,
|
||||||
|
}) => {
|
||||||
|
const originalPrompt = "/ppt-master Build the quarterly presentation";
|
||||||
|
const followUpPrompt = "Continue with the approved default layout";
|
||||||
|
const olderRows = Array.from({ length: 50 }, (_, index) => {
|
||||||
|
const seq = index + 1;
|
||||||
|
if (index === 0) {
|
||||||
|
return {
|
||||||
|
run_id: "run-initial",
|
||||||
|
seq,
|
||||||
|
content: {
|
||||||
|
type: "human",
|
||||||
|
id: "history-prompt",
|
||||||
|
content: [{ type: "text", text: originalPrompt }],
|
||||||
|
},
|
||||||
|
metadata: { caller: "lead_agent" },
|
||||||
|
created_at: "2025-06-03T12:00:00Z",
|
||||||
|
};
|
||||||
|
}
|
||||||
|
if (index === 1) {
|
||||||
|
return {
|
||||||
|
run_id: "run-initial",
|
||||||
|
seq,
|
||||||
|
content: {
|
||||||
|
type: "ai",
|
||||||
|
id: "history-answer",
|
||||||
|
content: "Initial design is ready",
|
||||||
|
},
|
||||||
|
metadata: { caller: "lead_agent" },
|
||||||
|
created_at: "2025-06-03T12:00:01Z",
|
||||||
|
};
|
||||||
|
}
|
||||||
|
return {
|
||||||
|
run_id: "run-initial",
|
||||||
|
seq,
|
||||||
|
content: {
|
||||||
|
type: "ai",
|
||||||
|
id: `history-step-${seq}`,
|
||||||
|
content: `Historical presentation step ${seq}`,
|
||||||
|
...(index === 49
|
||||||
|
? { additional_kwargs: { turn_duration: 704 } }
|
||||||
|
: {}),
|
||||||
|
},
|
||||||
|
metadata: { caller: "lead_agent" },
|
||||||
|
created_at: "2025-06-03T12:00:02Z",
|
||||||
|
};
|
||||||
|
});
|
||||||
|
const initialRows = Array.from({ length: 50 }, (_, index) => {
|
||||||
|
const seq = index + 51;
|
||||||
|
return {
|
||||||
|
run_id: "run-initial",
|
||||||
|
seq,
|
||||||
|
content: {
|
||||||
|
type: "ai",
|
||||||
|
id: `history-step-${seq}`,
|
||||||
|
content: `Historical presentation step ${seq}`,
|
||||||
|
},
|
||||||
|
metadata: { caller: "lead_agent" },
|
||||||
|
created_at: "2025-06-03T12:00:03Z",
|
||||||
|
};
|
||||||
|
});
|
||||||
|
const shiftedRows = Array.from({ length: 50 }, (_, index) => {
|
||||||
|
const seq = index + 101;
|
||||||
|
return {
|
||||||
|
run_id: "run-shifted",
|
||||||
|
seq,
|
||||||
|
content: {
|
||||||
|
type: "ai",
|
||||||
|
id: `shifted-step-${seq}`,
|
||||||
|
content: `New presentation step ${seq}`,
|
||||||
|
},
|
||||||
|
metadata: { caller: "lead_agent" },
|
||||||
|
created_at: "2025-06-03T12:01:00Z",
|
||||||
|
};
|
||||||
|
});
|
||||||
|
let latestPageRequestCount = 0;
|
||||||
|
let cursorPageRequestCount = 0;
|
||||||
|
|
||||||
|
mockLangGraphAPI(page, {
|
||||||
|
threads: [
|
||||||
|
{
|
||||||
|
thread_id: MOCK_THREAD_ID,
|
||||||
|
title: "Long presentation task",
|
||||||
|
updated_at: "2025-06-03T12:00:00Z",
|
||||||
|
// This scenario exercises persisted run-event pagination. Keep the
|
||||||
|
// checkpoint empty so its generic mock messages do not interfere
|
||||||
|
// with optimistic -> server reconciliation after the follow-up.
|
||||||
|
messages: [],
|
||||||
|
},
|
||||||
|
],
|
||||||
|
});
|
||||||
|
await page.route(
|
||||||
|
new RegExp(`/api/threads/${MOCK_THREAD_ID}/messages/page(?:\\?.*)?$`),
|
||||||
|
async (route) => {
|
||||||
|
if (route.request().method() !== "GET") {
|
||||||
|
return route.fallback();
|
||||||
|
}
|
||||||
|
|
||||||
|
const beforeSeq = new URL(route.request().url()).searchParams.get(
|
||||||
|
"before_seq",
|
||||||
|
);
|
||||||
|
const isLatestPage = beforeSeq === null;
|
||||||
|
const rows = isLatestPage
|
||||||
|
? latestPageRequestCount === 0
|
||||||
|
? initialRows
|
||||||
|
: shiftedRows
|
||||||
|
: beforeSeq === "101"
|
||||||
|
? initialRows
|
||||||
|
: olderRows;
|
||||||
|
const hasMore = isLatestPage || beforeSeq === "101";
|
||||||
|
await route.fulfill({
|
||||||
|
status: 200,
|
||||||
|
contentType: "application/json",
|
||||||
|
body: JSON.stringify({
|
||||||
|
data: rows,
|
||||||
|
has_more: hasMore,
|
||||||
|
next_before_seq: hasMore ? (rows[0]?.seq ?? null) : null,
|
||||||
|
}),
|
||||||
|
});
|
||||||
|
if (isLatestPage) {
|
||||||
|
latestPageRequestCount += 1;
|
||||||
|
} else {
|
||||||
|
cursorPageRequestCount += 1;
|
||||||
|
}
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
await page.goto(`/workspace/chats/${MOCK_THREAD_ID}`);
|
||||||
|
await expect
|
||||||
|
.poll(() => cursorPageRequestCount, { timeout: 15_000 })
|
||||||
|
.toBeGreaterThan(0);
|
||||||
|
await expect(page.getByText(originalPrompt)).toBeVisible({
|
||||||
|
timeout: 15_000,
|
||||||
|
});
|
||||||
|
await expect(page.getByText("Completed in 11m 44s")).toBeVisible();
|
||||||
|
|
||||||
|
const latestPageRequestsBeforeSubmit = latestPageRequestCount;
|
||||||
|
const textarea = page.locator("textarea[name='message']");
|
||||||
|
await textarea.fill(followUpPrompt);
|
||||||
|
await textarea.press("Enter");
|
||||||
|
|
||||||
|
await expect
|
||||||
|
.poll(() => latestPageRequestCount, { timeout: 15_000 })
|
||||||
|
.toBeGreaterThan(latestPageRequestsBeforeSubmit);
|
||||||
|
await expect(page.getByText(originalPrompt)).toBeVisible();
|
||||||
|
await expect(page.getByText(followUpPrompt)).toBeVisible();
|
||||||
|
await expect(page.getByText("Completed in 11m 44s")).toBeVisible();
|
||||||
|
|
||||||
|
const originalBox = await page.getByText(originalPrompt).boundingBox();
|
||||||
|
const followUpBox = await page.getByText(followUpPrompt).boundingBox();
|
||||||
|
expect(originalBox).not.toBeNull();
|
||||||
|
expect(followUpBox).not.toBeNull();
|
||||||
|
expect(originalBox!.y).toBeLessThan(followUpBox!.y);
|
||||||
|
});
|
||||||
|
|
||||||
test("shows a completed run duration once after multi-step history", async ({
|
test("shows a completed run duration once after multi-step history", async ({
|
||||||
page,
|
page,
|
||||||
}) => {
|
}) => {
|
||||||
|
|||||||
@ -11,10 +11,13 @@ import {
|
|||||||
getSummarizationMiddlewareMessages,
|
getSummarizationMiddlewareMessages,
|
||||||
getThreadHistoryNextPageParam,
|
getThreadHistoryNextPageParam,
|
||||||
getVisibleOptimisticMessages,
|
getVisibleOptimisticMessages,
|
||||||
|
mergeRenderedMessageLedger,
|
||||||
mergeTransientHistoryBridge,
|
mergeTransientHistoryBridge,
|
||||||
mergeTransientHistoryBridgeOrder,
|
mergeTransientHistoryBridgeOrder,
|
||||||
mergeMessages,
|
mergeMessages,
|
||||||
|
parseThreadMessagesPageResponse,
|
||||||
pruneConfirmedTransientMessages,
|
pruneConfirmedTransientMessages,
|
||||||
|
reconcileThreadHistoryRows,
|
||||||
removeSetItems,
|
removeSetItems,
|
||||||
resolveThreadTransientHistoryBridge,
|
resolveThreadTransientHistoryBridge,
|
||||||
resolveTransientHistoryBridge,
|
resolveTransientHistoryBridge,
|
||||||
@ -22,10 +25,10 @@ import {
|
|||||||
} from "@/core/threads/hooks";
|
} from "@/core/threads/hooks";
|
||||||
import type { RunMessage } from "@/core/threads/types";
|
import type { RunMessage } from "@/core/threads/types";
|
||||||
|
|
||||||
function runMessage(seq?: number): RunMessage {
|
function runMessage(seq: number): RunMessage {
|
||||||
return {
|
return {
|
||||||
run_id: "run-1",
|
run_id: "run-1",
|
||||||
...(seq === undefined ? {} : { seq }),
|
seq,
|
||||||
content: {} as Message,
|
content: {} as Message,
|
||||||
metadata: { caller: "" },
|
metadata: { caller: "" },
|
||||||
created_at: "2026-05-22T00:00:00Z",
|
created_at: "2026-05-22T00:00:00Z",
|
||||||
@ -108,6 +111,7 @@ test("mergeMessages preserves historical run metadata on a live checkpoint repla
|
|||||||
[
|
[
|
||||||
{
|
{
|
||||||
run_id: "run-1",
|
run_id: "run-1",
|
||||||
|
seq: 1,
|
||||||
content: persistedAi,
|
content: persistedAi,
|
||||||
metadata: { caller: "lead_agent" },
|
metadata: { caller: "lead_agent" },
|
||||||
created_at: "2026-07-21T00:00:00Z",
|
created_at: "2026-07-21T00:00:00Z",
|
||||||
@ -496,6 +500,54 @@ test("buildThreadMessagesPageUrl returns a relative URL behind nginx", () => {
|
|||||||
);
|
);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("parseThreadMessagesPageResponse accepts a valid history page", () => {
|
||||||
|
const response = {
|
||||||
|
data: [runMessage(1), runMessage(2)],
|
||||||
|
has_more: true,
|
||||||
|
next_before_seq: 1,
|
||||||
|
};
|
||||||
|
|
||||||
|
expect(parseThreadMessagesPageResponse(response)).toBe(response);
|
||||||
|
});
|
||||||
|
|
||||||
|
test.each([
|
||||||
|
["missing", undefined],
|
||||||
|
["non-numeric", "2"],
|
||||||
|
["fractional", 2.5],
|
||||||
|
["unsafe", Number.MAX_SAFE_INTEGER + 1],
|
||||||
|
])(
|
||||||
|
"parseThreadMessagesPageResponse rejects a %s row seq",
|
||||||
|
(_description, seq) => {
|
||||||
|
expect(() =>
|
||||||
|
parseThreadMessagesPageResponse({
|
||||||
|
data: [{ ...runMessage(1), seq }],
|
||||||
|
has_more: false,
|
||||||
|
next_before_seq: null,
|
||||||
|
}),
|
||||||
|
).toThrow("invalid seq");
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
test("parseThreadMessagesPageResponse rejects duplicate row seq values", () => {
|
||||||
|
expect(() =>
|
||||||
|
parseThreadMessagesPageResponse({
|
||||||
|
data: [runMessage(1), runMessage(1)],
|
||||||
|
has_more: false,
|
||||||
|
next_before_seq: null,
|
||||||
|
}),
|
||||||
|
).toThrow("duplicate seq");
|
||||||
|
});
|
||||||
|
|
||||||
|
test("parseThreadMessagesPageResponse rejects an invalid pagination cursor", () => {
|
||||||
|
expect(() =>
|
||||||
|
parseThreadMessagesPageResponse({
|
||||||
|
data: [runMessage(1)],
|
||||||
|
has_more: true,
|
||||||
|
next_before_seq: null,
|
||||||
|
}),
|
||||||
|
).toThrow("invalid next_before_seq");
|
||||||
|
});
|
||||||
|
|
||||||
test("flattenThreadHistoryPages prepends backward pages in global seq order", () => {
|
test("flattenThreadHistoryPages prepends backward pages in global seq order", () => {
|
||||||
expect(
|
expect(
|
||||||
flattenThreadHistoryPages([
|
flattenThreadHistoryPages([
|
||||||
@ -537,6 +589,96 @@ test("flattenThreadHistoryPages retains backward pages when the latest page refr
|
|||||||
).toEqual([1, 2, 3, 4, 5]);
|
).toEqual([1, 2, 3, 4, 5]);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("reconcileThreadHistoryRows retains rows displaced from a moving latest page", () => {
|
||||||
|
const previousRows = Array.from({ length: 50 }, (_, index) =>
|
||||||
|
runMessage(index + 1),
|
||||||
|
);
|
||||||
|
const currentRows = Array.from({ length: 50 }, (_, index) =>
|
||||||
|
runMessage(index + 51),
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(
|
||||||
|
reconcileThreadHistoryRows(previousRows, currentRows, false).map(
|
||||||
|
(message) => message.seq,
|
||||||
|
),
|
||||||
|
).toEqual(Array.from({ length: 100 }, (_, index) => index + 1));
|
||||||
|
});
|
||||||
|
|
||||||
|
test("reconcileThreadHistoryRows refreshes overlapping rows without moving them", () => {
|
||||||
|
const stale = runMessage(50);
|
||||||
|
const refreshed = {
|
||||||
|
...runMessage(50),
|
||||||
|
content: {
|
||||||
|
id: "message-50",
|
||||||
|
type: "ai",
|
||||||
|
content: "final response",
|
||||||
|
additional_kwargs: { turn_duration: 704 },
|
||||||
|
} as Message,
|
||||||
|
};
|
||||||
|
|
||||||
|
const reconciled = reconcileThreadHistoryRows(
|
||||||
|
[runMessage(49), stale],
|
||||||
|
[refreshed, runMessage(51)],
|
||||||
|
false,
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(reconciled.map((message) => message.seq)).toEqual([49, 50, 51]);
|
||||||
|
expect(reconciled[1]).toBe(refreshed);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("reconcileThreadHistoryRows trusts a complete snapshot and prunes missing rows", () => {
|
||||||
|
const currentRows = [runMessage(2), runMessage(3)];
|
||||||
|
|
||||||
|
expect(
|
||||||
|
reconcileThreadHistoryRows(
|
||||||
|
[runMessage(1), ...currentRows],
|
||||||
|
currentRows,
|
||||||
|
true,
|
||||||
|
),
|
||||||
|
).toEqual(currentRows);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("reconcileThreadHistoryRows preserves existing history when a row seq is invalid", () => {
|
||||||
|
const previousRows = [runMessage(1), runMessage(2)];
|
||||||
|
const invalidRow = {
|
||||||
|
...runMessage(3),
|
||||||
|
seq: undefined,
|
||||||
|
} as unknown as RunMessage;
|
||||||
|
const consoleError = rs
|
||||||
|
.spyOn(console, "error")
|
||||||
|
.mockImplementation(() => undefined);
|
||||||
|
|
||||||
|
expect(reconcileThreadHistoryRows(previousRows, [invalidRow], false)).toBe(
|
||||||
|
previousRows,
|
||||||
|
);
|
||||||
|
expect(consoleError).toHaveBeenCalledOnce();
|
||||||
|
|
||||||
|
consoleError.mockRestore();
|
||||||
|
});
|
||||||
|
|
||||||
|
test("reconcileThreadHistoryRows keeps insertion order on an invalid initial snapshot", () => {
|
||||||
|
const first = {
|
||||||
|
...runMessage(1),
|
||||||
|
seq: undefined,
|
||||||
|
content: { id: "first", type: "human", content: "first" } as Message,
|
||||||
|
} as unknown as RunMessage;
|
||||||
|
const second = {
|
||||||
|
...runMessage(2),
|
||||||
|
seq: undefined,
|
||||||
|
content: { id: "second", type: "ai", content: "second" } as Message,
|
||||||
|
} as unknown as RunMessage;
|
||||||
|
const consoleError = rs
|
||||||
|
.spyOn(console, "error")
|
||||||
|
.mockImplementation(() => undefined);
|
||||||
|
|
||||||
|
const currentRows = [first, second];
|
||||||
|
|
||||||
|
expect(reconcileThreadHistoryRows([], currentRows, false)).toBe(currentRows);
|
||||||
|
expect(consoleError).toHaveBeenCalledOnce();
|
||||||
|
|
||||||
|
consoleError.mockRestore();
|
||||||
|
});
|
||||||
|
|
||||||
test("infinite history refetch recalculates older-page cursors from the refreshed newest page", async () => {
|
test("infinite history refetch recalculates older-page cursors from the refreshed newest page", async () => {
|
||||||
const queryClient = new QueryClient({
|
const queryClient = new QueryClient({
|
||||||
defaultOptions: { queries: { retry: false } },
|
defaultOptions: { queries: { retry: false } },
|
||||||
@ -651,24 +793,28 @@ test("buildVisibleHistoryMessages filters superseded runs but keeps regenerated
|
|||||||
const rows: RunMessage[] = [
|
const rows: RunMessage[] = [
|
||||||
{
|
{
|
||||||
run_id: "run-old",
|
run_id: "run-old",
|
||||||
|
seq: 1,
|
||||||
content: oldHuman,
|
content: oldHuman,
|
||||||
metadata: { caller: "lead_agent" },
|
metadata: { caller: "lead_agent" },
|
||||||
created_at: "2026-06-18T00:00:00Z",
|
created_at: "2026-06-18T00:00:00Z",
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
run_id: "run-old",
|
run_id: "run-old",
|
||||||
|
seq: 2,
|
||||||
content: oldAi,
|
content: oldAi,
|
||||||
metadata: { caller: "lead_agent" },
|
metadata: { caller: "lead_agent" },
|
||||||
created_at: "2026-06-18T00:00:01Z",
|
created_at: "2026-06-18T00:00:01Z",
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
run_id: "run-new",
|
run_id: "run-new",
|
||||||
|
seq: 3,
|
||||||
content: newHuman,
|
content: newHuman,
|
||||||
metadata: { caller: "lead_agent" },
|
metadata: { caller: "lead_agent" },
|
||||||
created_at: "2026-06-18T00:00:02Z",
|
created_at: "2026-06-18T00:00:02Z",
|
||||||
},
|
},
|
||||||
{
|
{
|
||||||
run_id: "run-new",
|
run_id: "run-new",
|
||||||
|
seq: 4,
|
||||||
content: newAi,
|
content: newAi,
|
||||||
metadata: { caller: "lead_agent" },
|
metadata: { caller: "lead_agent" },
|
||||||
created_at: "2026-06-18T00:00:03Z",
|
created_at: "2026-06-18T00:00:03Z",
|
||||||
@ -687,6 +833,7 @@ test("buildVisibleHistoryMessages attaches run_id to each content message (#3779
|
|||||||
const rows: RunMessage[] = [
|
const rows: RunMessage[] = [
|
||||||
{
|
{
|
||||||
run_id: "run-1",
|
run_id: "run-1",
|
||||||
|
seq: 1,
|
||||||
content: { id: "ai-1", type: "ai", content: "answer" } as Message,
|
content: { id: "ai-1", type: "ai", content: "answer" } as Message,
|
||||||
metadata: { caller: "lead_agent" },
|
metadata: { caller: "lead_agent" },
|
||||||
created_at: "2026-06-26T00:00:00Z",
|
created_at: "2026-06-26T00:00:00Z",
|
||||||
@ -821,6 +968,53 @@ test("resolveTransientHistoryBridge does not collapse an unloaded gap before its
|
|||||||
).toEqual(["event-seq-35", "event-seq-88"]);
|
).toEqual(["event-seq-35", "event-seq-88"]);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("resolveTransientHistoryBridge restores a prefix that was already rendered", () => {
|
||||||
|
const protectedPrompt = {
|
||||||
|
id: "protected-prompt",
|
||||||
|
type: "human",
|
||||||
|
content: "/ppt-master 帮我做个ppt",
|
||||||
|
} as Message;
|
||||||
|
const intermediateReply = {
|
||||||
|
id: "intermediate-reply",
|
||||||
|
type: "ai",
|
||||||
|
content: "正在生成页面",
|
||||||
|
} as Message;
|
||||||
|
const pageAnchor = {
|
||||||
|
id: "page-anchor",
|
||||||
|
type: "ai",
|
||||||
|
content: "继续导出 SVG",
|
||||||
|
} as Message;
|
||||||
|
const latestAnswer = {
|
||||||
|
id: "latest-answer",
|
||||||
|
type: "ai",
|
||||||
|
content: "任务仍在执行",
|
||||||
|
} as Message;
|
||||||
|
const captured = [protectedPrompt, intermediateReply, pageAnchor];
|
||||||
|
const bridgeOrder = mergeTransientHistoryBridgeOrder([], captured);
|
||||||
|
const previouslyRenderedOrder = [
|
||||||
|
protectedPrompt,
|
||||||
|
intermediateReply,
|
||||||
|
pageAnchor,
|
||||||
|
latestAnswer,
|
||||||
|
]
|
||||||
|
.map((message) => `message:${message.id}`)
|
||||||
|
.filter((identity): identity is string => Boolean(identity));
|
||||||
|
|
||||||
|
expect(
|
||||||
|
resolveTransientHistoryBridge(
|
||||||
|
[pageAnchor, latestAnswer],
|
||||||
|
captured,
|
||||||
|
bridgeOrder,
|
||||||
|
previouslyRenderedOrder,
|
||||||
|
).map((message) => message.id),
|
||||||
|
).toEqual([
|
||||||
|
"protected-prompt",
|
||||||
|
"intermediate-reply",
|
||||||
|
"page-anchor",
|
||||||
|
"latest-answer",
|
||||||
|
]);
|
||||||
|
});
|
||||||
|
|
||||||
test("resolveTransientHistoryBridge does not duplicate once canonical history catches up", () => {
|
test("resolveTransientHistoryBridge does not duplicate once canonical history catches up", () => {
|
||||||
expect(
|
expect(
|
||||||
resolveTransientHistoryBridge(
|
resolveTransientHistoryBridge(
|
||||||
@ -1093,6 +1287,189 @@ test("computeSummarizationTransientMessages captures live turns dropped before t
|
|||||||
).toEqual([summarizationHuman1, summarizationAi1, summarizationHuman2]);
|
).toEqual([summarizationHuman1, summarizationAi1, summarizationHuman2]);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
test("computeSummarizationTransientMessages rescues rendered processing steps missing from a stale live snapshot", () => {
|
||||||
|
const processingMessages = Array.from({ length: 10 }, (_, index) => {
|
||||||
|
const step = index + 1;
|
||||||
|
return {
|
||||||
|
id: `processing-${step}`,
|
||||||
|
type: step % 2 === 0 ? "tool" : "ai",
|
||||||
|
...(step % 2 === 0 ? { tool_call_id: `call-${step - 1}` } : {}),
|
||||||
|
content: `step ${step}`,
|
||||||
|
} as Message;
|
||||||
|
});
|
||||||
|
// The SDK has already advanced to the 10-message post-compaction window,
|
||||||
|
// while the UI's previous committed frame still contains steps 1..10.
|
||||||
|
const staleLiveSnapshot = processingMessages.slice(4);
|
||||||
|
const summarizationMessages = [
|
||||||
|
{
|
||||||
|
id: "__remove_all__",
|
||||||
|
type: "remove",
|
||||||
|
content: "",
|
||||||
|
} as Message,
|
||||||
|
...staleLiveSnapshot,
|
||||||
|
];
|
||||||
|
|
||||||
|
expect(
|
||||||
|
computeSummarizationTransientMessages(
|
||||||
|
staleLiveSnapshot,
|
||||||
|
summarizationMessages,
|
||||||
|
new Set(),
|
||||||
|
processingMessages,
|
||||||
|
),
|
||||||
|
).toEqual(processingMessages.slice(0, 4));
|
||||||
|
});
|
||||||
|
|
||||||
|
test("computeSummarizationTransientMessages rescues steps between a protected input and retained tail", () => {
|
||||||
|
const protectedInput = {
|
||||||
|
id: "protected-input",
|
||||||
|
type: "human",
|
||||||
|
content: "Run a long sequential research task",
|
||||||
|
} as Message;
|
||||||
|
const removedSteps = Array.from({ length: 4 }, (_, index) => ({
|
||||||
|
id: `protected-window-step-${index + 1}`,
|
||||||
|
type: "ai",
|
||||||
|
content: `completed ${index + 1}`,
|
||||||
|
})) as Message[];
|
||||||
|
const retainedTail = {
|
||||||
|
id: "retained-tail",
|
||||||
|
type: "tool",
|
||||||
|
tool_call_id: "retained-call",
|
||||||
|
content: "latest search result",
|
||||||
|
} as Message;
|
||||||
|
const renderedMessages = [protectedInput, ...removedSteps, retainedTail];
|
||||||
|
const retainedWindow = [protectedInput, retainedTail];
|
||||||
|
|
||||||
|
expect(
|
||||||
|
computeSummarizationTransientMessages(
|
||||||
|
retainedWindow,
|
||||||
|
[
|
||||||
|
{
|
||||||
|
id: "__remove_all__",
|
||||||
|
type: "remove",
|
||||||
|
content: "",
|
||||||
|
} as Message,
|
||||||
|
...retainedWindow,
|
||||||
|
],
|
||||||
|
new Set(),
|
||||||
|
renderedMessages,
|
||||||
|
),
|
||||||
|
).toEqual(removedSteps);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("repeated compaction keeps every previously rendered processing step in order", () => {
|
||||||
|
const processingMessages = Array.from({ length: 12 }, (_, index) => {
|
||||||
|
const step = index + 1;
|
||||||
|
return {
|
||||||
|
id: `repeat-processing-${step}`,
|
||||||
|
type: step % 2 === 0 ? "tool" : "ai",
|
||||||
|
...(step % 2 === 0 ? { tool_call_id: `repeat-call-${step - 1}` } : {}),
|
||||||
|
content: `step ${step}`,
|
||||||
|
} as Message;
|
||||||
|
});
|
||||||
|
const removeAll = {
|
||||||
|
id: "__remove_all__",
|
||||||
|
type: "remove",
|
||||||
|
content: "",
|
||||||
|
} as Message;
|
||||||
|
|
||||||
|
const firstTail = processingMessages.slice(4, 10);
|
||||||
|
const firstMoved = computeSummarizationTransientMessages(
|
||||||
|
firstTail,
|
||||||
|
[removeAll, ...firstTail],
|
||||||
|
new Set(),
|
||||||
|
processingMessages.slice(0, 10),
|
||||||
|
);
|
||||||
|
const firstBridge = mergeTransientHistoryBridge([], firstMoved);
|
||||||
|
const firstMerged = mergeMessages(firstBridge, firstTail, []);
|
||||||
|
|
||||||
|
const secondTail = processingMessages.slice(6);
|
||||||
|
const secondMoved = computeSummarizationTransientMessages(
|
||||||
|
secondTail,
|
||||||
|
[removeAll, ...secondTail],
|
||||||
|
new Set(),
|
||||||
|
processingMessages,
|
||||||
|
);
|
||||||
|
const secondBridge = mergeTransientHistoryBridge(firstBridge, secondMoved);
|
||||||
|
const secondMerged = mergeMessages(secondBridge, secondTail, []);
|
||||||
|
|
||||||
|
expect(firstMerged.map((message) => message.id)).toEqual(
|
||||||
|
processingMessages.slice(0, 10).map((message) => message.id),
|
||||||
|
);
|
||||||
|
expect(secondMerged.map((message) => message.id)).toEqual(
|
||||||
|
processingMessages.map((message) => message.id),
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("rendered message ledger survives rolling live windows before repeated compaction", () => {
|
||||||
|
const processingMessages = Array.from({ length: 23 }, (_, index) => {
|
||||||
|
const step = index + 1;
|
||||||
|
return {
|
||||||
|
id: `rolling-processing-${step}`,
|
||||||
|
type: "ai",
|
||||||
|
content: `completed ${step}/26`,
|
||||||
|
} as Message;
|
||||||
|
});
|
||||||
|
const firstVisibleWindow = processingMessages.slice(0, 13);
|
||||||
|
const secondVisibleWindow = processingMessages.slice(8, 18);
|
||||||
|
const thirdVisibleWindow = processingMessages.slice(13);
|
||||||
|
|
||||||
|
const firstLedger = mergeRenderedMessageLedger([], firstVisibleWindow);
|
||||||
|
const secondLedger = mergeRenderedMessageLedger(
|
||||||
|
firstLedger,
|
||||||
|
secondVisibleWindow,
|
||||||
|
);
|
||||||
|
const thirdLedger = mergeRenderedMessageLedger(
|
||||||
|
secondLedger,
|
||||||
|
thirdVisibleWindow,
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(thirdLedger.map((message) => message.id)).toEqual(
|
||||||
|
processingMessages.map((message) => message.id),
|
||||||
|
);
|
||||||
|
|
||||||
|
// A later compaction retains only 19..23. The accumulated display ledger
|
||||||
|
// must still supply 14..18 (and every older displayed step) to the bridge.
|
||||||
|
const retainedTail = processingMessages.slice(18);
|
||||||
|
const moved = computeSummarizationTransientMessages(
|
||||||
|
retainedTail,
|
||||||
|
[
|
||||||
|
{
|
||||||
|
id: "__remove_all__",
|
||||||
|
type: "remove",
|
||||||
|
content: "",
|
||||||
|
} as Message,
|
||||||
|
...retainedTail,
|
||||||
|
],
|
||||||
|
new Set(),
|
||||||
|
thirdLedger,
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(moved.map((message) => message.id)).toEqual(
|
||||||
|
processingMessages.slice(0, 18).map((message) => message.id),
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
test("rendered message ledger does not retain explicitly superseded messages", () => {
|
||||||
|
const retained = {
|
||||||
|
id: "retained-answer",
|
||||||
|
type: "ai",
|
||||||
|
content: "keep me",
|
||||||
|
} as Message;
|
||||||
|
const superseded = {
|
||||||
|
id: "superseded-answer",
|
||||||
|
type: "ai",
|
||||||
|
content: "replace me",
|
||||||
|
} as Message;
|
||||||
|
|
||||||
|
expect(
|
||||||
|
mergeRenderedMessageLedger(
|
||||||
|
[retained, superseded],
|
||||||
|
[retained],
|
||||||
|
new Set([superseded.id!]),
|
||||||
|
),
|
||||||
|
).toEqual([retained]);
|
||||||
|
});
|
||||||
|
|
||||||
test("computeSummarizationTransientMessages excludes already-summarized control messages", () => {
|
test("computeSummarizationTransientMessages excludes already-summarized control messages", () => {
|
||||||
const priorSummary = {
|
const priorSummary = {
|
||||||
id: "summary-0",
|
id: "summary-0",
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user