From ee3159a01d1518c8031f904e8668b6a7da5b6839 Mon Sep 17 00:00:00 2001 From: kuse <3133746534@qq.com> Date: Thu, 1 Oct 2026 08:20:42 +0800 Subject: [PATCH] fix(threads): re-anchor re-persisted history rows to their earliest seq (#6128) * fix(threads): re-anchor re-persisted history rows to their earliest seq reconcileThreadHistoryRows collapsed a repeated (run_id, identity) to its latest-seq copy, deleting the earliest-seq row before buildVisibleHistoryMessages could apply its earliest-seq-wins anchoring. A message re-persisted under a newer seq therefore jumped to the tail of the thread, contradicting the documented contract in this function's docstring, in buildVisibleHistoryMessages, and in the backend get_message_seqs earliest-seq-wins rule. Re-anchor the surviving row to the earliest seq its identity held in the merged window after deduping, and make the retained-history stability check value-based so re-anchored rows do not force a state update on every pass. Fixes #6125 * fix(threads): address history reconciliation review --------- Co-authored-by: liwenjie200543 <175601277+liwenjie200543@users.noreply.github.com> Co-authored-by: Willem Jiang --- frontend/src/core/threads/hooks.ts | 68 ++++++++++- .../unit/core/threads/message-order.test.ts | 115 ++++++++++++++++++ 2 files changed, 178 insertions(+), 5 deletions(-) diff --git a/frontend/src/core/threads/hooks.ts b/frontend/src/core/threads/hooks.ts index 40cf93679e..0d5f7e6a73 100644 --- a/frontend/src/core/threads/hooks.ts +++ b/frontend/src/core/threads/hooks.ts @@ -663,16 +663,74 @@ export function reconcileThreadHistoryRows( rowsBySeq.set(row.seq, row); } - const reconciled = dedupeRunMessagesByIdentity( - [...rowsBySeq.values()].sort((left, right) => left.seq - right.seq), + const sortedRows = [...rowsBySeq.values()].sort( + (left, right) => left.seq - right.seq, ); + const reconciled = dedupeRunMessagesByIdentity(sortedRows); + // Deduping collapses a re-persisted identity to its newest copy, but per + // the backend `get_message_seqs` earliest-seq-wins rule the message keeps + // the position it first occupied (see buildVisibleHistoryMessages). + // Re-anchor the surviving row to the earliest seq the identity held in + // this window, otherwise a re-persisted update pushes the message towards + // the tail. Prefer visible rows so a hidden control copy cannot move its + // visible twin; hidden rows provide a fallback only when no visible copy exists. + const earliestSeqByRunIdentity = new Map(); + const earliestVisibleSeqByRunIdentity = new Map(); + for (const row of sortedRows) { + const identity = messageIdentity(row.content); + if (!identity || !isValidMessageSeq(row.seq)) { + continue; + } + const key = `${row.run_id}:${identity}`; + const known = earliestSeqByRunIdentity.get(key); + if (known === undefined || row.seq < known) { + earliestSeqByRunIdentity.set(key, row.seq); + } + if (!isHiddenFromUIMessage(row.content)) { + const knownVisible = earliestVisibleSeqByRunIdentity.get(key); + if (knownVisible === undefined || row.seq < knownVisible) { + earliestVisibleSeqByRunIdentity.set(key, row.seq); + } + } + } + const anchored = reconciled.map((row) => { + const identity = messageIdentity(row.content); + if (!identity) { + return row; + } + const key = `${row.run_id}:${identity}`; + const earliestSeq = + earliestVisibleSeqByRunIdentity.get(key) ?? + earliestSeqByRunIdentity.get(key); + if (earliestSeq === undefined || row.seq === earliestSeq) { + return row; + } + return { ...row, seq: earliestSeq }; + }); + // Re-anchoring can change row order. Sort before comparing with the retained + // snapshot so unchanged reconciliations can still reuse the previous array. + anchored.sort((left, right) => left.seq - right.seq); if ( - reconciled.length === previousRows.length && - reconciled.every((row, index) => row === previousRows[index]) + anchored.length === previousRows.length && + anchored.every((row, index) => { + const previous = previousRows[index]; + if (!previous) { + return false; + } + // Re-anchoring rebuilds the row object on every pass; treat a row whose + // run_id, seq and content reference all match as unchanged so the + // retained-history state can stay referentially stable. + return ( + row === previous || + (row.run_id === previous.run_id && + row.seq === previous.seq && + row.content === previous.content) + ); + }) ) { return previousRows; } - return reconciled; + return anchored; } // mergeMessages now lives in ./message-order (pure, unit-testable ordering diff --git a/frontend/tests/unit/core/threads/message-order.test.ts b/frontend/tests/unit/core/threads/message-order.test.ts index a6502b2fd5..bac54da2ce 100644 --- a/frontend/tests/unit/core/threads/message-order.test.ts +++ b/frontend/tests/unit/core/threads/message-order.test.ts @@ -6,6 +6,7 @@ import { buildVisibleHistoryMessages, mergeRenderedMessageLedger, mergeMessages, + reconcileThreadHistoryRows, resolveThreadTransientHistoryBridge, resolveTransientHistoryBridge, } from "@/core/threads/hooks"; @@ -543,3 +544,117 @@ test.each(["before", "after"] as const)( expect(idsOf(mergeMessages(resolved, [], []))).toEqual(idsOf(previous)); }, ); + +test("reconcile anchors a re-persisted message at its earliest seq", () => { + // Same run, same message id, re-persisted under a later journal seq while + // the older row only survives in the retained snapshot. + const previousRows = [row("r1", 95, msg("m1", "ai", "partial"))]; + const currentRows = [row("r1", 130, msg("m1", "ai", "final"))]; + + const reconciled = reconcileThreadHistoryRows( + previousRows, + currentRows, + false, + ); + + expect(reconciled).toHaveLength(1); + // earliest-seq-wins: the re-persisted update must not push the message + // towards the tail of the thread. + expect(reconciled[0]!.seq).toBe(95); + // content converges to the newest copy. + expect(reconciled[0]!.content).toBe(currentRows[0]!.content); + + const visible = buildVisibleHistoryMessages(reconciled, new Set()); + expect(visible).toHaveLength(1); + expect(visible[0]!.additional_kwargs?.[MESSAGE_SEQ_KEY]).toBe(95); + expect(visible[0]!.content).toBe("final"); +}); + +test("reconcile still converges content for same-seq refreshed copies", () => { + const previousRows = [row("r1", 95, msg("m1", "ai", "partial"))]; + const currentRows = [row("r1", 95, msg("m1", "ai", "final"))]; + + const reconciled = reconcileThreadHistoryRows( + previousRows, + currentRows, + false, + ); + + expect(reconciled).toHaveLength(1); + expect(reconciled[0]!.seq).toBe(95); + expect(reconciled[0]!.content).toBe(currentRows[0]!.content); +}); + +test.each([false, true])( + "reconcile prefers the earliest visible row over a hidden control copy (complete=%s)", + (isAuthoritativeComplete) => { + const hidden = row( + "r1", + 90, + withKwargs(msg("req-1", "system", "reminder"), { + hide_from_ui: true, + }), + ); + const middle = row("r1", 95, msg("m1", "ai", "earlier answer")); + const visible = row("r1", 100, msg("req-1__user", "human", "question")); + const refreshed = row( + "r1", + 130, + msg("req-1__user", "human", "updated question"), + ); + const previousRows = [hidden, middle, visible]; + + const reconciled = reconcileThreadHistoryRows( + previousRows, + [...previousRows, refreshed], + isAuthoritativeComplete, + ); + + const human = reconciled.find((entry) => entry.content.type === "human")!; + expect(human.seq).toBe(100); + expect(human.content).toBe(refreshed.content); + const messages = buildVisibleHistoryMessages(reconciled, new Set()); + expect(idsOf(messages)).toEqual(["m1", "req-1__user"]); + expect(seqsOf(messages)).toEqual([95, 100]); + }, +); + +test("reconcile falls back to the earliest hidden row when no visible copy exists", () => { + const hidden = (content: string) => + withKwargs(msg("control", "system", content), { hide_from_ui: true }); + const previous = row("r1", 90, hidden("original reminder")); + const refreshed = row("r1", 130, hidden("updated reminder")); + + const reconciled = reconcileThreadHistoryRows([previous], [refreshed], false); + + expect(reconciled).toHaveLength(1); + expect(reconciled[0]!.seq).toBe(90); + expect(reconciled[0]!.content).toBe(refreshed.content); + expect(reconciled[0]!.content.additional_kwargs?.hide_from_ui).toBe(true); +}); + +test("reconcile sorts re-anchored rows and preserves a stable retained snapshot", () => { + const original = row("r1", 95, msg("a1", "ai", "partial")); + const middle = row("r1", 100, msg("m1", "ai", "middle")); + const refreshed = row("r1", 130, msg("a1", "ai", "final")); + const currentRows = [middle, refreshed]; + + const reconciled = reconcileThreadHistoryRows( + [original, middle], + currentRows, + false, + ); + + expect(reconciled.map((entry) => entry.seq)).toEqual([95, 100]); + expect(reconciled[0]!.content).toBe(refreshed.content); + expect(reconciled[1]).toBe(middle); + expect( + buildVisibleHistoryMessages(reconciled, new Set()).map( + (entry) => entry.content, + ), + ).toEqual(["final", "middle"]); + // Compare the sorted rows before taking the stable-output shortcut. + expect(reconcileThreadHistoryRows(reconciled, currentRows, false)).toBe( + reconciled, + ); +});