mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-27 05:30:39 +00:00
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 <willem.jiang@gmail.com>
This commit is contained in:
parent
4ced9dddf5
commit
ee3159a01d
@ -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<string, number>();
|
||||
const earliestVisibleSeqByRunIdentity = new Map<string, number>();
|
||||
for (const row of sortedRows) {
|
||||
const identity = messageIdentity(row.content);
|
||||
if (!identity || !isValidMessageSeq(row.seq)) {
|
||||
continue;
|
||||
}
|
||||
const key = `${row.run_id}:${identity}`;
|
||||
const known = earliestSeqByRunIdentity.get(key);
|
||||
if (known === undefined || row.seq < known) {
|
||||
earliestSeqByRunIdentity.set(key, row.seq);
|
||||
}
|
||||
if (!isHiddenFromUIMessage(row.content)) {
|
||||
const knownVisible = earliestVisibleSeqByRunIdentity.get(key);
|
||||
if (knownVisible === undefined || row.seq < knownVisible) {
|
||||
earliestVisibleSeqByRunIdentity.set(key, row.seq);
|
||||
}
|
||||
}
|
||||
}
|
||||
const anchored = reconciled.map((row) => {
|
||||
const identity = messageIdentity(row.content);
|
||||
if (!identity) {
|
||||
return row;
|
||||
}
|
||||
const key = `${row.run_id}:${identity}`;
|
||||
const earliestSeq =
|
||||
earliestVisibleSeqByRunIdentity.get(key) ??
|
||||
earliestSeqByRunIdentity.get(key);
|
||||
if (earliestSeq === undefined || row.seq === earliestSeq) {
|
||||
return row;
|
||||
}
|
||||
return { ...row, seq: earliestSeq };
|
||||
});
|
||||
// Re-anchoring can change row order. Sort before comparing with the retained
|
||||
// snapshot so unchanged reconciliations can still reuse the previous array.
|
||||
anchored.sort((left, right) => left.seq - right.seq);
|
||||
if (
|
||||
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
|
||||
|
||||
@ -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,
|
||||
);
|
||||
});
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user