From 0464502af1c51d7185e6f51d4de65b074a16d5aa Mon Sep 17 00:00:00 2001 From: Beverly621 <205182701+Beverly621@users.noreply.github.com> Date: Sat, 12 Sep 2026 05:27:04 -0400 Subject: [PATCH] fix(frontend): keep human input cards with their turn (#4892) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(frontend): keep human input cards with their turn * fix(frontend): keep human input cards with the correct turn Multi-turn ordering in restoreLocalTurnMessageOrder could place an `ask_clarification` (needYourHelp) card on the wrong side of a newly submitted human message, and after an interrupt/stop it could move the current run's own already-executed steps above the human that started them. - Restore established messages that a live checkpoint tail wove after the new human (displacedBaselineMessages). - Treat cards/messages confirmed only by the REST history page as established-past-turn too (confirmedHistoryIdentities), not as in-flight pending steps (displacedHistoryMessages). - Never displace the CURRENT run's own steps after an interrupt/stop; they belong after the human even once canonical history confirms them (currentTurnRunIds, anchored by the pending human's run_id). Fixes #4889 * fix(frontend): preserve ordering across displaced messages * fix(frontend): close canonical history ordering gaps * fix(frontend): preserve current turn anchor after compaction * fix(frontend): anchor the local turn on the submitted human identity Follow-up to #4892. R2 is reachable through the full hook chain: when the checkpoint baseline covers only the latest turn, the server echo of the submitted human confirms the optimistic copy against the unthrottled SDK state while the ~80ms render snapshot cannot show it yet; the baseline-only anchor scan then promoted an older history-only human into the current turn's anchor and moved established history behind it. - Record a LocalTurnAnchor at dispatch: one client-generated human id is shared by the optimistic display copy and the submitted message, so the server X__user echo confirms the exact identity already on screen. - restoreLocalTurnMessageOrder repairs only when that identity is present in the display; a null anchor (hidden human-input reply, regenerate replay) or a not-yet-rendered identity keeps established history untouched. - Optimistic confirmation now observes the same coalesced render snapshot (identity match first, rendered human-count growth as fallback for runtime-re-keyed first turns) instead of the per-chunk array. - Edit replays adopt the prepare response's replacement identity; the render ledger excludes unconfirmed optimistic copies by identity now that the local input no longer uses an opt- prefix, so a failed send cannot pin a message the server never saw. - Anchor lifecycle matches the previous baseline: kept across finish/stop/error until canonical data takes over, replaced by the next local submit, cleared on send failure, thread switch, and replay gaps. * test(threads): type submit mock calls in local-turn-order dom tests * fix(frontend): bound local turn repair to pre-submit history * fix(frontend): preserve pre-submit bridge ordering --------- Co-authored-by: 肘子香香 Co-authored-by: 霍英豪 Co-authored-by: wangzeren <1004695029@qq.com> --- frontend/src/AGENTS.md | 2 +- frontend/src/core/threads/hooks.ts | 447 ++++++- .../threads/local-turn-order.dom.test.tsx | 1071 ++++++++++++++++- .../unit/core/threads/message-merge.test.ts | 625 ++++++++++ .../unit/core/threads/send-message.test.ts | 16 + 5 files changed, 2094 insertions(+), 67 deletions(-) diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index 148dcef96..dbdcf3247 100644 --- a/frontend/src/AGENTS.md +++ b/frontend/src/AGENTS.md @@ -11,7 +11,7 @@ full-file action; do not mount CodeMirror for that artifact until the user requests and receives the complete content. The Gateway retains range ownership and returns 206/416 through `FileResponse`. -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. A checkpoint/transient prefix whose canonical position is still behind an unloaded cursor page is woven in before the first shared anchor, not discarded: both the checkpoint and seq-sorted history place it earlier, so that position is known even when the pages between are not. Position authority is seq-first and lives in `core/threads/message-order.ts` (re-exported through `hooks.ts`): every normalized identity tracks latest visible content and trusted position separately; a valid `deerflow_seq` (positive safe integer, earliest value wins per identity, hidden control copies contribute a position only when no visible copy carries one) joins an ascending skeleton that outranks identity-anchor weaving, which remains the fallback for no-seq segments whose internal order is preserved. Live-only and rescued messages with trusted seq also serve as anchors for adjacent no-seq segments. A trailing segment follows its last live-only positioned anchor within the loaded window; after a shared anchor or a rescued prefix before the window, it stays at the tail. Optimistic messages remain last; the transient bridge inserts positioned rows before weaving so the insertion cannot reverse previously displayed steps. Rescued sequence anchors preserve preceding captured steps even before React has rendered them; only a leading prefix anchored to loaded history requires previously rendered ordering to cross an unloaded cursor gap. Content replacement never drops the known seq, `run_id`, or `turn_duration`, and the compaction transient bridge plus rendered ledger share the same position priority. `deerflow_seq` is server-owned display metadata and is never written back into a checkpoint. It must never be appended to the tail (#4065) — the tail is provably wrong — but suppressing it entirely is how a user's own question vanished from a long thread once the first 50-row history page no longer reached back to it (#4666). A collapsed unloaded gap is recoverable by paging; a dropped message is not. Weaving alone restores the message but not its exact position — after compaction the live window carries too few anchors — so both sides now carry the backend's thread-global `additional_kwargs.deerflow_seq`: `buildVisibleHistoryMessages` copies each row's `seq`, and the Gateway stamps it onto `values` frame messages it has already persisted. A live message whose seq is below the loaded window's lower bound is placed ahead of everything on screen instead of before the nearest anchor, which is what puts a compaction-rescued first user turn back at the head rather than mid-transcript. That split happens _before_ the anchor walk, not inside it: a compacted checkpoint can share no identity at all with the loaded page — it keeps only the current run's recent tail, while the page on screen was fetched turns earlier — and the anchor walk then never runs at all, which is precisely when a rescued turn most needs its seq. Doing the split inside the walk left that case appending the message after the whole window (#4666), the one arrangement #4065 proved wrong. A message without a seq (still streaming, so not in the feed yet) keeps the weaving path — the tail is already its correct position. Optimistic messages are then added without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded. Dynamic context re-keys the submitted user message from `X` to `X__user`; UI identity matching normalizes that reserved suffix only for human messages so the submitted frame and checkpoint replacement remain one visible turn. A locally submitted turn also records its pre-submit identity baseline: if `messages-tuple` publishes new AI/tool steps before canonical history (or replay-gap recovery) publishes that turn's human message, render ordering moves only those non-baseline visible steps behind the new human while leaving history, hidden controls, and reconnected runs untouched. Keep that local order anchor through finish, stop, and stream error because the SDK's settled frame can retain transient event order; replace it on the next local submit and clear it on thread switch or replay-gap recovery. +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. A checkpoint/transient prefix whose canonical position is still behind an unloaded cursor page is woven in before the first shared anchor, not discarded: both the checkpoint and seq-sorted history place it earlier, so that position is known even when the pages between are not. Position authority is seq-first and lives in `core/threads/message-order.ts` (re-exported through `hooks.ts`): every normalized identity tracks latest visible content and trusted position separately; a valid `deerflow_seq` (positive safe integer, earliest value wins per identity, hidden control copies contribute a position only when no visible copy carries one) joins an ascending skeleton that outranks identity-anchor weaving, which remains the fallback for no-seq segments whose internal order is preserved. Live-only and rescued messages with trusted seq also serve as anchors for adjacent no-seq segments. A trailing segment follows its last live-only positioned anchor within the loaded window; after a shared anchor or a rescued prefix before the window, it stays at the tail. Optimistic messages remain last; the transient bridge inserts positioned rows before weaving so the insertion cannot reverse previously displayed steps. Rescued sequence anchors preserve preceding captured steps even before React has rendered them; only a leading prefix anchored to loaded history requires previously rendered ordering to cross an unloaded cursor gap. Content replacement never drops the known seq, `run_id`, or `turn_duration`, and the compaction transient bridge plus rendered ledger share the same position priority. `deerflow_seq` is server-owned display metadata and is never written back into a checkpoint. It must never be appended to the tail (#4065) — the tail is provably wrong — but suppressing it entirely is how a user's own question vanished from a long thread once the first 50-row history page no longer reached back to it (#4666). A collapsed unloaded gap is recoverable by paging; a dropped message is not. Weaving alone restores the message but not its exact position — after compaction the live window carries too few anchors — so both sides now carry the backend's thread-global `additional_kwargs.deerflow_seq`: `buildVisibleHistoryMessages` copies each row's `seq`, and the Gateway stamps it onto `values` frame messages it has already persisted. A live message whose seq is below the loaded window's lower bound is placed ahead of everything on screen instead of before the nearest anchor, which is what puts a compaction-rescued first user turn back at the head rather than mid-transcript. That split happens _before_ the anchor walk, not inside it: a compacted checkpoint can share no identity at all with the loaded page — it keeps only the current run's recent tail, while the page on screen was fetched turns earlier — and the anchor walk then never runs at all, which is precisely when a rescued turn most needs its seq. Doing the split inside the walk left that case appending the message after the whole window (#4666), the one arrangement #4065 proved wrong. A message without a seq (still streaming, so not in the feed yet) keeps the weaving path — the tail is already its correct position. Optimistic messages are then added without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded. Dynamic context re-keys each submitted user message from the client-generated `local-human-*` identity `X` to the visible server echo `X__user`; UI identity matching normalizes that reserved suffix only for human messages so the optimistic input and checkpoint replacement remain one visible turn. At dispatch, a local-turn anchor snapshots the checkpoint identity baseline, canonical history identities and maximum trusted seq, and any pre-existing transient-bridge identities. Render repair uses `confirmedHistoryIdentities` plus `preSubmitMaxSeq` to restore baseline or history-confirmed messages above that exact human anchor, while moving only speculative non-baseline AI/tool steps behind it; `currentTurnRunIds` keeps already-persisted steps of the active turn below their human. Keep the anchor scoped to its originating thread through finish, stop, and stream error because the SDK's settled frame can retain transient event order; replace it on the next local submit and clear it on thread switch or replay-gap recovery. 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. The Settings > Tools MCP switch calls the targeted `PATCH /api/mcp/config` diff --git a/frontend/src/core/threads/hooks.ts b/frontend/src/core/threads/hooks.ts index fb4c01af5..db2e25f08 100644 --- a/frontend/src/core/threads/hooks.ts +++ b/frontend/src/core/threads/hooks.ts @@ -34,6 +34,7 @@ import { taskEventToSubtaskUpdate } from "../tasks/lifecycle"; import { messageToStep } from "../tasks/steps"; import type { UploadedFileInfo } from "../uploads"; import { promptInputFilePartToFile, uploadFiles } from "../uploads"; +import { uuid } from "../utils/uuid"; import { branchThreadFromTurn, @@ -161,16 +162,24 @@ export function buildThreadSubmitMessages({ additionalKwargs, additionalInputMessages = [], filesForSubmit = [], + humanMessageId, }: { text: string; additionalKwargs?: Record; 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 (`__user`) + * confirms the exact message the user already sees. + */ + humanMessageId?: string; }): Message[] { return [ ...additionalInputMessages, { type: "human", + ...(humanMessageId ? { id: humanMessageId } : {}), content: [ { type: "text", @@ -190,6 +199,31 @@ export function buildThreadSubmitMessages({ const EMPTY_MESSAGES: Message[] = []; const EMPTY_RUN_MESSAGES: RunMessage[] = []; const EMPTY_MESSAGE_IDENTITIES: readonly string[] = []; +const EMPTY_MESSAGE_IDENTITIES_SET: ReadonlySet = new Set(); +/** + * 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:`) 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; + /** Canonical REST-history identities already loaded when this turn began. */ + preSubmitHistoryIdentities: ReadonlySet; + /** Transient-bridge identities already established before this turn began. */ + preSubmitBridgeIdentities: ReadonlySet; + /** + * 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; @@ -207,6 +241,58 @@ const SUMMARIZATION_MIDDLEWARE_UPDATE_KEYS = new Set([ "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 { + 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(); messages.forEach((message, index) => { @@ -456,31 +542,133 @@ export function reconcileThreadHistoryRows( // 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 | null, + confirmedHistoryIdentities: ReadonlySet = EMPTY_MESSAGE_IDENTITIES_SET, +): Set { + const runIds = new Set(); + 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 - * first new human message without disturbing established history or hidden - * checkpoint controls. The caller keeps the baseline after stream completion - * because the SDK may retain its transient event order until the next submit. + * 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, + confirmedHistoryIdentities: ReadonlySet = EMPTY_MESSAGE_IDENTITIES_SET, + currentTurnRunIds: ReadonlySet = EMPTY_MESSAGE_IDENTITIES_SET, + anchorHumanIdentity?: string | null, + canonicalHistoryIdentities: ReadonlySet = confirmedHistoryIdentities, ): Message[] { - const pendingHumanIndex = messages.findIndex((message) => { - const identity = messageIdentity(message); - return ( - message.type === "human" && - !isHiddenFromUIMessage(message) && - identity !== undefined && - !baselineMessageIdentities.has(identity) - ); - }); - if (pendingHumanIndex <= 0) { + // 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[] = []; @@ -490,22 +678,43 @@ export function restoreLocalTurnMessageOrder( (message.type === "ai" || message.type === "tool") && !isHiddenFromUIMessage(message) && identity !== undefined && - !baselineMessageIdentities.has(identity); + !baselineMessageIdentities.has(identity) && + ((!isCanonicalHistoryMessage(identity) && + !isConfirmedHistoryMessage(identity)) || + isCurrentTurnStep(message)); if (isVisiblePendingStep) { earlyPendingSteps.push(message); } else { stablePrefix.push(message); } } - if (earlyPendingSteps.length === 0) { + 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, - ...messages.slice(pendingHumanIndex + 1), + ...stableSuffix, ]; } @@ -1695,7 +1904,7 @@ export function useThreadStream({ transientHistoryThreadIdRef.current = null; summarizedRef.current = new Set(); pendingUsageBaselineMessageIdsRef.current = new Set(); - localTurnOrderBaselineIdentitiesRef.current = null; + localTurnAnchorRef.current = null; tasksRef.current = {}; setTasks({}); invalidateStoppedThreadCaches(queryClient, threadIdRef.current, isMock); @@ -1809,7 +2018,17 @@ export function useThreadStream({ () => (threadId ? history : []), [history, threadId], ); - const humanMessageCount = persistedMessages.filter( + // 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 }); @@ -1820,7 +2039,7 @@ export function useThreadStream({ // 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 localTurnOrderBaselineIdentitiesRef = useRef | null>(null); + const localTurnAnchorRef = useRef(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. @@ -1869,7 +2088,7 @@ export function useThreadStream({ }; summarizedRef.current = new Set(); pendingUsageBaselineMessageIdsRef.current = new Set(); - localTurnOrderBaselineIdentitiesRef.current = null; + localTurnAnchorRef.current = null; pendingPreparedReplayRef.current = null; setPendingSupersededRunIds(new Set()); setPendingSupersededMessageIds(new Set()); @@ -1919,9 +2138,11 @@ export function useThreadStream({ // Clear optimistic when server messages arrive. // For messages with a human optimistic message, wait until the server's - // human message has arrived to avoid clearing before canonical history (or - // replay-gap recovery) reports the input after individual messages-tuple - // events for AI messages. + // 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(() => { @@ -1938,12 +2159,12 @@ export function useThreadStream({ useEffect(() => { if ( optimisticMessageCount > 0 && - areOptimisticMessagesConfirmed(optimisticMessages, persistedMessages) + areOptimisticMessagesConfirmed(optimisticMessages, renderMessages) ) { setOptimisticMessages([]); setOptimisticThreadId(null); } - }, [optimisticMessageCount, optimisticMessages, persistedMessages]); + }, [optimisticMessageCount, optimisticMessages, renderMessages]); const sendMessage = useCallback( async ( @@ -1971,9 +2192,33 @@ export function useThreadStream({ .map(messageIdentity) .filter((id): id is string => Boolean(id)), ); - localTurnOrderBaselineIdentitiesRef.current = new Set( - pendingUsageBaselineMessageIdsRef.current, - ); + // 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( @@ -1984,7 +2229,6 @@ export function useThreadStream({ }), ); - const hideFromUI = options?.additionalKwargs?.hide_from_ui === true; const optimisticAdditionalKwargs = { ...options?.additionalKwargs, ...(optimisticFiles.length > 0 ? { files: optimisticFiles } : {}), @@ -1994,7 +2238,7 @@ export function useThreadStream({ if (!hideFromUI) { newOptimistic.push({ type: "human", - id: `opt-human-${Date.now()}`, + id: humanMessageId, content: text ? [{ type: "text", text }] : "", additional_kwargs: optimisticAdditionalKwargs, }); @@ -2101,6 +2345,7 @@ export function useThreadStream({ additionalKwargs: options?.additionalKwargs, additionalInputMessages: options?.additionalInputMessages, filesForSubmit, + humanMessageId, }), }, { @@ -2140,7 +2385,7 @@ export function useThreadStream({ setOptimisticThreadId(null); setLiveMessagesThreadId(null); setIsUploading(false); - localTurnOrderBaselineIdentitiesRef.current = null; + localTurnAnchorRef.current = null; throw error; } finally { sendInFlightRef.current = false; @@ -2153,6 +2398,7 @@ export function useThreadStream({ queryClient, humanMessageCount, persistedMessages, + visibleHistory, ], ); @@ -2178,9 +2424,27 @@ export function useThreadStream({ .map(messageIdentity) .filter((id): id is string => Boolean(id)), ); - localTurnOrderBaselineIdentitiesRef.current = new Set( - pendingUsageBaselineMessageIdsRef.current, - ); + 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; @@ -2199,6 +2463,15 @@ export function useThreadStream({ 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, @@ -2267,7 +2540,7 @@ export function useThreadStream({ setOptimisticMessages([]); setOptimisticThreadId(null); setLiveMessagesThreadId(null); - localTurnOrderBaselineIdentitiesRef.current = null; + localTurnAnchorRef.current = null; if (preparedSupersededRunId) { const supersededRunId = preparedSupersededRunId; pendingPreparedReplayRef.current = null; @@ -2284,7 +2557,14 @@ export function useThreadStream({ sendInFlightRef.current = false; } }, - [context, humanMessageCount, persistedMessages, queryClient, thread], + [ + context, + humanMessageCount, + persistedMessages, + queryClient, + thread, + visibleHistory, + ], ); const regenerateMessage = useCallback( @@ -2369,14 +2649,6 @@ export function useThreadStream({ messagesRef.current = persistedMessages; } - // Render-facing coalesced snapshot. Refs, counters and usage tracking keep - // consuming the per-chunk array above so lifecycle semantics (optimistic - // clearing, summarization capture, token-usage baselines) are unchanged. - const renderMessages = useCoalescedStreamMessages( - persistedMessages, - thread.isLoading, - ); - const rawVisibleOptimisticMessages = getVisibleOptimisticMessages( optimisticThreadId === currentViewThreadId ? optimisticMessages : [], prevHumanMsgCountRef.current, @@ -2433,10 +2705,58 @@ export function useThreadStream({ renderMessages, visibleOptimisticMessages, ); - const localTurnOrderBaseline = localTurnOrderBaselineIdentitiesRef.current; - return localTurnOrderBaseline === null + const localTurnAnchor = + localTurnAnchorRef.current?.threadId === threadId + ? 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, localTurnOrderBaseline); + : restoreLocalTurnMessageOrder( + merged, + localTurnAnchor.baselineIdentities, + confirmedHistoryIdentities, + currentTurnRunIds, + localTurnAnchor.humanIdentity, + canonicalHistoryIdentities, + ); }, [ previouslyRenderedOrder, renderMessages, @@ -2446,10 +2766,27 @@ export function useThreadStream({ visibleOptimisticMessages, ]); useEffect(() => { - const visibleMergedMessages = mergedMessages.filter( - (message) => - !isHiddenFromUIMessage(message) && !message.id?.startsWith("opt-"), + // 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 @@ -2467,7 +2804,15 @@ export function useThreadStream({ .map(messageIdentity) .filter(isNonEmptyString), }; - }, [mergedMessages, pendingSupersededMessageIds, thread.isLoading, threadId]); + }, [ + mergedMessages, + optimisticMessages, + optimisticThreadId, + currentViewThreadId, + pendingSupersededMessageIds, + thread.isLoading, + threadId, + ]); const pendingUsageMessages = thread.isLoading ? getMessagesAfterBaseline( persistedMessages, diff --git a/frontend/tests/unit/core/threads/local-turn-order.dom.test.tsx b/frontend/tests/unit/core/threads/local-turn-order.dom.test.tsx index c53b0a57a..f2f64e562 100644 --- a/frontend/tests/unit/core/threads/local-turn-order.dom.test.tsx +++ b/frontend/tests/unit/core/threads/local-turn-order.dom.test.tsx @@ -1,12 +1,14 @@ import type { Message } from "@langchain/langgraph-sdk"; -import { expect, rs, test } from "@rstest/core"; +import { afterEach, beforeEach, expect, rs, test } from "@rstest/core"; import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; import { act, renderHook } from "@testing-library/react"; import { createElement, type ReactNode } from "react"; import { I18nContext } from "@/core/i18n/context"; import { enUS } from "@/core/i18n/locales/en-US"; +import { isHiddenFromUIMessage } from "@/core/messages/utils"; import { DEFAULT_LOCAL_SETTINGS } from "@/core/settings/local"; +import type { RunMessage } from "@/core/threads/types"; const streamMockState = rs.hoisted(() => ({ isLoading: false, @@ -14,6 +16,11 @@ const streamMockState = rs.hoisted(() => ({ onFinish: undefined as | ((state: { values: { messages: Message[] } }) => void) | undefined, + onError: undefined as ((error: unknown) => void) | undefined, + onUpdateEvent: undefined as + | ((data: unknown, options: { mutate: (updater: unknown) => void }) => void) + | undefined, + onCustomEvent: undefined as ((event: unknown) => void) | undefined, stop: rs.fn(async () => undefined), submit: rs.fn(async () => undefined), })); @@ -21,8 +28,17 @@ const streamMockState = rs.hoisted(() => ({ rs.mock("@langchain/langgraph-sdk/react", () => ({ useStream: (options: { onFinish?: (state: { values: { messages: Message[] } }) => void; + onError?: (error: unknown) => void; + onUpdateEvent?: ( + data: unknown, + options: { mutate: (updater: unknown) => void }, + ) => void; + onCustomEvent?: (event: unknown) => void; }) => { streamMockState.onFinish = options.onFinish; + streamMockState.onError = options.onError; + streamMockState.onUpdateEvent = options.onUpdateEvent; + streamMockState.onCustomEvent = options.onCustomEvent; return { isLoading: streamMockState.isLoading, messages: streamMockState.messages, @@ -38,13 +54,77 @@ rs.mock("@langchain/langgraph-sdk/react", () => ({ }, })); -test("keeps early streamed steps behind a local user message after finish", async () => { - const { useThreadStream } = await import("@/core/threads/hooks"); - const queryClient = new QueryClient({ - defaultOptions: { queries: { retry: false } }, +function humanMessage( + id: string, + text: string, + extra?: Record, +): Message { + return { + id, + type: "human", + content: [{ type: "text", text }], + ...(extra ? { additional_kwargs: extra } : {}), + } as Message; +} + +function aiMessage( + id: string, + text: string, + extra?: { run_id?: string; seq?: number }, +): Message { + return { + id, + type: "ai", + content: text, + ...(extra?.run_id ? { run_id: extra.run_id } : {}), + ...(extra?.seq !== undefined + ? { additional_kwargs: { deerflow_seq: extra.seq } } + : {}), + } as Message; +} + +function historyRow(seq: number, runId: string, content: Message): RunMessage { + return { + run_id: runId, + seq, + content, + metadata: { caller: "" }, + created_at: "2026-09-08T00:00:00Z", + } as RunMessage; +} + +function historyPageResponse(rows: RunMessage[]): Response { + return new Response( + JSON.stringify({ data: rows, has_more: false, next_before_seq: null }), + { status: 200, headers: { "Content-Type": "application/json" } }, + ); +} + +/** Serve per-thread history pages through the wrapped API fetcher. */ +function stubHistoryFetch(pagesByThread: Record RunMessage[]>) { + rs.stubGlobal("fetch", async (input: unknown) => { + const url = String(input); + const pageMatch = /\/api\/threads\/([^/]+)\/messages\/page/.exec(url); + if (pageMatch) { + return historyPageResponse(pagesByThread[pageMatch[1]!]?.() ?? []); + } + throw new Error(`Unexpected fetch: ${url}`); }); - const wrapper = ({ children }: { children: ReactNode }) => - createElement( +} + +function visibleMessageIds(messages: Message[]): string[] { + return messages + .filter((message) => !isHiddenFromUIMessage(message)) + .map((message) => String(message.id)); +} + +function createWrapper(queryClient: QueryClient) { + return function ThreadStreamTestWrapper({ + children, + }: { + children: ReactNode; + }) { + return createElement( QueryClientProvider, { client: queryClient }, createElement( @@ -59,6 +139,129 @@ test("keeps early streamed steps behind a local user message after finish", asyn children, ), ); + }; +} + +async function flushFrames() { + // React Query delivers cache updates through setTimeout(0) and the history + // retained-rows effect adds another render cycle, so a flush fires due + // 0ms timers (never the 80ms coalesce trailing flush) and lets the render + // chain settle — a fixed number of rounds, never a real sleep. + for (let index = 0; index < 6; index += 1) { + await act(async () => { + if (rs.isFakeTimers()) { + await rs.advanceTimersByTimeAsync(0); + } + }); + } +} + +beforeEach(() => { + streamMockState.isLoading = false; + streamMockState.messages = []; + streamMockState.onFinish = undefined; + streamMockState.onError = undefined; + streamMockState.onUpdateEvent = undefined; + streamMockState.onCustomEvent = undefined; + streamMockState.stop.mockClear(); + streamMockState.submit.mockClear(); +}); + +afterEach(() => { + rs.useRealTimers(); + rs.unstubAllGlobals(); +}); + +// Compaction trimmed the checkpoint tail to the latest turn while the REST +// history feed still holds both turns: A更早(1), H旧(2), H近(3), A近(4). +const EARLIER_ANSWER = aiMessage("earlier-answer", "Earlier answer"); +const OLD_HUMAN = humanMessage("old-human", "An older request"); +const RECENT_HUMAN = humanMessage("recent-human", "The recent request"); +const RECENT_ANSWER = aiMessage("recent-answer", "The recent answer"); + +function seededHistoryRows(): RunMessage[] { + return [ + historyRow(1, "run-earlier", EARLIER_ANSWER), + historyRow(2, "run-old", OLD_HUMAN), + historyRow(3, "run-recent", RECENT_HUMAN), + historyRow(4, "run-recent", RECENT_ANSWER), + ]; +} + +function checkpointMessages(): Message[] { + return [ + { + id: "recent-human", + type: "human", + content: [{ type: "text", text: "The recent request" }], + additional_kwargs: { deerflow_seq: 3 }, + } as Message, + aiMessage("recent-answer", "The recent answer", { seq: 4 }), + ]; +} + +async function renderSeededThread(options?: { + threadId?: string; + isMock?: boolean; + historyRows?: () => RunMessage[]; +}) { + const threadId = options?.threadId ?? "thread-1"; + if (!options?.isMock) { + stubHistoryFetch({ + [threadId]: options?.historyRows ?? (() => []), + }); + } + const { useThreadStream } = await import("@/core/threads/hooks"); + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + const hook = renderHook( + ({ currentThreadId }: { currentThreadId?: string } = {}) => + useThreadStream({ + context: DEFAULT_LOCAL_SETTINGS.context, + isMock: options?.isMock ?? false, + threadId: currentThreadId ?? threadId, + }), + { wrapper: createWrapper(queryClient) }, + ); + await flushFrames(); + return hook; +} + +/** Submit one visible turn and return the id shared by display and submit. */ +async function submitVisibleTurn( + result: { + current: { + sendMessage: ( + threadId: string, + message: { files: []; text: string }, + ) => Promise; + thread: { messages: Message[] }; + }; + }, + text = "Continue the work", +): Promise { + await act(async () => { + await result.current.sendMessage("thread-1", { files: [], text }); + }); + const displayed = result.current.thread.messages.at(-1); + const submittedId = displayed?.id; + expect(typeof submittedId).toBe("string"); + // The submit mock is an untyped rs.fn(); cast the call list once so the + // tuple indexing below typechecks. + const submitCalls = streamMockState.submit.mock.calls as unknown as Array< + [{ messages: Message[] }] + >; + const submittedInput = submitCalls.at(-1)?.[0]; + expect(submittedInput?.messages.at(-1)?.id).toBe(submittedId); + return submittedId!; +} + +test("keeps early streamed steps behind a local user message after finish", async () => { + const { useThreadStream } = await import("@/core/threads/hooks"); + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); const { rerender, result } = renderHook( () => useThreadStream({ @@ -66,23 +269,20 @@ test("keeps early streamed steps behind a local user message after finish", asyn isMock: true, threadId: "thread-1", }), - { wrapper }, + { wrapper: createWrapper(queryClient) }, ); - await act(async () => { - await result.current.sendMessage("thread-1", { - files: [], - text: "Build a presentation", - }); - }); + const submittedId = await submitVisibleTurn(result, "Build a presentation"); const earlyAssistantStep = { id: "early-assistant-step", type: "ai", content: "Reading the presentation skill", } as Message; + // The server keeps the client-submitted id and the runtime injects the + // visible copy as `__user`; both normalize onto the same identity. const injectedHuman = { - id: "current-request__user", + id: `${submittedId}__user`, type: "human", content: "Build a presentation", } as Message; @@ -108,3 +308,844 @@ test("keeps early streamed steps behind a local user message after finish", asyn earlyAssistantStep, ]); }); + +test("keeps established history order while the submitted human is outside the render snapshot", async () => { + rs.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + let rows = seededHistoryRows(); + streamMockState.messages = checkpointMessages(); + const { rerender, result } = await renderSeededThread({ + historyRows: () => rows, + }); + + // Frame 0: history beyond the checkpoint baseline renders in feed order. + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + ]); + + const submittedId = await submitVisibleTurn(result); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + submittedId, + ]); + + // Frame 1 (leading-edge flush): the first AI step of the new turn arrives + // before the server echo of the human input. + const newStep = aiMessage("new-step", "Working on the follow-up", { + run_id: "run-new", + }); + streamMockState.messages = [RECENT_HUMAN, RECENT_ANSWER, newStep]; + streamMockState.isLoading = true; + rerender(); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + submittedId, + "new-step", + ]); + + // Frame 2: the server human copy reaches the unthrottled SDK state, but the + // ~80ms render snapshot cannot see it yet. The established history must keep + // its order and the local input must stay visible — no frame may drop the + // current turn's human or relocate the history-only `old-human`. + const serverHuman = { + id: `${submittedId}__user`, + type: "human", + content: "Continue the work", + run_id: "run-new", + } as Message; + streamMockState.messages = [ + RECENT_HUMAN, + RECENT_ANSWER, + newStep, + serverHuman, + ]; + rerender(); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + submittedId, + "new-step", + ]); + + // Frame 3: the throttled snapshot catches up; the confirmed server copy + // takes over the same identity exactly once. + await act(async () => { + await rs.advanceTimersByTimeAsync(100); + }); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + `${submittedId}__user`, + "new-step", + ]); + + // Frame 4: finish triggers the history refetch; the canonical page now + // commits the whole turn and the display converges to seq order. + rows = [ + ...seededHistoryRows(), + historyRow(5, "run-new", serverHuman), + historyRow(6, "run-new", newStep), + ]; + await act(async () => { + streamMockState.onFinish?.({ + values: { messages: streamMockState.messages }, + }); + streamMockState.isLoading = false; + }); + rerender(); + await flushFrames(); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + `${submittedId}__user`, + "new-step", + ]); +}); + +test("keeps a pre-submit older-turn rescue above the submitted human", async () => { + rs.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const rescuedOldHuman = { + id: "rescued-old-human", + type: "human", + content: "Older request", + run_id: "run-old", + additional_kwargs: { deerflow_seq: 1 }, + } as Message; + const rescuedOldAnswer = aiMessage("rescued-old-answer", "Older answer", { + run_id: "run-old", + seq: 2, + }); + const recentHuman = { + id: "recent-human", + type: "human", + content: "Recent request", + run_id: "run-recent", + additional_kwargs: { deerflow_seq: 3 }, + } as Message; + const recentAnswer = aiMessage("recent-answer", "Recent answer", { + run_id: "run-recent", + seq: 4, + }); + const rows = [ + historyRow(3, "run-recent", recentHuman), + historyRow(4, "run-recent", recentAnswer), + ]; + streamMockState.messages = [ + rescuedOldHuman, + rescuedOldAnswer, + recentHuman, + recentAnswer, + ]; + const { rerender, result } = await renderSeededThread({ + historyRows: () => rows, + }); + + const removeAll = { + id: "__remove_all__", + type: "remove", + content: "", + } as Message; + const hiddenSummary = humanMessage("summary-old", "Conversation summary", { + hide_from_ui: true, + }); + act(() => { + streamMockState.onUpdateEvent?.( + { + "DeerFlowSummarizationMiddleware.before_model": { + messages: [removeAll, hiddenSummary, recentHuman, recentAnswer], + }, + }, + { mutate: () => undefined }, + ); + streamMockState.messages = [hiddenSummary, recentHuman, recentAnswer]; + rerender(); + }); + await act(async () => { + await rs.advanceTimersByTimeAsync(100); + }); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "rescued-old-human", + "rescued-old-answer", + "recent-human", + "recent-answer", + ]); + + await act(async () => { + await result.current.sendMessage("thread-1", { + files: [], + text: "Continue the work", + }); + }); + const submitCalls = streamMockState.submit.mock.calls as unknown as Array< + [{ messages: Message[] }] + >; + const submittedId = submitCalls.at(-1)?.[0].messages.at(-1)?.id; + expect(typeof submittedId).toBe("string"); + const serverHuman = { + id: `${submittedId}__user`, + type: "human", + content: "Continue the work", + run_id: "run-current", + } as Message; + const currentStep = aiMessage("current-step", "Current step", { + run_id: "run-current", + }); + streamMockState.messages = [ + hiddenSummary, + recentHuman, + recentAnswer, + serverHuman, + currentStep, + ]; + streamMockState.isLoading = true; + rerender(); + await act(async () => { + await rs.advanceTimersByTimeAsync(100); + }); + + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "rescued-old-human", + "rescued-old-answer", + "recent-human", + "recent-answer", + `${submittedId}__user`, + "current-step", + ]); +}); + +test("keeps a transiently rescued current-turn step behind its submitted human", async () => { + rs.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const previousHuman = humanMessage("previous-human", "Previous request"); + const previousAnswer = aiMessage("previous-answer", "Previous answer", { + run_id: "run-previous", + }); + const rows = [ + historyRow(1, "run-previous", previousHuman), + historyRow(2, "run-previous", previousAnswer), + ]; + streamMockState.messages = [previousHuman, previousAnswer]; + const { rerender, result } = await renderSeededThread({ + historyRows: () => rows, + }); + + const submittedId = await submitVisibleTurn(result); + const serverHuman = { + id: `${submittedId}__user`, + type: "human", + content: "Continue the work", + run_id: "run-current", + } as Message; + // Deliberately omit run_id: this is the streamed step whose ownership has + // not yet been restored by canonical history. + const rescuedStep = aiMessage("rescued-step", "First current step"); + const retainedStep = aiMessage("retained-step", "Latest current step", { + run_id: "run-current", + }); + streamMockState.messages = [serverHuman, rescuedStep, retainedStep]; + streamMockState.isLoading = true; + rerender(); + await act(async () => { + await rs.advanceTimersByTimeAsync(100); + }); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "previous-human", + "previous-answer", + `${submittedId}__user`, + "rescued-step", + "retained-step", + ]); + + const removeAll = { + id: "__remove_all__", + type: "remove", + content: "", + } as Message; + const hiddenSummary = humanMessage( + "summary-current", + "Conversation summary", + { hide_from_ui: true }, + ); + act(() => { + streamMockState.onUpdateEvent?.( + { + "DeerFlowSummarizationMiddleware.before_model": { + messages: [removeAll, hiddenSummary, serverHuman, retainedStep], + }, + }, + { mutate: () => undefined }, + ); + streamMockState.messages = [hiddenSummary, serverHuman, retainedStep]; + rerender(); + }); + await act(async () => { + await rs.advanceTimersByTimeAsync(100); + }); + + // The rescued step is transient current-turn state, not canonical old + // history, so it must remain behind the submitted human. + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "previous-human", + "previous-answer", + `${submittedId}__user`, + "rescued-step", + "retained-step", + ]); +}); + +test("does not move a later externally submitted answer ahead of the retained local anchor", async () => { + rs.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const previousHuman = humanMessage("previous-human", "Previous request"); + const previousAnswer = aiMessage("previous-answer", "Previous answer", { + run_id: "run-previous", + }); + let rows = [ + historyRow(1, "run-previous", previousHuman), + historyRow(2, "run-previous", previousAnswer), + ]; + streamMockState.messages = [previousHuman, previousAnswer]; + const { rerender, result } = await renderSeededThread({ + historyRows: () => rows, + }); + + const submittedId = await submitVisibleTurn(result); + const localHuman = { + id: `${submittedId}__user`, + type: "human", + content: "Continue the work", + run_id: "run-local", + } as Message; + const localAnswer = aiMessage("local-answer", "Local answer", { + run_id: "run-local", + }); + streamMockState.messages = [localHuman, localAnswer]; + streamMockState.isLoading = true; + rerender(); + await act(async () => { + await rs.advanceTimersByTimeAsync(100); + }); + + const remoteHuman = humanMessage("remote-human", "Later request elsewhere"); + const remoteAnswer = aiMessage("remote-answer", "Later external answer", { + run_id: "run-remote", + }); + rows = [ + historyRow(1, "run-previous", previousHuman), + historyRow(2, "run-previous", previousAnswer), + historyRow(3, "run-local", localHuman), + historyRow(4, "run-local", localAnswer), + historyRow(5, "run-remote", remoteHuman), + historyRow(6, "run-remote", remoteAnswer), + ]; + await act(async () => { + streamMockState.onFinish?.({ + values: { messages: streamMockState.messages }, + }); + streamMockState.isLoading = false; + }); + rerender(); + await flushFrames(); + + // The local anchor survives finish only to repair its own turn. Canonical + // messages submitted later by another client must retain their feed order. + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "previous-human", + "previous-answer", + `${submittedId}__user`, + "local-answer", + "remote-human", + "remote-answer", + ]); +}); + +test("hidden human-input reply never borrows an older visible human as its turn anchor", async () => { + const rows = seededHistoryRows(); + streamMockState.messages = checkpointMessages(); + const { rerender, result } = await renderSeededThread({ + historyRows: () => rows, + }); + + // A clarification answer submits hidden: no visible optimistic human exists + // for this turn at all. + await act(async () => { + await result.current.sendMessage( + "thread-1", + { files: [], text: "For your clarification, my answer is: staging" }, + undefined, + { additionalKwargs: { hide_from_ui: true } }, + ); + }); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + ]); + + const hiddenReply = humanMessage( + "hidden-reply", + "For your clarification, my answer is: staging", + { hide_from_ui: true }, + ); + const replyStep = aiMessage("reply-step", "Applying the answer", { + run_id: "run-reply", + }); + streamMockState.messages = [ + RECENT_HUMAN, + RECENT_ANSWER, + hiddenReply, + replyStep, + ]; + streamMockState.isLoading = true; + rerender(); + + // The hidden turn must not re-anchor on `old-human`: visible history order + // is already correct and stays untouched, the new step appends at the tail. + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + "reply-step", + ]); +}); + +test("keeps the new turn in place when the history page resolves after submit", async () => { + rs.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + let resolveHistory: ((response: Response) => void) | undefined; + rs.stubGlobal( + "fetch", + () => + new Promise((resolve) => { + resolveHistory = resolve; + }), + ); + streamMockState.messages = checkpointMessages(); + const { useThreadStream } = await import("@/core/threads/hooks"); + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + const { rerender, result } = renderHook( + () => + useThreadStream({ + context: DEFAULT_LOCAL_SETTINGS.context, + isMock: false, + threadId: "thread-1", + }), + { wrapper: createWrapper(queryClient) }, + ); + await flushFrames(); + + // History is still loading: only the checkpoint tail is visible. + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "recent-human", + "recent-answer", + ]); + + const submittedId = await submitVisibleTurn(result); + const newStep = aiMessage("new-step", "Working on the follow-up", { + run_id: "run-new", + }); + streamMockState.messages = [RECENT_HUMAN, RECENT_ANSWER, newStep]; + streamMockState.isLoading = true; + rerender(); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "recent-human", + "recent-answer", + submittedId, + "new-step", + ]); + + // The page resolves mid-turn: older turns slot in ahead of the established + // anchor without disturbing the in-flight turn. + await act(async () => { + resolveHistory?.(historyPageResponse(seededHistoryRows())); + }); + await flushFrames(); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + submittedId, + "new-step", + ]); +}); + +test("clears the local turn anchor when switching threads before the pending flush", async () => { + rs.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + const thread2Human = humanMessage("t2-human", "Another thread request"); + const thread2Answer = aiMessage("t2-answer", "Another thread answer"); + streamMockState.messages = checkpointMessages(); + const { rerender, result } = await renderSeededThread({ + historyRows: () => seededHistoryRows(), + }); + + const submittedId = await submitVisibleTurn(result); + const newStep = aiMessage("new-step", "Working on the follow-up", { + run_id: "run-new", + }); + streamMockState.messages = [RECENT_HUMAN, RECENT_ANSWER, newStep]; + streamMockState.isLoading = true; + rerender(); + expect(visibleMessageIds(result.current.thread.messages)).toContain( + submittedId, + ); + + // Switch threads while a trailing coalesce flush is still pending. The + // other thread's view must not inherit this turn's anchor or messages. + stubHistoryFetch({ + "thread-1": () => seededHistoryRows(), + "thread-2": () => [ + historyRow(1, "run-t2", thread2Human), + historyRow(2, "run-t2", thread2Answer), + ], + }); + streamMockState.messages = [thread2Human, thread2Answer]; + streamMockState.isLoading = false; + rerender({ currentThreadId: "thread-2" }); + await flushFrames(); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "t2-human", + "t2-answer", + ]); + + // The pending timer from the previous thread's stream must not repaint its + // messages into this view either. + await act(async () => { + await rs.advanceTimersByTimeAsync(200); + }); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "t2-human", + "t2-answer", + ]); +}); + +test("releases the in-flight turn state when the submit fails", async () => { + const rows = seededHistoryRows(); + streamMockState.messages = checkpointMessages(); + const { rerender, result } = await renderSeededThread({ + historyRows: () => rows, + }); + + streamMockState.submit.mockRejectedValueOnce(new Error("network down")); + await act(async () => { + await result.current + .sendMessage("thread-1", { files: [], text: "This send fails" }) + .catch(() => undefined); + }); + + // The failed input is withdrawn and must not linger through any ledger. + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + ]); + + // The in-flight guard is released: the next send proceeds and anchors on + // its own fresh identity, never on the failed one. + const nextId = await submitVisibleTurn(result, "Retry the request"); + const newStep = aiMessage("new-step", "Working on the retry", { + run_id: "run-retry", + }); + streamMockState.messages = [RECENT_HUMAN, RECENT_ANSWER, newStep]; + streamMockState.isLoading = true; + rerender(); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + nextId, + "new-step", + ]); +}); + +test("anchors an edit-regenerate replay on the replacement human identity", async () => { + const threadHuman = humanMessage("srv-h1", "Original question"); + const threadAnswer = aiMessage("srv-a1", "Original answer"); + const rows = [ + historyRow(1, "run-1", threadHuman), + historyRow(2, "run-1", threadAnswer), + ]; + streamMockState.messages = [threadHuman, threadAnswer]; + + const replacementHuman = humanMessage("repl-h1", "Edited question"); + rs.stubGlobal("fetch", async (input: unknown, _init?: RequestInit) => { + const url = String(input); + if (url.includes("/runs/edit-regenerate/prepare")) { + return new Response( + JSON.stringify({ + input: { messages: [replacementHuman] }, + checkpoint: { + checkpoint_ns: "", + checkpoint_id: "cp-1", + checkpoint_map: null, + }, + metadata: {}, + target_run_id: "run-1", + replacement_human_message_id: "repl-h1", + source_message_ids: ["srv-h1", "srv-a1"], + }), + { status: 200, headers: { "Content-Type": "application/json" } }, + ); + } + if (url.includes("/messages/page")) { + return historyPageResponse(rows); + } + throw new Error(`Unexpected fetch: ${url}`); + }); + + const { useThreadStream } = await import("@/core/threads/hooks"); + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + const { rerender, result } = renderHook( + () => + useThreadStream({ + context: DEFAULT_LOCAL_SETTINGS.context, + isMock: false, + threadId: "thread-1", + }), + { wrapper: createWrapper(queryClient) }, + ); + await flushFrames(); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "srv-h1", + "srv-a1", + ]); + + await act(async () => { + await result.current.editAndRegenerateMessage( + "thread-1", + "srv-h1", + "Edited question", + ); + }); + + // The superseded turn is masked; the replacement human from the prepare + // response is shown optimistically with its server-assigned identity. + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "repl-h1", + ]); + + // The replay's first step arrives before the replacement human echo: the + // replay anchors on the prepare identity, not on any older human. + const replayStep = aiMessage("repl-step", "Recomputing the answer", { + run_id: "run-2", + }); + streamMockState.messages = [replayStep]; + streamMockState.isLoading = true; + rerender(); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "repl-h1", + "repl-step", + ]); + + // Once the server copy lands, the same identity is shown exactly once and + // the optimistic copy is withdrawn. + const serverReplacement = { + id: "repl-h1__user", + type: "human", + content: "Edited question", + run_id: "run-2", + } as Message; + streamMockState.messages = [replayStep, serverReplacement]; + streamMockState.isLoading = false; + rerender(); + await flushFrames(); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "repl-h1__user", + "repl-step", + ]); +}); + +test("drops the local turn anchor after a stream replay gap", async () => { + const rows = seededHistoryRows(); + streamMockState.messages = checkpointMessages(); + const { rerender, result } = await renderSeededThread({ + historyRows: () => rows, + }); + + const submittedId = await submitVisibleTurn(result); + expect(visibleMessageIds(result.current.thread.messages)).toContain( + submittedId, + ); + + // A replay gap invalidates every locally kept ordering assumption. + act(() => { + streamMockState.onCustomEvent?.({ type: "stream_replay_gap" }); + }); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + ]); + + // Afterwards, reconnection semantics apply: an arriving step with no local + // anchor left must not reorder the established history. + const newStep = aiMessage("new-step", "Working on the follow-up", { + run_id: "run-new", + }); + streamMockState.messages = [RECENT_HUMAN, RECENT_ANSWER, newStep]; + streamMockState.isLoading = true; + rerender(); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + "new-step", + ]); +}); + +test("keeps the submitted turn anchored through stop and the final history refresh", async () => { + rs.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + let rows = seededHistoryRows(); + streamMockState.messages = checkpointMessages(); + const { rerender, result } = await renderSeededThread({ + historyRows: () => rows, + }); + + const submittedId = await submitVisibleTurn(result); + const newStep = aiMessage("new-step", "Working on the follow-up", { + run_id: "run-new", + }); + const serverHuman = { + id: `${submittedId}__user`, + type: "human", + content: "Continue the work", + run_id: "run-new", + } as Message; + streamMockState.messages = [ + RECENT_HUMAN, + RECENT_ANSWER, + newStep, + serverHuman, + ]; + streamMockState.isLoading = true; + rerender(); + await act(async () => { + await rs.advanceTimersByTimeAsync(100); + }); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + `${submittedId}__user`, + "new-step", + ]); + + // Stop flushes the turn into canonical history; the refresh must converge + // to the same order the stream already showed. + rows = [ + ...seededHistoryRows(), + historyRow(5, "run-new", serverHuman), + historyRow(6, "run-new", newStep), + ]; + await act(async () => { + await result.current.thread.stop(); + streamMockState.isLoading = false; + }); + rerender(); + await flushFrames(); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + `${submittedId}__user`, + "new-step", + ]); + + // The delayed finalization refetch changes nothing. + await act(async () => { + await rs.advanceTimersByTimeAsync(1600); + }); + await flushFrames(); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + `${submittedId}__user`, + "new-step", + ]); +}); + +test("keeps history stable when the stream errors after the human was confirmed", async () => { + rs.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] }); + let rows = seededHistoryRows(); + streamMockState.messages = checkpointMessages(); + const { rerender, result } = await renderSeededThread({ + historyRows: () => rows, + }); + + const submittedId = await submitVisibleTurn(result); + const newStep = aiMessage("new-step", "Working on the follow-up", { + run_id: "run-new", + }); + const serverHuman = { + id: `${submittedId}__user`, + type: "human", + content: "Continue the work", + run_id: "run-new", + } as Message; + streamMockState.messages = [ + RECENT_HUMAN, + RECENT_ANSWER, + newStep, + serverHuman, + ]; + streamMockState.isLoading = true; + rerender(); + await act(async () => { + await rs.advanceTimersByTimeAsync(100); + }); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + `${submittedId}__user`, + "new-step", + ]); + + // The stream errors; the error path invalidates history and the refreshed + // canonical page keeps the turn exactly where the user saw it. + rows = [ + ...seededHistoryRows(), + historyRow(5, "run-new", serverHuman), + historyRow(6, "run-new", newStep), + ]; + await act(async () => { + streamMockState.onError?.(new Error("stream broke")); + streamMockState.isLoading = false; + }); + rerender(); + await flushFrames(); + expect(visibleMessageIds(result.current.thread.messages)).toEqual([ + "earlier-answer", + "old-human", + "recent-human", + "recent-answer", + `${submittedId}__user`, + "new-step", + ]); +}); diff --git a/frontend/tests/unit/core/threads/message-merge.test.ts b/frontend/tests/unit/core/threads/message-merge.test.ts index 50f8f30ab..7d698e79a 100644 --- a/frontend/tests/unit/core/threads/message-merge.test.ts +++ b/frontend/tests/unit/core/threads/message-merge.test.ts @@ -10,6 +10,7 @@ import { computeSummarizationTransientMessages, countHumanMessagesExcludingSuperseded, flattenThreadHistoryPages, + getCurrentTurnRunIds, getSummarizationMiddlewareMessages, getThreadHistoryNextPageParam, getVisibleOptimisticMessages, @@ -1637,6 +1638,472 @@ test("local turn order keeps early streamed steps behind the user message", () = ]); }); +test("local turn order keeps an existing clarification card with its original turn", () => { + const previousHuman = { + id: "previous-human", + type: "human", + content: "Research today's market", + } as Message; + const previousAnswer = { + id: "previous-answer", + type: "ai", + content: "Here is the completed report", + } as Message; + const clarificationCard = { + id: "clarification-card", + type: "tool", + name: "ask_clarification", + tool_call_id: "clarification-call", + content: "Which market should I research?", + } as Message; + const currentHuman = { + id: "current-human", + type: "human", + content: "Summarize the report", + } as Message; + const currentStep = { + id: "current-step", + type: "ai", + content: "Reading the report", + } as Message; + const baselineIdentities = new Set([ + "message:previous-human", + "message:previous-answer", + "tool:clarification-call", + ]); + + // A live checkpoint tail can be woven after the newly persisted human + // message even though the card was already visible before submission. + expect( + restoreLocalTurnMessageOrder( + [ + previousHuman, + previousAnswer, + currentHuman, + clarificationCard, + currentStep, + ], + baselineIdentities, + ), + ).toEqual([ + previousHuman, + previousAnswer, + clarificationCard, + currentHuman, + currentStep, + ]); +}); + +test("local turn order preserves relative order across displaced message sources", () => { + const previousHuman = { + id: "previous-human", + type: "human", + content: "Research today's market", + } as Message; + const currentHuman = { + id: "current-human", + type: "human", + content: "Summarize the report", + } as Message; + const historyOnlyCard = { + id: "history-only-card", + type: "tool", + name: "ask_clarification", + tool_call_id: "history-only-call", + content: "Which market should I research?", + } as Message; + const baselineAnswer = { + id: "baseline-answer", + type: "ai", + content: "Here is the completed report", + } as Message; + const currentStep = { + id: "current-step", + type: "ai", + content: "Reading the report", + } as Message; + + expect( + restoreLocalTurnMessageOrder( + [ + previousHuman, + currentHuman, + historyOnlyCard, + baselineAnswer, + currentStep, + ], + new Set(["message:previous-human", "message:baseline-answer"]), + new Set(["tool:history-only-call"]), + ), + ).toEqual([ + previousHuman, + historyOnlyCard, + baselineAnswer, + currentHuman, + currentStep, + ]); +}); + +test("current turn run id derivation selects only visible unconfirmed steps", () => { + const baselineStep = { + id: "baseline-step", + type: "ai", + content: "Previous answer", + run_id: "run-previous", + } as Message; + const currentStep = { + id: "current-step", + type: "ai", + content: "Current progress", + run_id: "run-current", + } as Message; + const runlessStep = { + id: "runless-step", + type: "tool", + content: "Pending metadata", + tool_call_id: "runless-call", + } as Message; + const hiddenStep = { + id: "hidden-step", + type: "ai", + content: "Internal control", + run_id: "run-hidden", + additional_kwargs: { hide_from_ui: true }, + } as Message; + const confirmedHistoryStep = { + id: "confirmed-history-step", + type: "tool", + content: "Previously persisted result", + tool_call_id: "confirmed-history-call", + run_id: "run-history", + } as Message; + + expect( + getCurrentTurnRunIds( + [ + baselineStep, + currentStep, + runlessStep, + hiddenStep, + confirmedHistoryStep, + ], + new Set(["message:baseline-step"]), + new Set(["tool:confirmed-history-call"]), + ), + ).toEqual(new Set(["run-current"])); + expect( + getCurrentTurnRunIds([runlessStep], new Set(["message:baseline-step"])), + ).toEqual(new Set()); + expect(getCurrentTurnRunIds([currentStep], null)).toEqual(new Set()); +}); + +test("history-confirmed live cards are not derived as current-run steps", () => { + const previousHuman = { + id: "previous-human", + type: "human", + content: "Research today's market", + } as Message; + const currentHuman = { + id: "current-human", + type: "human", + content: "Summarize the report", + run_id: "run-current", + } as Message; + const oldHistoryCard = { + id: "old-history-card", + type: "tool", + name: "ask_clarification", + tool_call_id: "old-history-call", + content: "Which market should I research?", + run_id: "run-previous", + } as Message; + const baselineIdentities = new Set(["message:previous-human"]); + const historyIdentities = new Set(["tool:old-history-call"]); + const currentTurnRunIds = getCurrentTurnRunIds( + [oldHistoryCard], + baselineIdentities, + historyIdentities, + ); + + expect(currentTurnRunIds).toEqual(new Set()); + expect( + restoreLocalTurnMessageOrder( + [previousHuman, currentHuman, oldHistoryCard], + baselineIdentities, + historyIdentities, + currentTurnRunIds, + ), + ).toEqual([previousHuman, oldHistoryCard, currentHuman]); +}); + +test("local turn order repairs a displaced suffix when the human is first", () => { + const currentHuman = { + id: "current-human", + type: "human", + content: "Summarize the report", + run_id: "run-current", + } as Message; + const oldHistoryCard = { + id: "old-history-card", + type: "tool", + name: "ask_clarification", + tool_call_id: "old-history-call", + content: "Which market should I research?", + run_id: "run-previous", + } as Message; + + expect( + restoreLocalTurnMessageOrder( + [currentHuman, oldHistoryCard], + new Set(), + new Set(["tool:old-history-call"]), + ), + ).toEqual([oldHistoryCard, currentHuman]); +}); + +test("local turn order anchors on the latest non-baseline human after compaction", () => { + // A summarized checkpoint may omit an older human that remains in canonical + // history. The latest non-baseline human is the submitted turn even before + // the server assigns it an id; choosing the first identifiable one would move + // established intervening turns around the old human. + const oldHistoryHuman = { + id: "old-history-human", + type: "human", + content: "An earlier request omitted by the checkpoint", + } as Message; + const oldHistoryAnswer = { + id: "old-history-answer", + type: "ai", + content: "An earlier answer", + } as Message; + const previousHuman = { + id: "previous-human", + type: "human", + content: "The immediately previous request", + } as Message; + const previousAnswer = { + id: "previous-answer", + type: "ai", + content: "The immediately previous answer", + } as Message; + const currentHuman = { + type: "human", + content: "The newly submitted request", + } as Message; + const currentStep = { + id: "current-step", + type: "ai", + content: "Current streamed progress", + } as Message; + + expect( + restoreLocalTurnMessageOrder( + [ + oldHistoryHuman, + oldHistoryAnswer, + previousHuman, + previousAnswer, + currentHuman, + currentStep, + ], + new Set(["message:previous-human", "message:previous-answer"]), + new Set(["message:old-history-answer"]), + ), + ).toEqual([ + oldHistoryHuman, + oldHistoryAnswer, + previousHuman, + previousAnswer, + currentHuman, + currentStep, + ]); +}); + +test("history-confirmed current steps still move behind their human", () => { + const previousHuman = { + id: "previous-human", + type: "human", + content: "Research the robot sector", + } as Message; + const currentStep = { + id: "current-step", + type: "ai", + content: "Searching for catalysts", + run_id: "run-current", + } as Message; + const currentHuman = { + id: "current-human", + type: "human", + content: "Continue tracking the robot sector", + run_id: "run-current", + } as Message; + + expect( + restoreLocalTurnMessageOrder( + [previousHuman, currentStep, currentHuman], + new Set(["message:previous-human"]), + new Set(["message:current-step"]), + ), + ).toEqual([previousHuman, currentHuman, currentStep]); +}); + +test("local turn order keeps the current run's persisted steps after the new human", () => { + // After an interrupt/stop, the current turn's already-executed steps are + // flushed into canonical history. They belong AFTER the new human input even + // though history now confirms them — treating them as displaced old message + // would move the current run's own progress above the message that owns it. + const previousHuman = { + id: "previous-human", + type: "human", + content: "Research the robot sector", + } as Message; + const currentHuman = { + id: "current-human", + type: "human", + content: "Continue tracking the robot sector", + run_id: "run-current", + } as Message; + const currentAnsweredStep = { + id: "current-step", + type: "ai", + content: "Searching for catalysts", + run_id: "run-current", + } as Message; + const currentToolStep = { + id: "current-tool", + type: "tool", + content: "results", + tool_call_id: "call-current", + run_id: "run-current", + } as Message; + const baselineIdentities = new Set(["message:previous-human"]); + const historyIdentities = new Set([ + "message:previous-human", + "message:current-human", + "message:current-step", + "tool:call-current", + ]); + + // The human already at its correct position, followed by its own steps. + expect( + restoreLocalTurnMessageOrder( + [previousHuman, currentHuman, currentAnsweredStep, currentToolStep], + baselineIdentities, + historyIdentities, + new Set(["run-current"]), + ), + ).toEqual([ + previousHuman, + currentHuman, + currentAnsweredStep, + currentToolStep, + ]); + + // An empty explicit set exercises the pending-human run_id anchor and still + // protects the persisted current-run steps. + expect( + restoreLocalTurnMessageOrder( + [previousHuman, currentHuman, currentAnsweredStep, currentToolStep], + baselineIdentities, + historyIdentities, + new Set(), + ), + ).toEqual([ + previousHuman, + currentHuman, + currentAnsweredStep, + currentToolStep, + ]); + + // Even when the current turn's human is missing its run_id, explicit + // current-turn run ids keep its steps in place below the human. + const humanWithoutRun = { + id: "current-human", + type: "human", + content: "Continue tracking the robot sector", + } as Message; + expect( + restoreLocalTurnMessageOrder( + [previousHuman, humanWithoutRun, currentAnsweredStep, currentToolStep], + baselineIdentities, + historyIdentities, + new Set(["run-current"]), + ), + ).toEqual([ + previousHuman, + humanWithoutRun, + currentAnsweredStep, + currentToolStep, + ]); +}); + +test("local turn order restores a clarification card that only canonical history confirmed", () => { + // The card can reach the merged list through the REST history page without + // ever entering the live checkpoint `messages` value, so it is not in the + // pre-submit baseline. It must still be treated as an established-past-turn + // message (not an in-flight pending step) and stay above the new human. + const previousHuman = { + id: "previous-human", + type: "human", + content: "Research today's market", + } as Message; + const previousAnswer = { + id: "previous-answer", + type: "ai", + content: "Here is the completed report", + } as Message; + const clarificationCard = { + id: "clarification-card", + type: "tool", + name: "ask_clarification", + tool_call_id: "clarification-call", + content: "Which market should I research?", + } as Message; + const currentHuman = { + id: "current-human", + type: "human", + content: "Summarize the report", + } as Message; + const currentStep = { + id: "current-step", + type: "ai", + content: "Reading the report", + } as Message; + // Baseline captured at submit time: the previous turn is there, but the + // card has not reached the live checkpoint yet, so it is absent. + const baselineIdentities = new Set([ + "message:previous-human", + "message:previous-answer", + ]); + // Canonical history has already committed the card identity. + const historyIdentities = new Set([ + "message:previous-human", + "message:previous-answer", + "tool:clarification-call", + ]); + + // The merged list has the card woven after the new human (from history). + expect( + restoreLocalTurnMessageOrder( + [ + previousHuman, + previousAnswer, + currentHuman, + clarificationCard, + currentStep, + ], + baselineIdentities, + historyIdentities, + ), + ).toEqual([ + previousHuman, + previousAnswer, + clarificationCard, + currentHuman, + currentStep, + ]); +}); + test("reconnected turn order moves same-run steps back behind the user message", () => { // Reload mid-run: replayed `messages-tuple` steps reach the merged list // before the turn's human message (the retained replay buffer may have @@ -2306,6 +2773,164 @@ test("a checkpoint message earlier than the loaded window is placed by its seq e ]); }); +test("local turn order anchors on the exact submitted human identity (X__user normalized)", () => { + // The anchor recorded at submit time names one identity; the server's + // visible copy arrives as `__user` and normalizes onto it. + const previousHuman = { + id: "previous-human", + type: "human", + content: "Research the robot sector", + } as Message; + const previousAnswer = { + id: "previous-answer", + type: "ai", + content: "Done", + } as Message; + const earlyStep = { + id: "early-step", + type: "ai", + content: "Searching for catalysts", + run_id: "run-new", + } as Message; + const serverHumanCopy = { + id: "new-human__user", + type: "human", + content: "Continue tracking", + run_id: "run-new", + } as Message; + + expect( + restoreLocalTurnMessageOrder( + [previousHuman, previousAnswer, earlyStep, serverHumanCopy], + new Set(["message:previous-human", "message:previous-answer"]), + new Set(["message:previous-human", "message:previous-answer"]), + new Set(["run-new"]), + "message:new-human", + ), + ).toEqual([previousHuman, previousAnswer, serverHumanCopy, earlyStep]); +}); + +test("local turn order keeps established history while the anchored human is absent", () => { + // R2: the checkpoint baseline covers only the latest turn while canonical + // history holds an older one. Until the submitted human reaches the render + // snapshot, no reordering may happen — a history-only human outside the + // baseline is not proof of the current turn. + const earlierAnswer = { + id: "earlier-answer", + type: "ai", + content: "Earlier answer", + } as Message; + const oldHuman = { + id: "old-human", + type: "human", + content: "An older request", + } as Message; + const recentHuman = { + id: "recent-human", + type: "human", + content: "The recent request", + } as Message; + const recentAnswer = { + id: "recent-answer", + type: "ai", + content: "The recent answer", + } as Message; + const newStep = { + id: "new-step", + type: "ai", + content: "Working on the follow-up", + run_id: "run-new", + } as Message; + const display = [earlierAnswer, oldHuman, recentHuman, recentAnswer, newStep]; + const baseline = new Set(["message:recent-human", "message:recent-answer"]); + const confirmed = new Set([ + "message:earlier-answer", + "message:old-human", + "message:recent-human", + "message:recent-answer", + ]); + + expect( + restoreLocalTurnMessageOrder( + display, + baseline, + confirmed, + new Set(["run-new"]), + "message:new-human", + ), + ).toEqual(display); +}); + +test("local turn order with a null anchor never borrows a history-only human", () => { + // Hidden human-input replies and regenerate replays submit no visible + // human, so no human identity may anchor the repair at all. + const earlierAnswer = { + id: "earlier-answer", + type: "ai", + content: "Earlier answer", + } as Message; + const oldHuman = { + id: "old-human", + type: "human", + content: "An older request", + } as Message; + const recentHuman = { + id: "recent-human", + type: "human", + content: "The recent request", + } as Message; + const replyStep = { + id: "reply-step", + type: "ai", + content: "Applying the answer", + run_id: "run-reply", + } as Message; + const display = [earlierAnswer, oldHuman, recentHuman, replyStep]; + + expect( + restoreLocalTurnMessageOrder( + display, + new Set(["message:recent-human"]), + new Set(["message:earlier-answer", "message:old-human"]), + new Set(["run-reply"]), + null, + ), + ).toEqual(display); +}); + +test("local turn order repair is idempotent across repeated deliveries", () => { + const previousHuman = { + id: "previous-human", + type: "human", + content: "Research the robot sector", + } as Message; + const earlyStep = { + id: "early-step", + type: "ai", + content: "Searching for catalysts", + run_id: "run-new", + } as Message; + const submittedHuman = { + id: "local-human-1", + type: "human", + content: "Continue tracking", + } as Message; + const displaced = [previousHuman, earlyStep, submittedHuman]; + const baseline = new Set(["message:previous-human"]); + const args = [ + baseline, + new Set(["message:previous-human"]), + new Set(["run-new"]), + "message:local-human-1", + ] as const; + + const once = restoreLocalTurnMessageOrder(displaced, ...args); + expect(once).toEqual([previousHuman, submittedHuman, earlyStep]); + // Re-merging the same repaired snapshot converges to the identical order. + expect(restoreLocalTurnMessageOrder(once, ...args)).toEqual(once); + expect(restoreLocalTurnMessageOrder(displaced, ...args)).toEqual(once); +}); + test("mergeMessages preserves canonical seq when a live copy without seq replaces the content (R3)", () => { // Content replacement must not drop trusted ordering metadata: the live // checkpoint copy refreshes the text, but the thread-global position the diff --git a/frontend/tests/unit/core/threads/send-message.test.ts b/frontend/tests/unit/core/threads/send-message.test.ts index e2dd5d1fa..eb9cc9efd 100644 --- a/frontend/tests/unit/core/threads/send-message.test.ts +++ b/frontend/tests/unit/core/threads/send-message.test.ts @@ -96,3 +96,19 @@ test("keeps human input response metadata on the hidden user message", () => { }, ]); }); + +test("uses the caller-provided human message id for the visible user message", () => { + const messages = buildThreadSubmitMessages({ + text: "hello", + humanMessageId: "local-human-1", + }); + + expect(messages).toEqual([ + { + type: "human", + id: "local-human-1", + content: [{ type: "text", text: "hello" }], + additional_kwargs: {}, + }, + ]); +});