diff --git a/harness-tests/core/conversation-events.test.ts b/harness-tests/core/conversation-events.test.ts index 339559f43..595032b2f 100644 --- a/harness-tests/core/conversation-events.test.ts +++ b/harness-tests/core/conversation-events.test.ts @@ -1,3 +1,4 @@ +import { retainArchivedMessages } from '../../src/chat/core/conversation-snapshot' import { describe, expect, it } from 'vitest' import { StreamProcessor } from '@tanstack/ai' import { readMessageAttachments } from '../../src/chat/core/message-attachments' @@ -105,6 +106,7 @@ describe('conversation AG-UI projection', () => { role: 'user', parts: [{ type: 'text', content: 'A copied message' }], metadata: { gumInherited: true, obsolete: true }, + createdAt: new Date('2026-10-01T12:00:00Z'), }, ], } @@ -113,8 +115,14 @@ describe('conversation AG-UI projection', () => { const processor = new StreamProcessor() for (const event of conversationEvents(undefined, before, 'chat')) processor.processChunk(event) - for (const event of conversationEvents(before, after, 'chat')) + for (const event of conversationEvents(before, after, 'chat')) { processor.processChunk(event) + expect(processor.getMessages()).toHaveLength(1) + expect(processor.getMessages()[0].parts).toEqual(before.messages[0].parts) + expect(processor.getMessages()[0].createdAt).toEqual( + before.messages[0].createdAt, + ) + } expect(processor.getMessages()).toHaveLength(1) expect(processor.getMessages()[0]).toMatchObject({ metadata: { gumInherited: true }, @@ -238,6 +246,89 @@ describe('conversation AG-UI projection', () => { }) } }) + it('keeps displayed history when completed turns leave the live server window', () => { + const before: ProjectableConversation = { + ...initial(), + messages: [ + { + id: 'old', + role: 'user', + parts: [{ type: 'text', content: 'Earlier' }], + }, + { + id: 'current', + role: 'user', + parts: [{ type: 'text', content: 'Current' }], + }, + ], + } + const after: ProjectableConversation = { + ...before, + archivedTurns: 1, + messages: [ + before.messages[1], + { id: 'new', role: 'user', parts: [{ type: 'text', content: 'New' }] }, + ], + } + const processor = new StreamProcessor() + for (const event of conversationEvents(undefined, before, 'chat')) + processor.processChunk(event) + for (const event of conversationEvents(before, after, 'chat')) { + processor.processChunk(event) + expect(processor.getMessages()[0]).toMatchObject(before.messages[0]) + } + expect(processor.getMessages().map((message) => message.id)).toEqual([ + 'old', + 'current', + 'new', + ]) + }) + it('keeps archived rows across an authoritative metadata snapshot and clears on reset', () => { + const archived: ProjectableConversation = { + ...initial(), + archivedTurns: 1, + messages: [ + { + id: 'current', + role: 'user', + parts: [{ type: 'text', content: 'Current' }], + metadata: { obsolete: true }, + }, + ], + } + const displayed = [ + { + id: 'earlier', + role: 'user', + parts: [{ type: 'text', content: 'Earlier' }], + }, + ...archived.messages, + ] satisfies ProjectableConversation['messages'] + const next = structuredClone(archived) + next.messages[0].metadata = undefined + const processor = new StreamProcessor() + for (const event of conversationEvents( + undefined, + { ...archived, messages: displayed }, + 'chat', + )) + processor.processChunk(event) + for (const event of conversationEvents(archived, next, 'chat')) { + processor.processChunk( + retainArchivedMessages(event, processor.getMessages()), + ) + expect(processor.getMessages().map((message) => message.id)).toEqual([ + 'earlier', + 'current', + ]) + } + expect(processor.getMessages()[1].metadata).not.toHaveProperty('obsolete') + for (const event of conversationEvents(next, initial(), 'chat')) + processor.processChunk( + retainArchivedMessages(event, processor.getMessages()), + ) + expect(processor.getMessages()).toEqual([]) + }) it('publishes only appended text between saves and clears transcript on reset', () => { const old = { ...initial(), @@ -256,7 +347,7 @@ describe('conversation AG-UI projection', () => { messageId: 'a', delta: ' world', }) - expect(conversationEvents(next, initial(), 'chat')[0]).toEqual({ + expect(conversationEvents(next, initial(), 'chat')[0]).toMatchObject({ type: 'MESSAGES_SNAPSHOT', messages: [], }) diff --git a/src/chat/components/ConversationSession.tsx b/src/chat/components/ConversationSession.tsx index c4270ffc7..48edf94d8 100644 --- a/src/chat/components/ConversationSession.tsx +++ b/src/chat/components/ConversationSession.tsx @@ -1,3 +1,4 @@ +import { retainArchivedMessages } from '../core/conversation-snapshot' import { workspacePreviewsEnabled } from '../core/workspace-panels' import { useQueryClient } from '@tanstack/react-query' import { projectSavedEvent } from '../core/project-events' @@ -5,7 +6,6 @@ import { ProjectsPanel } from './ProjectsPanel' import { isPersonalAssistant } from '../core/bot-workspace' import './assistant-chat.css' import { DraftSyncNotice } from './DraftSyncNotice' -import { motion } from 'motion/react' import { useDebugDetails } from './useDebugDetails' import { Button } from './ui/Button' import { conversationActionRequest } from '../core/conversation-actions' @@ -461,6 +461,7 @@ export function ConversationSession({ const forkMessage = onFork ? (messageId: string) => onFork(messageId, destination.conversationId) : undefined + const displayedMessages = useRef(initialHistory.messages) const connection = useMemo(() => { const durable = durableStreamConnection({ sendUrl: apiUrl(`${path}/send`), @@ -468,8 +469,13 @@ export function ConversationSession({ initialOffset: initialHistory.streamOffset, emitSnapshotOnSubscribe: false, }) + const subscribe = + durable.subscribe as SubscribeConnectionAdapter['subscribe'] return { - subscribe: durable.subscribe as SubscribeConnectionAdapter['subscribe'], + subscribe: async function* (signal) { + for await (const chunk of subscribe(signal)) + yield retainArchivedMessages(chunk, displayedMessages.current) + }, send: async (messages, data) => { const message = messages.at(-1) if (!message || !('parts' in message)) @@ -525,6 +531,7 @@ export function ConversationSession({ } }, }) + displayedMessages.current = chat.messages const h = { ...metadata, messages: chat.messages } const emptyThreadState = useRef(false) emptyThreadState.current = @@ -1121,8 +1128,7 @@ export function ConversationSession({ origin={h.copyOrigin} /> )} - {thread && ( - @@ -1145,7 +1144,7 @@ export function ConversationSession({ {thread.source.text} {thread.source.truncated &&

Source excerpt

} - + )} {(!!h.archivedTurns || (!!search.message && !liveTarget)) && ( navigation.onSelectMessage(undefined)} > {turns.map((turn, index, turns) => ( @@ -1441,7 +1439,7 @@ export function ConversationSession({ ))} )} - +
diff --git a/src/chat/components/ConversationTurn.tsx b/src/chat/components/ConversationTurn.tsx index a5c1990c4..cbf9ed789 100644 --- a/src/chat/components/ConversationTurn.tsx +++ b/src/chat/components/ConversationTurn.tsx @@ -1,4 +1,3 @@ -import { motion } from 'motion/react' import type { UIMessage, ToolResultPart, ToolCallPart } from '@tanstack/ai' import { useEffect, useRef, useState, type ReactNode } from 'react' import { @@ -296,18 +295,7 @@ export function ConversationTurn({ )} {turn.prompt && ( - )} - + )}
{interim.map((m) => ( @@ -497,18 +485,7 @@ export function ConversationTurn({ finalParts.some( (part) => part.type === 'text' && part.content.trim(), ) && ( - {final.id !== responseEnd?.id && renderMessageThreads?.(final.id, { canCreate: false })} - + )} {showResponseActions && (
void onJumpToBottom?: () => void jumpToBottomContainer?: HTMLDivElement | null - followRequest?: string contentVersion?: string | number messageScope?: string }) { @@ -114,23 +112,6 @@ export function VirtualMessages({ } }, [rows.length, scrollElement, scrollMargin, virtualizer]) - const followedRequest = useRef(undefined) - useLayoutEffect(() => { - if ( - !followRequest || - followedRequest.current === followRequest || - !scrollElement || - scrollMargin === null || - !rows.length - ) - return - followedRequest.current = followRequest - pendingTarget.current = undefined - pendingFocus.current?.cancel() - pendingFocus.current = null - virtualizer.scrollToEnd() - }, [followRequest, scrollElement, scrollMargin, rows.length, virtualizer]) - const alignTarget = () => { if (!pendingTarget.current || !scrollElement) return if (pendingFocus.current && !pendingFocus.current.active()) { diff --git a/src/chat/core/conversation-snapshot.ts b/src/chat/core/conversation-snapshot.ts new file mode 100644 index 000000000..076e4d17a --- /dev/null +++ b/src/chat/core/conversation-snapshot.ts @@ -0,0 +1,26 @@ +import type { StreamChunk, UIMessage } from '@tanstack/ai' + +// The server window omits archived turns, but an open transcript keeps them. +// A reset has no boundary and still replaces the entire displayed history. +export function retainArchivedMessages( + chunk: StreamChunk, + displayed: UIMessage[], +): StreamChunk { + if (chunk.type !== 'MESSAGES_SNAPSHOT') return chunk + const boundary = chunk.rawEvent?.retainedHistoryBefore + if (typeof boundary !== 'string') return chunk + const index = displayed.findIndex((message) => message.id === boundary) + if (index < 1) return chunk + return { + ...chunk, + messages: [ + ...displayed.slice(0, index).map((message) => ({ + ...message, + content: message.parts + .flatMap((part) => (part.type === 'text' ? [part.content] : [])) + .join(''), + })), + ...chunk.messages, + ], + } +} diff --git a/src/chat/server/conversation-events.ts b/src/chat/server/conversation-events.ts index b2beca0da..349f0c32e 100644 --- a/src/chat/server/conversation-events.ts +++ b/src/chat/server/conversation-events.ts @@ -6,6 +6,7 @@ export interface ProjectableConversation { messages: UIMessage[] status: string activeRun: string | null + archivedTurns?: number error?: string } @@ -31,19 +32,34 @@ export function conversationEvents( !previous || previous.messages.some((m) => { const message = nextById.get(m.id) - // AG-UI metadata deltas merge keys. Replay is required when authoritative - // metadata changes so removed fields cannot remain on an existing message. + // Replace authoritative metadata atomically so removed fields cannot remain + // on an existing message, without clearing and replaying the transcript. return ( - !message || - JSON.stringify(m.metadata) !== JSON.stringify(message.metadata) + (!message && + (next.archivedTurns ?? 0) <= (previous.archivedTurns ?? 0)) || + (message && + JSON.stringify(m.metadata) !== JSON.stringify(message.metadata)) ) }) if (reset) { - // Rebuild history using the same events as live delivery, including tools. - events.push({ type: EventType.MESSAGES_SNAPSHOT, messages: [] }) + // The native processor preserves UI parts and timestamps in snapshots. + events.push({ + type: EventType.MESSAGES_SNAPSHOT, + rawEvent: { + retainedHistoryBefore: + (next.archivedTurns ?? 0) > 0 ? next.messages[0]?.id : undefined, + }, + messages: next.messages.map((message) => ({ + ...message, + metadata: message.metadata ?? {}, + content: message.parts + .flatMap((part) => (part.type === 'text' ? [part.content] : [])) + .join(''), + })), + }) } const before = new Map((reset ? [] : previous.messages).map((m) => [m.id, m])) - for (const message of next.messages) { + for (const message of reset ? [] : next.messages) { const old = before.get(message.id) let textChanged = false if (!old) @@ -163,6 +179,19 @@ export function conversationEvents( }, }) if (previous?.status === 'running' && next.status !== 'running') { + // Error events need an existing assistant anchor after a snapshot clears + // transient processor state. No text is replayed into that message. + if (next.status === 'error') { + const assistant = [...next.messages] + .reverse() + .find((message) => message.role === 'assistant') + if (assistant) + events.push({ + type: EventType.TEXT_MESSAGE_START, + messageId: assistant.id, + role: assistant.role, + }) + } events.push( next.status === 'error' ? {