github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-09-09 13:39:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(frontend): preserve trusted message positions through content merge (#5293)

Browse Source 
* test(threads): add red R3/R4 merge ordering regressions

* fix(threads): preserve trusted seq positions through content merge (R3/R4)

Extract the message ordering/identity logic into a pure
core/threads/message-order.ts module. Each normalized identity now tracks
latest visible content and trusted position separately: content replacement
no longer drops deerflow_seq/run_id/turn_duration (R3), and a seq-carrying
live message is placed by the ascending seq skeleton instead of the next
shared identity anchor (R4: 1,3,2,5 -> 1,2,3,5).

buildVisibleHistoryMessages converges repeated identities to the earliest
visible feed row (mirroring backend get_message_seqs), and the
summarization transient bridge plus rendered ledger share the same
position priority: trusted seq outranks anchor weaving, bridge refreshes
keep known seqs, and hidden control copies never contribute a visible
position.

* test(e2e): add long-thread ordering regression with compaction and pagination

Add tests/e2e/thread-ordering.spec.ts: a deterministic 68-row, 33-turn
fixture with two hidden compaction summaries, a paginated /messages/page
mock, and a live compaction during submit (real SSE frame shapes). Asserts
DOM group order at stage barriers, outline/scroll navigation across the
virtualized list, tool-card association, and order stability across reload.

Also close three mock gaps in mockLangGraphAPI (token-usage, mcp-tasks,
workspace-changes): unmocked they fell through to the absent gateway and
the 401 redirected thread pages to /login, breaking every thread-page spec
in a gateway-less Playwright environment.

* test(threads): address review on ordering regression coverage

- e2e: actually expand the collapsed web_search step and assert the
  intermediate result payload (realistic JSON array fixture); assert the
  new turn's DOM relative order via compareDocumentPosition instead of
  racing viewport coordinates; add a Custom Agent route regression sharing
  the same paginated fixture.
- Add a unit test for the hidden-control-only seq fallback path.
- Keep isNonEmptyString in hooks.ts (message-order.ts does not use it).
- Document the seq-first position authority contract in frontend/src/AGENTS.md.

* test(threads): type run_id fixtures via getMessageRunId accessor

* test(e2e): exercise the real collapsed-steps region for the tool payload check

The previous toolStep.click() was a no-op: as the last tool call, the
web_search step rendered unconditionally. Add a second tool call to the
turn-30 fixture so web_search falls into the collapsed moreSteps region,
assert the intermediate result payload is hidden while collapsed, then
click the "1 more step" button and assert it becomes visible.

* docs(frontend): prettier-format AGENTS.md merge contract

* fix(frontend): anchor mixed-sequence message segments

* test(auth): include project permissions in me contracts

* fix(frontend): anchor trailing steps to positioned live results

* fix(frontend): preserve prefixes before rescued sequence anchors
This commit is contained in:
Zeren Wang 2026-09-09 05:32:06 +02:00 committed by GitHub
parent a3848ef155
commit 05432f4b43
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
9 changed files with 1732 additions and 297 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -1222,6 +1222,8 @@ The Web UI shows the active goal above the composer. The same command is availab
### Manual Context Compaction
The Web UI preserves persisted message order when merging history with live updates. Streaming steps around a persisted result inside the loaded history stay together, including steps that arrive after the result. Steps captured during compaction also remain visible before their persisted result when history has not refreshed and the UI has not rendered them yet.
Compaction keeps the current user request and summarizes older assistant/tool activity. When rescuing that request leaves an assistant/tool-only summary window, input trimming favors its most recent content. For mixed histories whose user-message anchor falls outside the trimming budget, compaction retains the existing final-message fallback. `summarization.trim_tokens_to_summarize` (4000 by default) controls trimming of the raw summary input; escaping and prompt formatting add overhead beyond that budget. Setting this option to `null` disables input trimming for the summary model; choose that only when the model can accept the full history being compacted.
Use `/compact` in the Web UI composer to summarize older context for the current thread. DeerFlow keeps the full chat visible, but future model calls use the compacted summary plus recent messages. The command is ignored when there is not enough history to compact, and it is blocked while the thread has a run in flight, including when that run is owned by another Gateway worker. If a multi-worker reservation loses its lease, DeerFlow cancels the checkpoint writer before the replacing run proceeds and returns a retryable conflict after cleanup. Thread-title edits are serialized through the same state-write boundary and show a conflict without closing the rename dialog when a run is active.

2
frontend/AGENTS.md
View File

@ -36,7 +36,7 @@ Webpack is the default development bundler. Use `DEER_FLOW_DEV_BUNDLER=turbo` wi
Rstest runs them as two projects (`rstest.config.ts`). `*.test.ts` / `*.test.tsx` run in a plain **node** environment — that is nearly the whole suite, and it is the default for anything that is pure logic. `*.dom.test.ts` / `*.dom.test.tsx` run in **happy-dom**, for tests that need a document: hooks driven through `renderHook` from `@testing-library/react`, and components. Keep the split — a DOM environment costs roughly 3x the runtime of the node suite, so tests that do not render should not opt into it. A hook whose behavior only exists under real React (effect ordering, cleanup on unmount, re-render on store change) belongs in a `.dom.test.*` file rather than a node test that mocks `react` itself.
E2E tests live under `tests/e2e/` and use Playwright with Chromium. They mock all backend APIs via `page.route()` network interception and test real page interactions (navigation, chat input, streaming responses). Config: `playwright.config.ts`.
E2E tests live under `tests/e2e/` and use Playwright with Chromium. They mock all backend APIs via `page.route()` network interception and test real page interactions (navigation, chat input, streaming responses). Config: `playwright.config.ts`. The real-backend auth contract in `tests/e2e-real-backend/auth-disabled-contract.spec.ts` and `backend/tests/test_auth_me_permissions.py` pin the complete route-permission list; update both when adding registered permissions (including `projects:read/write/delete`).
## Architecture

2
frontend/src/AGENTS.md
View File

@ -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. 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 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.
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`

416
frontend/src/core/threads/hooks.ts
View File

@ -43,6 +43,15 @@ import {
searchThreadsByArchive,
type ThreadMetadataPatch,
} from "./api";
import {
dedupeMessagesByIdentity,
insertByTrustedSeq,
isValidMessageSeq,
MESSAGE_SEQ_KEY,
mergeMessages,
messageIdentity,
trustedMessageSeq,
} from "./message-order";
import {
hasRenderedThreadStateUpdate,
reduceThreadStateUpdates,
@ -181,11 +190,10 @@ export function buildThreadSubmitMessages({
const EMPTY_MESSAGES: Message[] = [];
const EMPTY_RUN_MESSAGES: RunMessage[] = [];
const EMPTY_MESSAGE_IDENTITIES: readonly string[] = [];
const INJECTED_USER_MESSAGE_ID_SUFFIX = "__user";
// Thread-global feed position, attached by the backend to history rows and to
// `values` frame messages it has already persisted. Mirrors MESSAGE_SEQ_KEY in
// `deerflow/runtime/events/message_identity.py`.
const MESSAGE_SEQ_KEY = "deerflow_seq";
function isNonEmptyString(value: string | undefined): value is string {
return typeof value === "string" && value.length > 0;
}
const EMPTY_THREAD_VALUES: AgentThreadState = {
title: "",
@ -194,102 +202,11 @@ const EMPTY_THREAD_VALUES: AgentThreadState = {
todos: [],
};
function isNonEmptyString(value: string | undefined): value is string {
return typeof value === "string" && value.length > 0;
}
const SUMMARIZATION_MIDDLEWARE_UPDATE_KEYS = new Set([
"SummarizationMiddleware.before_model",
"DeerFlowSummarizationMiddleware.before_model",
]);
/** Thread-global feed position, when the backend has attached one. */
function messageSeq(message: Message): number | undefined {
const seq = message.additional_kwargs?.[MESSAGE_SEQ_KEY];
return typeof seq === "number" ? seq : undefined;
}
function messageIdentity(message: Message): string | undefined {
if (
"tool_call_id" in message &&
typeof message.tool_call_id === "string" &&
message.tool_call_id.length > 0
) {
return `tool:${message.tool_call_id}`;
}
if (typeof message.id === "string" && message.id.length > 0) {
// DynamicContextMiddleware replaces the submitted HumanMessage(id=X) with
// a hidden SystemMessage(id=X) and the real HumanMessage(id=X__user).
// Treat those human copies as one UI message so a committed render ledger
// cannot retain X beside the later checkpoint copy X__user.
const messageId =
message.type === "human" &&
message.id.endsWith(INJECTED_USER_MESSAGE_ID_SUFFIX)
? message.id.slice(0, -INJECTED_USER_MESSAGE_ID_SUFFIX.length) ||
message.id
: message.id;
return `message:${messageId}`;
}
return undefined;
}
function dedupeMessagesByIdentity(messages: Message[]): Message[] {
const lastIndexByIdentity = new Map<string, number>();
const lastVisibleIndexByIdentity = new Map<string, number>();
// This is a UI-display dedupe rule, not a general LangChain message-stream
// contract. Hidden messages that share an identity with a visible message are
// treated as control messages for this merged view; hidden messages carrying
// independent tracing/task semantics should use a distinct id or a custom
// stream/state channel instead of relying on message dedupe preservation.
const preservedTurnDurations = new Map<string, number>();
messages.forEach((message, index) => {
const identity = messageIdentity(message);
if (identity) {
lastIndexByIdentity.set(identity, index);
if (!isHiddenFromUIMessage(message)) {
lastVisibleIndexByIdentity.set(identity, index);
}
if (message.additional_kwargs?.turn_duration !== undefined) {
preservedTurnDurations.set(
identity,
message.additional_kwargs.turn_duration as number,
);
}
}
});
return messages
.filter((message, index) => {
const identity = messageIdentity(message);
if (!identity) {
return true;
}
const visibleIndex = lastVisibleIndexByIdentity.get(identity);
if (visibleIndex !== undefined) {
return visibleIndex === index;
}
return lastIndexByIdentity.get(identity) === index;
})
.map((message) => {
const identity = messageIdentity(message);
if (
identity &&
preservedTurnDurations.has(identity) &&
message.additional_kwargs?.turn_duration === undefined
) {
return {
...message,
additional_kwargs: {
...message.additional_kwargs,
turn_duration: preservedTurnDurations.get(identity),
},
} as Message;
}
return message;
});
}
function dedupeRunMessagesByIdentity(messages: RunMessage[]): RunMessage[] {
const lastIndexByIdentity = new Map<string, number>();
messages.forEach((message, index) => {
@ -326,7 +243,32 @@ export function buildVisibleHistoryMessages(
const visibleRows = messageRows.filter(
(message) => !supersededRunIds.has(message.run_id),
);
return dedupeMessagesByIdentity([
// Content and position converge separately for a repeated identity: the
// newest visible row supplies the content, but the position stays the
// earliest trusted feed row — mirroring the backend `get_message_seqs`
// earliest-seq-wins rule, so a re-persisted update cannot push the message
// towards the tail. Hidden control copies never contribute a visible
// position (they only serve as a fallback when no visible row carries the
// identity).
const earliestSeqByIdentity = new Map<string, number>();
const earliestVisibleSeqByIdentity = new Map<string, number>();
for (const row of visibleRows) {
const identity = messageIdentity(row.content);
if (!identity || !isValidMessageSeq(row.seq)) {
continue;
}
const known = earliestSeqByIdentity.get(identity);
if (known === undefined || row.seq < known) {
earliestSeqByIdentity.set(identity, row.seq);
}
if (!isHiddenFromUIMessage(row.content)) {
const knownVisible = earliestVisibleSeqByIdentity.get(identity);
if (knownVisible === undefined || row.seq < knownVisible) {
earliestVisibleSeqByIdentity.set(identity, row.seq);
}
}
}
const deduped = dedupeMessagesByIdentity([
// Carry the owning run_id onto the content message so historical subtask
// cards can fetch their persisted step history on expand (#3779). run_id
// lives on the RunMessage wrapper and would otherwise be dropped here.
@ -342,6 +284,28 @@ export function buildVisibleHistoryMessages(
},
})),
]);
return deduped.map((message) => {
const identity = messageIdentity(message);
if (!identity) {
return message;
}
const earliestSeq =
earliestVisibleSeqByIdentity.get(identity) ??
earliestSeqByIdentity.get(identity);
if (
earliestSeq === undefined ||
message.additional_kwargs?.[MESSAGE_SEQ_KEY] === earliestSeq
) {
return message;
}
return {
...message,
additional_kwargs: {
...message.additional_kwargs,
[MESSAGE_SEQ_KEY]: earliestSeq,
},
} as Message;
});
}
export type ThreadMessagesPageResponse = {
@ -350,10 +314,6 @@ export type ThreadMessagesPageResponse = {
next_before_seq: number | null;
};
function isValidThreadMessageSeq(value: unknown): value is number {
return typeof value === "number" && Number.isSafeInteger(value) && value >= 1;
}
/**
* Validate the sequence fields that history reconciliation and pagination use
* as runtime identities. The static RunMessage type cannot protect this JSON
@ -379,7 +339,7 @@ export function parseThreadMessagesPageResponse(
typeof row === "object" && row !== null
? Reflect.get(row, "seq")
: undefined;
if (!isValidThreadMessageSeq(seq)) {
if (!isValidMessageSeq(seq)) {
throw new Error("Thread history returned a row with an invalid seq.");
}
if (seenSeqs.has(seq)) {
@ -389,7 +349,7 @@ export function parseThreadMessagesPageResponse(
}
if (
(hasMore && !isValidThreadMessageSeq(nextBeforeSeq)) ||
(hasMore && !isValidMessageSeq(nextBeforeSeq)) ||
(!hasMore && nextBeforeSeq !== null)
) {
throw new Error(
@ -464,7 +424,7 @@ export function reconcileThreadHistoryRows(
const sourceRows = isAuthoritativeComplete
? currentRows
: [...previousRows, ...currentRows];
if (sourceRows.some((row) => !isValidThreadMessageSeq(row.seq))) {
if (sourceRows.some((row) => !isValidMessageSeq(row.seq))) {
console.error(
"Thread history reconciliation received an invalid sequence value.",
);
@ -491,180 +451,10 @@ export function reconcileThreadHistoryRows(
return reconciled;
}
export function mergeMessages(
historyMessages: Message[],
threadMessages: Message[],
optimisticMessages: Message[],
): Message[] {
const savedTurnDurations = new Map<string, number>();
const savedRunIds = new Map<string, string>();
for (const msg of historyMessages) {
const identity = messageIdentity(msg);
const runId = getMessageRunId(msg);
if (identity && runId) {
savedRunIds.set(identity, runId);
}
if (identity && msg.additional_kwargs?.turn_duration !== undefined) {
savedTurnDurations.set(
identity,
msg.additional_kwargs.turn_duration as number,
);
}
}
const canonical = dedupeMessagesByIdentity(historyMessages);
const live = dedupeMessagesByIdentity(threadMessages);
const canonicalByIdentity = new Map(
canonical.flatMap((message) => {
const identity = messageIdentity(message);
return identity ? [[identity, message] as const] : [];
}),
);
const replacementByIdentity = new Map<string, Message>();
// This uses the same identity-anchor weaving shape as
// resolveTransientHistoryBridge, but intentionally remains separate: live
// messages may replace canonical copies and identity-less entries survive.
const beforeAnchor = new Map<string, Message[]>();
let pending: Message[] = [];
let lastAnchorIdentity: string | undefined;
// Lower bound of the history page window that is currently loaded. A live
// message whose seq is below it belongs before everything on screen, which
// is knowledge the anchor weaving below cannot reach: the anchor only says
// "earlier than this row", and after compaction the nearest anchor can sit
// deep inside the window (#4666 — measured at row 25 of 50).
const canonicalMinSeq = canonical.reduce<number | undefined>(
(min, message) => {
const seq = messageSeq(message);
return seq !== undefined && (min === undefined || seq < min) ? seq : min;
},
undefined,
);
const beforeWindow: Message[] = [];
// Split off what the feed places before the loaded window BEFORE the anchor
// walk rather than inside it. A summarized checkpoint can share no identity
// at all with the loaded page — compaction keeps only this run's recent tail,
// while the page on screen was fetched turns earlier — and the anchor loop
// then never runs. That is exactly when a rescued early turn most needs its
// seq: user submits, waits without reloading, compaction fires, and the
// message is appended to the tail instead (#4666).
const liveInWindow: Message[] = [];
for (const message of live) {
const seq = messageSeq(message);
if (
seq !== undefined &&
canonicalMinSeq !== undefined &&
seq < canonicalMinSeq
) {
beforeWindow.push(message);
} else {
liveInWindow.push(message);
}
}
// A summarized checkpoint is not necessarily a contiguous history suffix:
// middleware may retain protected prompt/input messages at the front and a
// recent tail at the back. Treat every shared identity as an ordering anchor,
// replacing the canonical copy in place. New live messages are woven before
// the next shared anchor (or after the last one), so a protected early input
// can never be moved to the tail by global last-copy deduplication.
for (const message of liveInWindow) {
const identity = messageIdentity(message);
const canonicalMessage = identity
? canonicalByIdentity.get(identity)
: undefined;
if (!identity || !canonicalMessage) {
pending.push(message);
continue;
}
// A summarized checkpoint may start with a protected message whose true
// canonical position is separated from this anchor by unloaded pages —
// rescued dynamic-context messages are the common case. Its position
// relative to this anchor is still known (both the checkpoint and
// seq-sorted history place it earlier), so it is woven in before the
// anchor like any other live-only segment. Dropping it instead was how a
// user's own question disappeared from a long thread once the first
// history page no longer reached back to it (#4666): a collapsed unloaded
// gap is recoverable by paging, a discarded message is not.
if (pending.length > 0) {
beforeAnchor.set(identity, [
...(beforeAnchor.get(identity) ?? []),
...pending,
]);
}
pending = [];
lastAnchorIdentity = identity;
// A hidden checkpoint control message must not replace a visible canonical
// user turn that happens to reuse its identity. In every other case the
// live checkpoint copy is fresher and replaces history without moving it.
if (
!isHiddenFromUIMessage(message) ||
isHiddenFromUIMessage(canonicalMessage)
) {
replacementByIdentity.set(identity, message);
}
}
let canonicalAndLive: Message[];
if (!lastAnchorIdentity) {
canonicalAndLive = [...canonical, ...liveInWindow];
} else {
canonicalAndLive = [];
for (const message of canonical) {
const identity = messageIdentity(message);
if (identity) {
canonicalAndLive.push(...(beforeAnchor.get(identity) ?? []));
}
const replacement = identity
? replacementByIdentity.get(identity)
: undefined;
canonicalAndLive.push(replacement ?? message);
}
// A trailing live-only segment is known to come after the last shared
// anchor, but that anchor may not be the end of canonical history (for
// example, another client may have persisted newer rows). Preserve the
// canonical source order before appending the live tail.
canonicalAndLive.push(...pending);
}
const merged = dedupeMessagesByIdentity([
...[...beforeWindow].sort(
(left, right) => (messageSeq(left) ?? 0) - (messageSeq(right) ?? 0),
),
...canonicalAndLive,
...optimisticMessages,
]);
return merged.map((message) => {
const identity = messageIdentity(message);
if (!identity) {
return message;
}
const shouldRestoreRunId =
savedRunIds.has(identity) && !getMessageRunId(message);
const shouldRestoreTurnDuration =
savedTurnDurations.has(identity) &&
message.additional_kwargs?.turn_duration === undefined;
if (shouldRestoreRunId || shouldRestoreTurnDuration) {
return {
...message,
...(shouldRestoreRunId ? { run_id: savedRunIds.get(identity) } : {}),
...(shouldRestoreTurnDuration
? {
additional_kwargs: {
...message.additional_kwargs,
turn_duration: savedTurnDurations.get(identity),
},
}
: {}),
} as Message;
}
return message;
});
}
// mergeMessages now lives in ./message-order (pure, unit-testable ordering
// module); it is imported above and re-exported here so existing consumers of
// this module keep working unchanged.
export { mergeMessages };
/**
* Keep messages from a locally submitted turn behind that turn's user input.
@ -973,8 +763,28 @@ export function resolveTransientHistoryBridge(
return visibleHistory;
}
// Trusted seq outranks identity anchors — the same position priority
// mergeMessages applies. A rescued message whose seq is known lands exactly
// where the feed places it, even when no bridge identity overlaps the
// loaded window; anchor weaving below remains the fallback for entries
// without a trustworthy position.
const seqPositioned: Message[] = [];
const unpositioned: Message[] = [];
for (const message of missing) {
if (trustedMessageSeq(message) !== undefined) {
seqPositioned.push(message);
} else {
unpositioned.push(message);
}
}
// Place rescued seq rows first so weaving can anchor unsequenced
// neighbors to them without a later insertion reversing their order.
const positionedHistory = insertByTrustedSeq(visibleHistory, seqPositioned);
const anchorIdentities = new Set(
positionedHistory.map(messageIdentity).filter(isNonEmptyString),
);
const missingByIdentity = new Map(
missing.flatMap((message) => {
unpositioned.flatMap((message) => {
const identity = messageIdentity(message);
return identity ? [[identity, message] as const] : [];
}),
@ -989,12 +799,17 @@ export function resolveTransientHistoryBridge(
);
let pending: Message[] = [];
let lastAnchorIdentity: string | undefined;
let hasCanonicalAnchor = false;
for (const identity of bridgeOrder) {
if (presentIdentities.has(identity)) {
if (anchorIdentities.has(identity)) {
if (pending.length > 0) {
if (hasCanonicalAnchor) {
// A rescued seq anchor preserves its captured prefix even if React
// has not rendered it yet. Only a leading prefix anchored to loaded
// history needs proof that it does not span an unloaded cursor gap.
if (
lastAnchorIdentity !== undefined ||
!presentIdentities.has(identity)
) {
beforeAnchor.set(identity, [
...(beforeAnchor.get(identity) ?? []),
...pending,
@ -1027,12 +842,7 @@ export function resolveTransientHistoryBridge(
}
}
}
// The prefix before the first loaded anchor has no trustworthy position:
// cursor pages containing its intervening history may not be loaded yet.
// The sole exception is a prefix whose exact relative position was
// already committed to the previous UI frame.
pending = [];
hasCanonicalAnchor = true;
lastAnchorIdentity = identity;
continue;
}
@ -1043,17 +853,17 @@ export function resolveTransientHistoryBridge(
}
}
// No bridge identity overlaps canonical history. This is the original
// No bridge identity overlaps a positioned row. This is the original
// persistence-gap case: loaded history is older and the rescued live turns
// belong after it.
if (!lastAnchorIdentity) {
return [...visibleHistory, ...missing];
return [...positionedHistory, ...unpositioned];
}
// A candidate added before its ordering snapshot (or carrying an identity
// absent from that snapshot) cannot be anchored. Keep it in capture order at
// the trailing edge of the anchored bridge rather than dropping it.
for (const message of missing) {
for (const message of unpositioned) {
const identity = messageIdentity(message);
if (identity && !emittedMissingIdentities.has(identity)) {
pending.push(message);
@ -1062,7 +872,7 @@ export function resolveTransientHistoryBridge(
}
const resolved: Message[] = [];
for (const message of visibleHistory) {
for (const message of positionedHistory) {
const identity = messageIdentity(message);
if (identity) {
resolved.push(...(beforeAnchor.get(identity) ?? []));
@ -1105,9 +915,27 @@ export function mergeTransientHistoryBridge(
(!isHiddenFromUIMessage(captured) || isHiddenFromUIMessage(existing))
) {
// Refresh the buffered snapshot without moving its first-known
// chronological position. Repeated compression can recapture protected
// prefix messages before a newer tail.
merged[existingIndex] = captured;
// chronological position — and without dropping a trusted position the
// earlier copy already carried. Repeated compression can recapture
// protected prefix messages before a newer tail; a refresh without seq
// must not reset the position the bridge resolver relies on.
const existingSeq = trustedMessageSeq(existing);
const capturedSeq = trustedMessageSeq(captured);
const keptSeq =
existingSeq !== undefined &&
(capturedSeq === undefined || existingSeq < capturedSeq)
? existingSeq
: capturedSeq;
merged[existingIndex] =
keptSeq !== undefined && capturedSeq !== keptSeq
? ({
...captured,
additional_kwargs: {
...captured.additional_kwargs,
[MESSAGE_SEQ_KEY]: keptSeq,
},
} as Message)
: captured;
}
}
return merged;

435
frontend/src/core/threads/message-order.ts Normal file
View File

@ -0,0 +1,435 @@
/**
* Trusted ordering for the merged chat message view.
*
* A normalized identity carries two independent facts, tracked separately:
*
* - latest visible content: live copies refresh history copies, but a hidden
* checkpoint control message never overwrites a visible user turn;
* - trusted position: a valid `deerflow_seq` stamped by the current thread's
* REST feed or by server-stamped state/live frames. Only positive safe
* integers qualify; a missing or invalid value never overwrites a known
* one, and several trusted values for one identity converge to the
* earliest feed position (mirroring the backend `get_message_seqs`
* earliest-seq-wins rule).
*
* Ordering uses a skeleton of every identity with a trusted seq, sorted by
* position, and weaves segments without seq around shared-identity anchors
* exactly as before: internal segment order is preserved, a segment goes
* before its next anchor. Live-only positioned entries within the loaded
* window also anchor trailing segments; rescued prefixes before that window
* leave new steps at the tail. Conflicting speculative constraints lose to
* the skeleton. `deerflow_seq` is server-owned display metadata — it is
* never written back into a checkpoint by the client.
*
* This module is pure: no React, no caches, no per-thread state. Callers
* scope inputs to the current thread, so switching threads, branching, or
* replaying cannot inherit positions.
*/
import type { Message } from "@langchain/langgraph-sdk";
import { getMessageRunId } from "../messages/run-duration";
import { isHiddenFromUIMessage } from "../messages/utils";
// Thread-global feed position, attached by the backend to history rows and to
// `values` frame messages it has already persisted. Mirrors MESSAGE_SEQ_KEY in
// `deerflow/runtime/events/message_identity.py`.
export const MESSAGE_SEQ_KEY = "deerflow_seq";
const INJECTED_USER_MESSAGE_ID_SUFFIX = "__user";
export function messageIdentity(message: Message): string | undefined {
if (
"tool_call_id" in message &&
typeof message.tool_call_id === "string" &&
message.tool_call_id.length > 0
) {
return `tool:${message.tool_call_id}`;
}
if (typeof message.id === "string" && message.id.length > 0) {
// DynamicContextMiddleware replaces the submitted HumanMessage(id=X) with
// a hidden SystemMessage(id=X) and the real HumanMessage(id=X__user).
// Treat those human copies as one UI message so a committed render ledger
// cannot retain X beside the later checkpoint copy X__user.
const messageId =
message.type === "human" &&
message.id.endsWith(INJECTED_USER_MESSAGE_ID_SUFFIX)
? message.id.slice(0, -INJECTED_USER_MESSAGE_ID_SUFFIX.length) ||
message.id
: message.id;
return `message:${messageId}`;
}
return undefined;
}
/** A seq is trustworthy only as a positive safe integer. */
export function isValidMessageSeq(value: unknown): value is number {
return typeof value === "number" && Number.isSafeInteger(value) && value >= 1;
}
/** Thread-global feed position, when the message carries a trustworthy one. */
export function trustedMessageSeq(message: Message): number | undefined {
const seq = message.additional_kwargs?.[MESSAGE_SEQ_KEY];
return isValidMessageSeq(seq) ? seq : undefined;
}
export function dedupeMessagesByIdentity(messages: Message[]): Message[] {
const lastIndexByIdentity = new Map<string, number>();
const lastVisibleIndexByIdentity = new Map<string, number>();
// This is a UI-display dedupe rule, not a general LangChain message-stream
// contract. Hidden messages that share an identity with a visible message are
// treated as control messages for this merged view; hidden messages carrying
// independent tracing/task semantics should use a distinct id or a custom
// stream/state channel instead of relying on message dedupe preservation.
const preservedTurnDurations = new Map<string, number>();
messages.forEach((message, index) => {
const identity = messageIdentity(message);
if (identity) {
lastIndexByIdentity.set(identity, index);
if (!isHiddenFromUIMessage(message)) {
lastVisibleIndexByIdentity.set(identity, index);
}
if (message.additional_kwargs?.turn_duration !== undefined) {
preservedTurnDurations.set(
identity,
message.additional_kwargs.turn_duration as number,
);
}
}
});
return messages
.filter((message, index) => {
const identity = messageIdentity(message);
if (!identity) {
return true;
}
const visibleIndex = lastVisibleIndexByIdentity.get(identity);
if (visibleIndex !== undefined) {
return visibleIndex === index;
}
return lastIndexByIdentity.get(identity) === index;
})
.map((message) => {
const identity = messageIdentity(message);
if (
identity &&
preservedTurnDurations.has(identity) &&
message.additional_kwargs?.turn_duration === undefined
) {
return {
...message,
additional_kwargs: {
...message.additional_kwargs,
turn_duration: preservedTurnDurations.get(identity),
},
} as Message;
}
return message;
});
}
/**
* Insert messages carrying a trusted seq into an already-ordered list at
* their position. Entries without a trustworthy position are never moved to
* accommodate an insertion, and an *inserted* message whose seq exceeds every
* known position keeps the established tail. Shared by mergeMessages'
* callers that overlay rescued messages onto canonical history, so the seq
* skeleton always outranks anchor guesses.
*/
export function insertByTrustedSeq(
base: Message[],
positioned: Message[],
): Message[] {
if (positioned.length === 0) {
return base;
}
const sorted = [...positioned].sort(
(left, right) =>
(trustedMessageSeq(left) ?? Number.POSITIVE_INFINITY) -
(trustedMessageSeq(right) ?? Number.POSITIVE_INFINITY),
);
const result: Message[] = [];
let next = 0;
for (const message of base) {
const seq = trustedMessageSeq(message);
if (seq !== undefined) {
while (
next < sorted.length &&
(trustedMessageSeq(sorted[next]!) ?? Number.POSITIVE_INFINITY) < seq
) {
result.push(sorted[next]!);
next += 1;
}
}
result.push(message);
}
while (next < sorted.length) {
result.push(sorted[next]!);
next += 1;
}
return result;
}
type PositionedMessage = {
message: Message;
// Tuple sort key. Seq-positioned entries sit at [seq, 0]; everything else
// anchors to a positioned neighbour (see mergeMessages). Equal keys fall
// back to insertion order via the stable sort, which keeps a no-seq
// segment's internal order without extra key spacing.
major: number;
minor: number;
};
/**
* Merge canonical history, the live checkpoint tail, and optimistic messages
* into one display list.
*
* Steps (design §4.2):
* 1. Content and position are merged per identity: the freshest visible copy
* supplies content, the earliest valid seq supplies the trusted position.
* 2. Every identity with a trusted seq joins an ascending skeleton — covering
* rows before the loaded window, gaps inside it, rows after it, and the
* no-shared-identity case alike (this replaces the old canonicalMinSeq
* local rule, which could not place a window-internal gap: R4).
* 3. Live-only messages without a seq keep the existing anchor weaving: a
* pending segment goes before its next shared or positioned anchor. A
* trailing segment follows its last live-only positioned anchor unless
* that anchor predates the loaded window; otherwise it keeps the tail.
* Its internal order is untouched.
* 4. When a speculative anchor and the skeleton disagree, the skeleton wins;
* established seq order is never reversed to fit a no-seq segment.
*
* Content replacement keeps trusted ordering metadata (R3): a live copy
* without a seq refreshes the text but inherits the known position, run_id,
* and turn_duration instead of resetting them.
*
* Runs in O(n log n) per structural change: identity lookups are Map/Set and
* the single sort happens once per merged snapshot, not per token.
*/
export function mergeMessages(
historyMessages: Message[],
threadMessages: Message[],
optimisticMessages: Message[],
): Message[] {
// Pass 1: per-identity trusted position and the historical metadata worth
// preserving. The earliest valid seq across copies wins; a hidden control
// copy only contributes a position when no visible copy carries one, so a
// re-keyed reminder can never drag the visible user turn to its own row.
const visibleSeqByIdentity = new Map<string, number>();
const anySeqByIdentity = new Map<string, number>();
const savedTurnDurations = new Map<string, number>();
const savedRunIds = new Map<string, string>();
const collectTrustedSeq = (message: Message) => {
const identity = messageIdentity(message);
if (!identity) {
return;
}
const seq = trustedMessageSeq(message);
if (seq === undefined) {
return;
}
const known = anySeqByIdentity.get(identity);
if (known === undefined || seq < known) {
anySeqByIdentity.set(identity, seq);
}
if (!isHiddenFromUIMessage(message)) {
const knownVisible = visibleSeqByIdentity.get(identity);
if (knownVisible === undefined || seq < knownVisible) {
visibleSeqByIdentity.set(identity, seq);
}
}
};
const trustedSeqOf = (identity: string | undefined) =>
identity === undefined
? undefined
: (visibleSeqByIdentity.get(identity) ?? anySeqByIdentity.get(identity));
for (const message of historyMessages) {
collectTrustedSeq(message);
const identity = messageIdentity(message);
const runId = getMessageRunId(message);
if (identity && runId) {
savedRunIds.set(identity, runId);
}
if (identity && message.additional_kwargs?.turn_duration !== undefined) {
savedTurnDurations.set(
identity,
message.additional_kwargs.turn_duration as number,
);
}
}
for (const message of threadMessages) {
collectTrustedSeq(message);
}
const canonical = dedupeMessagesByIdentity(historyMessages);
const live = dedupeMessagesByIdentity(threadMessages);
const canonicalByIdentity = new Map(
canonical.flatMap((message) => {
const identity = messageIdentity(message);
return identity ? [[identity, message] as const] : [];
}),
);
// Pass 2: walk the live tail. Shared identities are ordering anchors and
// may replace the canonical content; live-only messages with a trusted seq
// join the skeleton; the rest accumulate into no-seq segments woven before
// their next shared anchor.
const replacementByIdentity = new Map<string, Message>();
const beforeAnchor = new Map<string, Message[]>();
const skeletonLive: Message[] = [];
let pending: Message[] = [];
let trailingAnchorSeq: number | undefined;
for (const message of live) {
const identity = messageIdentity(message);
const canonicalMessage = identity
? canonicalByIdentity.get(identity)
: undefined;
if (identity && canonicalMessage) {
if (pending.length > 0) {
beforeAnchor.set(identity, [
...(beforeAnchor.get(identity) ?? []),
...pending,
]);
}
pending = [];
trailingAnchorSeq = undefined;
// A hidden checkpoint control message must not replace a visible
// canonical user turn that happens to reuse its identity. In every
// other case the live checkpoint copy is fresher and replaces history
// without moving it.
if (
!isHiddenFromUIMessage(message) ||
isHiddenFromUIMessage(canonicalMessage)
) {
replacementByIdentity.set(identity, message);
}
continue;
}
if (identity && trustedSeqOf(identity) !== undefined) {
// A positioned live-only result also anchors its preceding steps.
beforeAnchor.set(identity, pending);
pending = [];
skeletonLive.push(message);
trailingAnchorSeq = trustedSeqOf(identity);
continue;
}
pending.push(message);
}
// Only a live-only positioned anchor can pull trailing steps into a gap.
// After a shared anchor, preserve canonical source order before the tail.
const trailingPending = pending;
// Pass 3: position canonical entries. A canonical entry without a trusted
// seq stays in its established position relative to the previous
// seq-positioned entry (or at the front when none exists yet).
const entries: PositionedMessage[] = [];
let minorCounter = 0;
let previousCanonicalSeq = 0;
let firstCanonicalSeq = Number.POSITIVE_INFINITY;
for (const message of canonical) {
const identity = messageIdentity(message);
const seq = trustedSeqOf(identity);
let major: number;
let minor: number;
if (seq !== undefined) {
major = seq;
minor = 0;
previousCanonicalSeq = seq;
firstCanonicalSeq = Math.min(firstCanonicalSeq, seq);
} else {
major = previousCanonicalSeq;
minor = ++minorCounter;
}
if (identity) {
// A no-seq segment known to precede this anchor sorts immediately
// before it; stable sort keeps the segment's internal order.
const segment = beforeAnchor.get(identity);
if (segment) {
for (const segmentMessage of segment) {
entries.push({
message: segmentMessage,
major,
minor: minor - 0.5,
});
}
}
}
const replacement = identity
? replacementByIdentity.get(identity)
: undefined;
entries.push({ message: replacement ?? message, major, minor });
}
for (const message of skeletonLive) {
const seq = trustedSeqOf(messageIdentity(message));
if (seq !== undefined) {
for (const segmentMessage of beforeAnchor.get(
messageIdentity(message)!,
) ?? []) {
entries.push({ message: segmentMessage, major: seq, minor: -0.5 });
}
entries.push({ message, major: seq, minor: 0 });
}
}
// A rescued early input may precede an unloaded history gap while its
// followers belong to the new run (#4666). Keep those followers at the
// tail; within the loaded window, keep trailing steps beside their result.
const trailingMajor =
trailingAnchorSeq !== undefined && trailingAnchorSeq >= firstCanonicalSeq
? trailingAnchorSeq
: Number.POSITIVE_INFINITY;
for (const message of trailingPending) {
entries.push({ message, major: trailingMajor, minor: 0.5 });
}
for (const message of optimisticMessages) {
entries.push({
message,
major: Number.POSITIVE_INFINITY,
minor: ++minorCounter,
});
}
const ordered = entries
.slice()
.sort((left, right) => left.major - right.major || left.minor - right.minor)
.map((entry) => entry.message);
const merged = dedupeMessagesByIdentity(ordered);
// Pass 4: re-attach trusted ordering metadata that content replacement
// dropped. A missing or invalid seq never overwrites the known position;
// other additional_kwargs, run_id, and turn_duration of the winning content
// copy ride along untouched.
return merged.map((message) => {
const identity = messageIdentity(message);
if (!identity) {
return message;
}
const trustedSeq = trustedSeqOf(identity);
const shouldRestoreSeq =
trustedSeq !== undefined && trustedMessageSeq(message) !== trustedSeq;
const shouldRestoreRunId =
savedRunIds.has(identity) && !getMessageRunId(message);
const shouldRestoreTurnDuration =
savedTurnDurations.has(identity) &&
message.additional_kwargs?.turn_duration === undefined;
if (
!shouldRestoreSeq &&
!shouldRestoreRunId &&
!shouldRestoreTurnDuration
) {
return message;
}
return {
...message,
...(shouldRestoreRunId ? { run_id: savedRunIds.get(identity) } : {}),
additional_kwargs: {
...message.additional_kwargs,
...(shouldRestoreSeq ? { [MESSAGE_SEQ_KEY]: trustedSeq } : {}),
...(shouldRestoreTurnDuration
? { turn_duration: savedTurnDurations.get(identity) }
: {}),
},
} as Message;
});
}

483
frontend/tests/e2e/thread-ordering.spec.ts Normal file
View File

@ -0,0 +1,483 @@
import { expect, test, type Page, type Route } from "@playwright/test";
import {
mockLangGraphAPI,
MOCK_THREAD_ID,
MOCK_RUN_ID,
} from "./utils/mock-api";
/**
* Browser-level composition regression for the message-ordering contract
* (design §5.3): a deterministic fixture with more than one feed page
* (>50 rows), many turns, and two past compactions (the hidden summary
* rows), then a live third compaction during a newly submitted turn.
*
* Assertions happen at stage barriers: the relative DOM order of human
* messages and steps, the visible message set, and a tool card's
* association — then the same server data is reloaded and compared. The
* virtual list is exercised through scroll/outline navigation instead of
* assuming every message is in the DOM at once.
*/
const PAGE_SIZE = 50;
const THREAD = {
thread_id: MOCK_THREAD_ID,
title: "Ordering conversation",
updated_at: "2026-09-08T12:00:00Z",
};
type FeedMessage = Record<string, unknown>;
function turnMessages(turn: number): FeedMessage[] {
return [
{
type: "human",
id: `h-turn-${turn}`,
content: `turn-${turn} question`,
},
{ type: "ai", id: `a-turn-${turn}`, content: `turn-${turn} answer` },
];
}
function hiddenSummary(id: string, text: string): FeedMessage {
return {
type: "ai",
id,
name: "summary",
content: text,
additional_kwargs: { hide_from_ui: true },
};
}
/** 68 rows: 33 turns plus two hidden summary rows from past compactions. */
function buildFixtureMessages(): FeedMessage[] {
const messages: FeedMessage[] = [];
for (let turn = 0; turn <= 9; turn += 1) {
messages.push(...turnMessages(turn));
}
// First compaction collapsed everything above into this hidden summary.
messages.push(hiddenSummary("summary-1", "context summary one"));
for (let turn = 10; turn <= 19; turn += 1) {
messages.push(...turnMessages(turn));
}
// Second compaction.
messages.push(hiddenSummary("summary-2", "context summary two"));
for (let turn = 20; turn <= 29; turn += 1) {
messages.push(...turnMessages(turn));
}
// A tool-using turn so the tool card association is verifiable. Two tool
// calls put the web_search step above the LAST one — into the collapsed
// "more steps" region — so the payload check below exercises the real
// expand interaction instead of the always-open trailing step.
messages.push({
type: "human",
id: "h-turn-30",
content: "turn-30 question",
});
messages.push({
type: "ai",
id: "a-turn-30",
content: "",
tool_calls: [
{
id: "call-turn-30",
name: "web_search",
args: { query: "turn-30 lookup" },
type: "tool_call",
},
{
id: "call-turn-30-b",
name: "web_fetch",
args: { url: "https://example.test/turn-30" },
type: "tool_call",
},
],
});
messages.push({
type: "tool",
id: "t-turn-30",
tool_call_id: "call-turn-30",
name: "web_search",
// web_search steps render parsed search-result items, so the payload is
// the JSON-stringified array the real tool produces.
content: JSON.stringify([
{ url: "https://example.test/turn-30", title: "tool-30 result payload" },
]),
});
messages.push({
type: "tool",
id: "t-turn-30-b",
tool_call_id: "call-turn-30-b",
name: "web_fetch",
content: "fetched page body",
});
messages.push({
type: "ai",
id: "a-turn-30-final",
content: "turn-30 answer",
});
messages.push(...turnMessages(31));
return messages;
}
type FeedRow = {
run_id: string;
seq: number;
content: FeedMessage;
metadata: { caller: string };
created_at: string;
};
function toFeedRows(messages: FeedMessage[]): FeedRow[] {
return messages.map((message, index) => ({
run_id: `run-${MOCK_THREAD_ID}`,
seq: index + 1,
content: message,
metadata: { caller: "lead_agent" },
created_at: `2026-09-08T00:${String(Math.floor(index / 60)).padStart(2, "0")}:${String(index % 60).padStart(2, "0")}Z`,
}));
}
/** Register a paginating `/messages/page` backed by the mutable feed rows. */
async function mockPaginatedFeed(page: Page, rows: FeedRow[]) {
await page.route(/\/api\/threads\/[^/]+\/messages\/page/, (route) => {
if (route.request().method() !== "GET") {
return route.fallback();
}
const url = new URL(route.request().url());
const beforeSeqParam = url.searchParams.get("before_seq");
const eligible =
beforeSeqParam === null
? rows
: rows.filter((row) => row.seq < Number(beforeSeqParam));
const pageRows = eligible.slice(-PAGE_SIZE);
const hasMore = eligible.length > pageRows.length;
return route.fulfill({
status: 200,
contentType: "application/json",
body: JSON.stringify({
data: pageRows,
has_more: hasMore,
next_before_seq: hasMore ? pageRows[0]!.seq : null,
}),
});
});
}
/** Rendered message-group indices must appear in ascending DOM order. */
async function expectGroupIndicesAscending(page: Page) {
const indices = await page
.locator("[data-message-group-index]")
.evaluateAll((elements) =>
elements.map((element) =>
Number(element.getAttribute("data-message-group-index")),
),
);
expect(indices.length).toBeGreaterThan(0);
const sorted = [...indices].sort((left, right) => left - right);
expect(indices).toEqual(sorted);
}
async function jumpToChapter(page: Page, title: string | RegExp) {
await page.getByTestId("conversation-outline-trigger").click();
// The menu sits above a smoothly-scrolling virtual list; late chapters need
// an in-menu scroll, which can starve actionability checks. The item is
// resolved and attached, so click through.
await page.getByRole("menuitem", { name: title }).click({ force: true });
}
async function loadAllHistoryPages(page: Page) {
const loadMore = page.getByRole("button", { name: "Load more" });
// Each click prepends one older page; keep clicking until exhausted. The
// near-top sentinel may already have auto-loaded a page, so the button is
// not guaranteed to be there on every pass.
for (let attempt = 0; attempt < 10; attempt += 1) {
// Clicking a menu item closes the dropdown, so each pass starts from a
// closed menu: open it, read the earliest loaded chapter, then jump to it
// with the same click.
await page.getByTestId("conversation-outline-trigger").click();
const firstItem = page.getByRole("menuitem").first();
const firstTitle = (await firstItem.textContent()) ?? "";
await firstItem.click();
if (firstTitle.includes("turn-0 question")) {
return;
}
if (!(await loadMore.isVisible())) {
// The sentinel auto-load is in flight; give it a beat to land.
await page.waitForResponse(
(response) => response.url().includes("/messages/page"),
{ timeout: 5_000 },
);
continue;
}
const pageResponse = page.waitForResponse((response) =>
response.url().includes("/messages/page"),
);
await loadMore.click();
await pageResponse;
}
throw new Error("turn-0 chapter never became reachable");
}
test.describe("Thread message ordering", () => {
// Long conversations paginate and virtualize; navigation through the
// outline plus a reload needs more than the 30s default.
test.setTimeout(90_000);
test("paginated long history renders turns in feed order and survives refresh", async ({
page,
}) => {
const rows = toFeedRows(buildFixtureMessages());
mockLangGraphAPI(page, {
threads: [{ ...THREAD, messages: rows.map((row) => row.content) }],
});
await mockPaginatedFeed(page, rows);
await page.goto(`/workspace/chats/${MOCK_THREAD_ID}`);
// Newest page loads first and the view starts at the bottom.
await expect(page.getByText("turn-31 answer")).toBeVisible({
timeout: 15_000,
});
await expectGroupIndicesAscending(page);
// Hidden compaction summaries never render.
await expect(page.getByText("context summary one")).toHaveCount(0);
await expect(page.getByText("context summary two")).toHaveCount(0);
// Walk to the head of the conversation through pagination + outline
// navigation; the virtual list only renders the visited window. The walk
// ends on the turn-0 chapter.
await loadAllHistoryPages(page);
// Scope to the message list: the outline menu carries the same text.
await expect(
page.getByTestId("main-message-list").getByText("turn-0 question"),
).toBeVisible();
await expectGroupIndicesAscending(page);
// The tool turn keeps its card association: the web_search step sits in
// the collapsed "more steps" region, so its intermediate result payload
// is hidden until the region is expanded — and must still be there after.
await jumpToChapter(page, /turn-30 question/);
const mainList = page.getByTestId("main-message-list");
await expect(mainList.getByText("turn-30 answer")).toBeVisible();
const payload = mainList.getByText("tool-30 result payload");
await expect(payload).not.toBeVisible();
await mainList.getByRole("button", { name: "1 more step" }).click();
await expect(payload).toBeVisible();
// Same server data after a refresh reconstructs the same order.
await page.reload();
await expect(page.getByText("turn-31 answer")).toBeVisible({
timeout: 15_000,
});
await expectGroupIndicesAscending(page);
await loadAllHistoryPages(page);
await expect(
page.getByTestId("main-message-list").getByText("turn-0 question"),
).toBeVisible();
});
test("a live compaction during submit keeps history order and appends the new turn", async ({
page,
}) => {
const messages = buildFixtureMessages();
const rows = toFeedRows(messages);
mockLangGraphAPI(page, {
threads: [{ ...THREAD, messages: [...messages] }],
});
await mockPaginatedFeed(page, rows);
// A run stream in the real frame format: metadata, the summarization
// update (RemoveMessage(ALL) + hidden summary + retained tail), the
// retained tail re-arriving as messages-tuple frames, then the streamed
// answer and the end of the run. The feed grows by the new turn so the
// finishing history refetch observes it.
const compactionStream = async (route: Route) => {
const submitted = (
route.request().postDataJSON() as {
input?: { messages?: FeedMessage[] };
}
)?.input?.messages;
const submittedHuman = (submitted ?? []).find(
(message) => message.type === "human",
);
// The runtime keeps the just-submitted human in the post-compaction
// checkpoint state — the SDK-side human count only advances if the
// stream echoes it, which is also what hides the optimistic copy.
const serverHuman: FeedMessage = {
type: "human",
id: "h-turn-32",
content: submittedHuman?.content ?? "final-turn question",
};
const retainedTail = [...turnMessages(31)];
const removeAll = { id: "__remove_all__", type: "remove" };
const summary3 = hiddenSummary("summary-3", "context summary three");
const answer = {
type: "ai",
id: "a-turn-32",
content: "final-turn answer",
};
const frames = [
{
event: "metadata",
data: { run_id: MOCK_RUN_ID, thread_id: MOCK_THREAD_ID },
},
// The summarization middleware update the frontend's compaction
// rescue keys off: RemoveMessage(ALL) + hidden summary + retained
// tail, with the just-submitted human still in state.
{
event: "updates",
data: {
"DeerFlowSummarizationMiddleware.before_model": {
messages: [removeAll, summary3, ...retainedTail, serverHuman],
},
},
},
// The post-compaction checkpoint wholesale, as the established
// handleRunStream mock convention delivers it: hidden summary,
// retained tail, submitted human, and the streamed answer.
{
event: "values",
data: {
messages: [summary3, ...retainedTail, serverHuman, answer],
},
},
{ event: "end", data: {} },
];
// The persisted feed grows by the new turn so the finishing history
// refetch observes it, mirroring the journal flush on the real backend.
const base = rows.length;
const newTurn: FeedMessage[] = [serverHuman, answer];
rows.push(
...newTurn.map((message, index) => ({
run_id: `run-${MOCK_THREAD_ID}`,
seq: base + index + 1,
content: message,
metadata: { caller: "lead_agent" },
created_at: "2026-09-08T01:00:00Z",
})),
);
const body = frames
.map(
(frame) =>
`event: ${frame.event}\ndata: ${JSON.stringify(frame.data)}\n\n`,
)
.join("");
return route.fulfill({
status: 200,
contentType: "text/event-stream",
body,
});
};
await page.route("**/api/langgraph/threads/*/runs/stream", (route) =>
compactionStream(route),
);
await page.route("**/api/langgraph/runs/stream", (route) =>
compactionStream(route),
);
await page.goto(`/workspace/chats/${MOCK_THREAD_ID}`);
await expect(page.getByText("turn-31 answer")).toBeVisible({
timeout: 15_000,
});
const textarea = page.getByPlaceholder(/how can i assist you/i);
await textarea.fill("final-turn question");
await textarea.press("Enter");
// The streamed answer lands below the submitted question. Note: the
// count-based optimistic clearing does not fire when compaction shrinks
// the SDK human count below its pre-submit value (a known limitation
// tracked with the turn-ownership work), so a trailing optimistic twin
// can remain; the ordering contract asserted here is about the canonical
// server copy.
await expect(page.getByText("final-turn answer")).toBeVisible({
timeout: 15_000,
});
// DOM relative order, not viewport coordinates: stick-to-bottom smooth
// scrolling makes two separate boundingBox reads race each other.
const questionBeforeAnswer = await page.evaluate(() => {
const list = document.querySelector('[data-testid="main-message-list"]');
if (!list) {
return null;
}
const leaf = (text: string) =>
[...list.querySelectorAll("div, p")].find(
(element) =>
element.children.length === 0 && element.textContent === text,
);
const question = leaf("final-turn question");
const answer = leaf("final-turn answer");
if (!question || !answer) {
return null;
}
return (
(question.compareDocumentPosition(answer) &
Node.DOCUMENT_POSITION_FOLLOWING) !==
0
);
});
expect(questionBeforeAnswer).toBe(true);
// The finishing refetch observed the extended feed; the established
// order — including the compacted head — is unchanged.
await expectGroupIndicesAscending(page);
await expect(page.getByText("turn-31 answer")).toBeVisible();
await expect(page.getByText("context summary three")).toHaveCount(0);
await loadAllHistoryPages(page);
await expect(
page.getByTestId("main-message-list").getByText("turn-0 question"),
).toBeVisible();
// Refresh against the same server data: identical order.
await page.reload();
await expect(page.getByText("final-turn answer")).toBeVisible({
timeout: 15_000,
});
await expectGroupIndicesAscending(page);
});
test("custom agent chat shares the same ordering across pagination and refresh", async ({
page,
}) => {
// The Custom Agent route renders the same ChatPage/message pipeline; this
// pins the shared link so the ordering contract cannot regress on one
// path only.
const rows = toFeedRows(buildFixtureMessages());
mockLangGraphAPI(page, {
agents: [
{
name: "ordering-agent",
description: "Agent for the ordering regression",
},
],
threads: [
{
...THREAD,
agent_name: "ordering-agent",
messages: rows.map((row) => row.content),
},
],
});
await mockPaginatedFeed(page, rows);
await page.goto(`/workspace/agents/ordering-agent/chats/${MOCK_THREAD_ID}`);
await expect(page.getByText("turn-31 answer")).toBeVisible({
timeout: 15_000,
});
await expectGroupIndicesAscending(page);
await loadAllHistoryPages(page);
await expect(
page.getByTestId("main-message-list").getByText("turn-0 question"),
).toBeVisible();
await expectGroupIndicesAscending(page);
await page.reload();
await expect(page.getByText("turn-31 answer")).toBeVisible({
timeout: 15_000,
});
await expectGroupIndicesAscending(page);
});
});

65
frontend/tests/e2e/utils/mock-api.ts
View File

@ -1098,6 +1098,71 @@ export function mockLangGraphAPI(page: Page, options?: MockAPIOptions) {
return route.fallback();
});
// Token usage — the chat header polls this per thread. Without a mock the
// request falls through to a gateway that is not running under Playwright,
// and a 401 there redirects the whole page to /login mid-test.
void page.route("**/api/threads/*/token-usage", (route) => {
if (route.request().method() === "GET") {
const threadId = /\/api\/threads\/([^/]+)\/token-usage/.exec(
route.request().url(),
)?.[1];
return route.fulfill({
status: 200,
contentType: "application/json",
body: JSON.stringify({
thread_id: threadId ?? "unknown",
total_tokens: 0,
total_input_tokens: 0,
total_output_tokens: 0,
total_runs: 0,
by_model: {},
by_caller: { lead_agent: 0, subagent: 0, middleware: 0 },
context_usage: null,
}),
});
}
return route.fallback();
});
// MCP background tasks — same fallthrough-to-401 problem as token-usage.
void page.route("**/api/threads/*/mcp-tasks*", (route) => {
if (route.request().method() === "GET") {
return route.fulfill({
status: 200,
contentType: "application/json",
body: JSON.stringify([]),
});
}
return route.fallback();
});
// Workspace changes — the run-scoped badge query. Unmocked it 401s against
// the absent gateway and the fetcher redirects the page to /login.
void page.route("**/api/threads/*/runs/*/workspace-changes*", (route) => {
if (route.request().method() === "GET") {
return route.fulfill({
status: 200,
contentType: "application/json",
body: JSON.stringify({
run_id: "mock-run",
thread_id: "mock-thread",
status: "success",
summary: {
created: 0,
modified: 0,
deleted: 0,
symlink_created: 0,
additions: 0,
deletions: 0,
truncated: false,
},
changes: [],
}),
});
}
return route.fallback();
});
// Thread history — useStream fetches state history on mount
void page.route("**/api/langgraph/threads/*/history", (route) => {
const url = route.request().url();

79
frontend/tests/unit/core/threads/message-merge.test.ts
View File

@ -2,6 +2,7 @@ import type { Message } from "@langchain/langgraph-sdk";
import { expect, rs, test } from "@rstest/core";
import { InfiniteQueryObserver, QueryClient } from "@tanstack/react-query";
import { getMessageRunId } from "@/core/messages/run-duration";
import {
buildThreadMessagesPageUrl,
buildVisibleHistoryMessages,
@ -142,7 +143,9 @@ test("mergeMessages preserves historical run metadata on a live checkpoint repla
{
...checkpointAi,
run_id: "run-1",
additional_kwargs: { turn_duration: 114 },
// The replacement keeps the trusted feed position alongside the run
// metadata: dropping deerflow_seq here was defect R3.
additional_kwargs: { turn_duration: 114, deerflow_seq: 1 },
},
]);
});
@ -2302,3 +2305,77 @@ test("a checkpoint message earlier than the loaded window is placed by its seq e
"…new step 2",
]);
});
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
// feed established stays attached to the merged message.
const history = buildVisibleHistoryMessages(
[
{
run_id: "run-1",
seq: 1,
content: { id: "h1", type: "human", content: "question" } as Message,
metadata: { caller: "lead_agent" },
created_at: "2026-09-08T00:00:00Z",
},
{
run_id: "run-1",
seq: 2,
content: { id: "a1", type: "ai", content: "draft" } as Message,
metadata: { caller: "lead_agent" },
created_at: "2026-09-08T00:00:01Z",
},
],
new Set(),
);
const live = [
{ id: "h1", type: "human", content: "question" } as Message,
{ id: "a1", type: "ai", content: "final answer" } as Message,
];
const merged = mergeMessages(history, live, []);
expect(merged.map((message) => message.content)).toEqual([
"question",
"final answer",
]);
expect(
merged.map((message) => message.additional_kwargs?.deerflow_seq),
).toEqual([1, 2]);
expect(merged.map((message) => getMessageRunId(message))).toEqual([
"run-1",
"run-1",
]);
});
test("mergeMessages places a live message with seq inside the loaded window by position (R4)", () => {
// Identity anchors alone weave a pending live message before the NEXT shared
// anchor, inverting positions the server already settled: history
// H-first(1), A-second(3), A-last(5) plus live H-second(2), A-last(5) came
// out as 1,3,2,5. The trusted seq skeleton must produce 1,2,3,5.
const withSeq = (message: Message, seq: number) =>
({
...message,
additional_kwargs: { ...message.additional_kwargs, deerflow_seq: seq },
}) as Message;
const history = [
withSeq({ id: "h-first", type: "human", content: "first" } as Message, 1),
withSeq({ id: "a-second", type: "ai", content: "second" } as Message, 3),
withSeq({ id: "a-last", type: "ai", content: "last" } as Message, 5),
];
const live = [
withSeq(
{ id: "h-second", type: "human", content: "second q" } as Message,
2,
),
withSeq({ id: "a-last", type: "ai", content: "last" } as Message, 5),
];
const merged = mergeMessages(history, live, []);
expect(
merged.map((message) => message.additional_kwargs?.deerflow_seq),
).toEqual([1, 2, 3, 5]);
});

545
frontend/tests/unit/core/threads/message-order.test.ts Normal file
View File

@ -0,0 +1,545 @@
import type { Message } from "@langchain/langgraph-sdk";
import { expect, test } from "@rstest/core";
import { getMessageRunId } from "@/core/messages/run-duration";
import {
buildVisibleHistoryMessages,
mergeRenderedMessageLedger,
mergeMessages,
resolveThreadTransientHistoryBridge,
resolveTransientHistoryBridge,
} from "@/core/threads/hooks";
import {
dedupeMessagesByIdentity,
insertByTrustedSeq,
MESSAGE_SEQ_KEY,
mergeMessages as mergeMessagesFromModule,
messageIdentity,
trustedMessageSeq,
} from "@/core/threads/message-order";
import type { RunMessage } from "@/core/threads/types";
// mergeMessages is re-exported from hooks for compatibility; both names must
// resolve to the same implementation.
expect(mergeMessagesFromModule).toBe(mergeMessages);
function msg(
id: string,
type: "human" | "ai" | "tool" | "system",
content: string,
seq?: number,
extra?: Record<string, unknown>,
): Message {
return {
id,
type,
content,
...(extra ?? {}),
additional_kwargs: seq !== undefined ? { [MESSAGE_SEQ_KEY]: seq } : {},
} as Message;
}
function withKwargs(
message: Message,
kwargs: Record<string, unknown>,
): Message {
return {
...message,
additional_kwargs: { ...message.additional_kwargs, ...kwargs },
} as Message;
}
function row(
runId: string,
seq: number,
content: Message,
index = 0,
): RunMessage {
return {
run_id: runId,
seq,
content,
metadata: { caller: "lead_agent" },
created_at: `2026-09-08T00:00:${String(index).padStart(2, "0")}Z`,
};
}
function seqsOf(messages: Message[]): unknown[] {
return messages.map(
(message) => message.additional_kwargs?.[MESSAGE_SEQ_KEY],
);
}
function idsOf(messages: Message[]): (string | undefined)[] {
return messages.map((message) => message.id);
}
test("messageIdentity collapses X and X__user and keys tools by tool_call_id", () => {
expect(messageIdentity(msg("req-1", "human", "q"))).toBe("message:req-1");
expect(messageIdentity(msg("req-1__user", "human", "q"))).toBe(
"message:req-1",
);
// Only human copies collapse: a hidden system reminder legitimately reuses
// the original id and keeps its own identity.
expect(messageIdentity(msg("req-1", "system", "reminder"))).toBe(
"message:req-1",
);
expect(
messageIdentity({
id: "tool-own-id",
type: "tool",
content: "result",
tool_call_id: "call-1",
} as Message),
).toBe("tool:call-1");
});
test("trustedMessageSeq accepts only positive safe integers", () => {
expect(trustedMessageSeq(msg("a", "ai", "x", 3))).toBe(3);
expect(trustedMessageSeq(msg("a", "ai", "x"))).toBeUndefined();
const invalid = [
null,
"7",
Number.NaN,
1.5,
0,
-2,
Number.MAX_SAFE_INTEGER + 1,
];
for (const value of invalid) {
expect(
trustedMessageSeq(
withKwargs(msg("a", "ai", "x"), { [MESSAGE_SEQ_KEY]: value }),
),
`seq=${String(value)}`,
).toBeUndefined();
}
});
test("live content refresh without seq preserves seq, run_id, and turn_duration", () => {
const history = [
withKwargs(
{
...msg("h1", "human", "question", 1),
run_id: "run-1",
} as unknown as Message,
{ turn_duration: 42 },
),
msg("a1", "ai", "draft", 2),
];
const live = [msg("h1", "human", "question"), msg("a1", "ai", "final")];
const merged = mergeMessages(history, live, []);
expect(idsOf(merged)).toEqual(["h1", "a1"]);
expect(merged.map((message) => message.content)).toEqual([
"question",
"final",
]);
expect(seqsOf(merged)).toEqual([1, 2]);
expect(getMessageRunId(merged[0]!)).toBe("run-1");
expect(merged[0]!.additional_kwargs?.turn_duration).toBe(42);
});
test("hidden control copy never overwrites the visible human or its position", () => {
// DynamicContextMiddleware shape: hidden SystemMessage(id=X) plus the
// visible HumanMessage(id=X__user). One visible message survives, carrying
// the trusted position.
const hiddenControl = withKwargs(msg("req-1", "system", "reminder", 4), {
hide_from_ui: true,
});
const visibleHuman = msg("req-1__user", "human", "real question", 5);
const merged = mergeMessages(
[visibleHuman],
[hiddenControl, msg("req-1__user", "human", "real question")],
[],
);
expect(idsOf(merged)).toEqual(["req-1__user"]);
// The surviving message is the visible human — never the hidden control —
// and the hidden reminder's earlier row does not drag it off its visible
// position.
expect(merged[0]!.type).toBe("human");
expect(merged[0]!.additional_kwargs?.[MESSAGE_SEQ_KEY]).toBe(5);
});
test("tool results dedupe by tool_call_id and keep the known position", () => {
const historyTool = {
id: "tool-old-row",
type: "tool",
content: "partial",
tool_call_id: "call-1",
additional_kwargs: { [MESSAGE_SEQ_KEY]: 6 },
} as Message;
const liveTool = {
id: "tool-new-row",
type: "tool",
content: "complete",
tool_call_id: "call-1",
} as Message;
const merged = mergeMessages(
[msg("h1", "human", "q", 5), historyTool],
[msg("h1", "human", "q"), liveTool],
[],
);
expect(merged).toHaveLength(2);
const tool = merged.find((message) => message.type === "tool")!;
expect(tool.content).toBe("complete");
expect(tool.tool_call_id).toBe("call-1");
expect(tool.additional_kwargs?.[MESSAGE_SEQ_KEY]).toBe(6);
});
test("invalid seq values never overwrite a trusted position or break the merge", () => {
const invalid = [null, "7", Number.NaN, 1.5, Number.MAX_SAFE_INTEGER + 1];
for (const bad of invalid) {
const history = [msg("h1", "human", "q", 3), msg("a1", "ai", "a", 4)];
const live = [
withKwargs(msg("h1", "human", "q"), { [MESSAGE_SEQ_KEY]: bad }),
withKwargs(msg("a1", "ai", "a2"), { [MESSAGE_SEQ_KEY]: bad }),
];
const merged = mergeMessages(history, live, []);
expect(idsOf(merged), `seq=${String(bad)}`).toEqual(["h1", "a1"]);
expect(seqsOf(merged), `seq=${String(bad)}`).toEqual([3, 4]);
expect(merged[1]!.content).toBe("a2");
}
});
test("a live-only message with seq fills the window-internal gap (1,3,5 + 2,5)", () => {
const history = [
msg("h1", "human", "first", 1),
msg("a1", "ai", "second", 3),
msg("a2", "ai", "last", 5),
];
const live = [msg("h2", "human", "middle", 2), msg("a2", "ai", "last", 5)];
expect(seqsOf(mergeMessages(history, live, []))).toEqual([1, 2, 3, 5]);
});
test("seq positions merge even when the two sides share no identity", () => {
const history = [
msg("old-1", "human", "old q", 10),
msg("old-2", "ai", "old a", 11),
];
const live = [
msg("rescued", "human", "rescued turn", 2),
msg("tail-step", "ai", "streaming step"),
];
const merged = mergeMessages(history, live, []);
expect(idsOf(merged)).toEqual(["rescued", "old-1", "old-2", "tail-step"]);
});
test("two no-seq segments between known anchors keep their internal order", () => {
const history = [msg("h1", "human", "q1", 1), msg("a9", "ai", "a9", 9)];
const live = [
msg("h1", "human", "q1"),
msg("s1", "ai", "step 1"),
msg("s2", "tool", "step 2"),
msg("a9", "ai", "a9"),
msg("s3", "ai", "step 3"),
msg("s4", "ai", "step 4"),
];
const merged = mergeMessages(history, live, []);
expect(idsOf(merged)).toEqual(["h1", "s1", "s2", "a9", "s3", "s4"]);
});
test("repeated feed updates keep latest content at the earliest visible position", () => {
// Real feed shape for an identity persisted by several run events: the same
// message id occupies multiple rows. The backend resolves such an identity
// to its earliest feed position (get_message_seqs), so the visible copy
// must carry the newest content with the earliest seq.
const rows = [
row("run-1", 1, msg("h1", "human", "question"), 0),
row("run-1", 2, msg("a1", "ai", "first draft"), 1),
row("run-1", 3, msg("a1", "ai", "updated answer"), 2),
];
const visible = buildVisibleHistoryMessages(rows, new Set());
expect(visible.map((message) => message.content)).toEqual([
"question",
"updated answer",
]);
expect(seqsOf(visible)).toEqual([1, 2]);
});
test("a hidden control row does not contribute the visible position", () => {
const hiddenRow = row(
"run-1",
2,
withKwargs(msg("req-1", "system", "reminder"), { hide_from_ui: true }),
0,
);
const visibleRow = row("run-1", 5, msg("req-1__user", "human", "q"), 1);
const visible = buildVisibleHistoryMessages(
[visibleRow, hiddenRow],
new Set(),
);
expect(visible).toHaveLength(1);
expect(visible[0]!.type).toBe("human");
// Earliest *visible* row wins; the hidden control copy at seq 2 must not
// pull the user message forward.
expect(visible[0]!.additional_kwargs?.[MESSAGE_SEQ_KEY]).toBe(5);
});
test("an established seq order is never reversed by a conflicting live order", () => {
const history = [msg("h1", "human", "q", 1), msg("a1", "ai", "a", 2)];
// Live delivers the same identities in inverted order: the skeleton keeps
// the server-settled sequence.
const live = [msg("a1", "ai", "a", 2), msg("h1", "human", "q", 1)];
expect(seqsOf(mergeMessages(history, live, []))).toEqual([1, 2]);
});
test("merging is idempotent and insensitive to duplicate redelivery order", () => {
const history = [
msg("h1", "human", "q1", 1),
msg("a1", "ai", "a1", 2),
msg("h2", "human", "q2", 4),
];
const live = [msg("a2", "ai", "a2 streaming", 5), msg("h2", "human", "q2")];
const optimistic = [msg("opt-1", "human", "draft")];
const once = mergeMessages(history, live, optimistic);
const twice = mergeMessages(once, live, optimistic);
expect(idsOf(twice)).toEqual(idsOf(once));
expect(seqsOf(twice)).toEqual(seqsOf(once));
// The same deterministic information delivered in a different order
// converges to the same sequence (optimistic tail keeps arrival order).
const reordered = mergeMessages(
[...history].reverse(),
[...live].reverse(),
optimistic,
);
expect(idsOf(reordered)).toEqual(idsOf(once));
});
test("mergeMessages never mutates its inputs", () => {
const history = [msg("h1", "human", "q", 1)];
const live = [msg("h1", "human", "q"), msg("a1", "ai", "a")];
const historySnapshot = JSON.stringify(history);
const liveSnapshot = JSON.stringify(live);
mergeMessages(history, live, []);
expect(JSON.stringify(history)).toBe(historySnapshot);
expect(JSON.stringify(live)).toBe(liveSnapshot);
});
test("two compactions with a moving window keep visible identities and never revive superseded runs", () => {
// Turn 1 and 2 rendered; compaction 1 retains a tail; the page window moves;
// compaction 2 fires. Across both cycles every displayed identity survives
// in ascending seq order, and the superseded run's rows stay hidden.
const supersededRun = "run-old";
const rows = [
row(supersededRun, 1, msg("h-old", "human", "superseded question"), 0),
row(supersededRun, 2, msg("a-old", "ai", "superseded answer"), 1),
row("run-2", 3, msg("h1", "human", "q1"), 2),
row("run-2", 4, msg("a1", "ai", "a1"), 3),
row("run-3", 5, msg("h2", "human", "q2"), 4),
row("run-3", 6, msg("a2", "ai", "a2"), 5),
];
const history = buildVisibleHistoryMessages(rows, new Set([supersededRun]));
expect(idsOf(history)).toEqual(["h1", "a1", "h2", "a2"]);
// First compaction: live window keeps only the recent tail.
const firstLedger = mergeRenderedMessageLedger([], history);
const firstCompactionLive = [msg("h2", "human", "q2"), msg("a2", "ai", "a2")];
const afterFirst = mergeMessages(history, firstCompactionLive, []);
expect(idsOf(afterFirst)).toEqual(["h1", "a1", "h2", "a2"]);
// The window moves: the refreshed page no longer reaches turn 1. The
// rendered ledger still anchors it.
const movedWindow = [msg("h2", "human", "q2", 5), msg("a2", "ai", "a2", 6)];
const ledger = mergeRenderedMessageLedger(firstLedger, movedWindow);
// Second compaction removes everything again; the ledger plus the fresh
// window must still reconstruct the full visible conversation in order.
const afterSecond = mergeMessages(ledger, movedWindow, []);
expect(idsOf(afterSecond)).toEqual(["h1", "a1", "h2", "a2"]);
expect(seqsOf(afterSecond)).toEqual([3, 4, 5, 6]);
// Superseded rows never reappear, even though the ledger has seen them not.
expect(idsOf(afterSecond)).not.toContain("h-old");
expect(idsOf(afterSecond)).not.toContain("a-old");
});
test("bridge resolution never inherits positions across threads", () => {
const historyB = [msg("hb", "human", "thread B question", 1)];
const transientA = [msg("ha", "human", "thread A rescued", 7)];
// Wrong-thread bridge is rejected outright, seq or not.
expect(
resolveThreadTransientHistoryBridge(
historyB,
transientA,
"thread-a",
"thread-b",
),
).toBe(historyB);
});
test("bridge places a rescued message with trusted seq inside the loaded window", () => {
const history = [msg("h1", "human", "q1", 1), msg("a1", "ai", "a1", 3)];
const rescued = [msg("mid", "ai", "rescued step", 2)];
const resolved = resolveTransientHistoryBridge(history, rescued);
expect(idsOf(resolved)).toEqual(["h1", "mid", "a1"]);
expect(seqsOf(resolved)).toEqual([1, 2, 3]);
});
test("insertByTrustedSeq keeps the tail for positions beyond every known seq", () => {
const base = [msg("h1", "human", "q", 1), msg("a1", "ai", "a", 2)];
const positioned = [msg("new", "ai", "late", 9)];
expect(idsOf(insertByTrustedSeq(base, positioned))).toEqual([
"h1",
"a1",
"new",
]);
expect(insertByTrustedSeq(base, [])).toBe(base);
});
test("dedupeMessagesByIdentity keeps the last visible copy per identity", () => {
const hidden = withKwargs(msg("x", "human", "hidden"), {
hide_from_ui: true,
});
const visible = msg("x", "human", "visible");
expect(dedupeMessagesByIdentity([hidden, visible])).toEqual([visible]);
});
test("a seq carried only by a hidden control copy still positions its visible twin", () => {
// Fallback path: no *visible* copy of the identity carries a seq, so the
// hidden control copy's valid seq is the only trustworthy position. Using
// it keeps the visible twin inside the skeleton instead of dropping it to
// the unpositioned tail; a visible copy with its own seq would win instead
// (covered by the hidden-control test above).
const hiddenControl = withKwargs(msg("m1", "system", "control", 2), {
hide_from_ui: true,
});
const history = [
msg("h0", "human", "q", 1),
hiddenControl,
msg("a3", "ai", "tail", 3),
];
const live = [msg("m1", "ai", "visible twin")];
const merged = mergeMessages(history, live, []);
expect(idsOf(merged)).toEqual(["h0", "m1", "a3"]);
expect(seqsOf(merged)).toEqual([1, 2, 3]);
expect(merged[1]!.content).toBe("visible twin");
});
test("live-only sequenced results anchor preceding unsequenced steps", () => {
const start = msg("start", "human", "start", 1);
const end = msg("end", "ai", "end", 9);
const step = msg("step", "ai", "step");
const result = msg("result", "ai", "result", 3);
expect(
idsOf(mergeMessages([start, end], [start, step, result, end], [])),
).toEqual(["start", "step", "result", "end"]);
});
test.each([true, false])(
"live-only sequenced results anchor trailing steps (shared identity: %s)",
(shared) => {
const start = msg("start", "human", "start", 1);
const end = msg("end", "ai", "end", 9);
const before = msg("before", "ai", "preceding step");
const result = msg("result", "ai", "result", 3);
const after = msg("after", "ai", "following step");
const last = msg("last", "ai", "last step");
const optimistic = msg("optimistic", "human", "next question");
const live = [...(shared ? [start] : []), before, result, after, last];
expect(idsOf(mergeMessages([start, end], live, [optimistic]))).toEqual([
"start",
"before",
"result",
"after",
"last",
"end",
"optimistic",
]);
},
);
test("a later shared anchor ends the live-only result's trailing segment", () => {
const start = msg("start", "human", "start", 1);
const result = msg("result", "ai", "result", 3);
const shared = msg("shared", "ai", "shared", 5);
const end = msg("end", "ai", "end", 9);
const after = msg("after", "ai", "new step");
expect(
idsOf(
mergeMessages([start, shared, end], [start, result, shared, after], []),
),
).toEqual(["start", "result", "shared", "end", "after"]);
});
test.each([2, 9])(
"bridge preserves an unrendered prefix before a rescued seq anchor (history ends at %s)",
(endSeq) => {
const start = msg("old-start", "human", "old question", 1);
const end = msg("old-end", "ai", "old answer", endSeq);
const step = msg("step", "ai", "not rendered before compaction");
const result = msg("result", "ai", "persisted result", 3);
// Compaction can capture these messages before React commits a frame,
// while canonical history still lacks both of them.
const resolved = resolveTransientHistoryBridge(
[start, end],
[step, result],
);
const expected =
endSeq < 3 ? [start, end, step, result] : [start, step, result, end];
expect(resolved).toEqual(expected);
expect(mergeMessages(resolved, [], [])).toEqual(expected);
// Once history catches up, canonical content wins without duplicates.
expect(resolveTransientHistoryBridge(expected, [step, result])).toBe(
expected,
);
},
);
test("a rescued seq anchor preserves following steps before the first loaded anchor", () => {
const start = msg("start", "human", "old question", 1);
const end = msg("end", "ai", "old answer", 9);
const before = msg("before", "ai", "preceding step");
const result = msg("result", "ai", "persisted result", 3);
const after = msg("after", "ai", "following step");
expect(
resolveTransientHistoryBridge([start, end], [before, result, after, end]),
).toEqual([start, before, result, after, end]);
});
test.each(["before", "after"] as const)(
"bridge keeps an unsequenced step %s its rescued sequenced neighbor",
(side) => {
const start = msg("start", "human", "start", 1);
const end = msg("end", "ai", "end", 9);
const step = msg("step", "ai", "step");
const result = msg("result", "ai", "result", 3);
const rescued = side === "before" ? [step, result] : [result, step];
const previous = [start, ...rescued, end];
const order = previous.map((message) => messageIdentity(message)!);
const resolved = resolveTransientHistoryBridge(
[start, end],
rescued,
order,
order,
);
expect(idsOf(resolved)).toEqual(idsOf(previous));
expect(idsOf(mergeMessages(resolved, [], []))).toEqual(idsOf(previous));
},
);