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