diff --git a/client/src/components/Chat/Messages/Content/ContentParts.tsx b/client/src/components/Chat/Messages/Content/ContentParts.tsx index 2324486d00..2133d60a7c 100644 --- a/client/src/components/Chat/Messages/Content/ContentParts.tsx +++ b/client/src/components/Chat/Messages/Content/ContentParts.tsx @@ -8,7 +8,12 @@ import type { } from 'librechat-data-provider'; import type { ReactNode, ReactElement } from 'react'; import type { ToolCallGroupExpansionState } from './ToolCallGroup'; -import { mapAttachments, filterAttachmentsForPart, groupSequentialToolCalls } from '~/utils'; +import { + mapAttachments, + getPartKeyIndex, + filterAttachmentsForPart, + groupSequentialToolCalls, +} from '~/utils'; import WorkspaceChanges, { partitionWorkspaceChanges } from './Parts/WorkspaceChanges'; import { groupActivityPhases, lastCursorContentIdx } from '~/utils/activityLabels'; import { ParallelContentRenderer, type PartWithIndex } from './ParallelContent'; @@ -51,20 +56,20 @@ const getToolGroupId = (parts: PartWithIndex[], fallbackScope: number): string = * absorbs the block's leading THINK part when its text lands, so keying on * `parts[0]` would flip the key mid-run — remounting the group and losing * whatever the user had expanded. The tool calls themselves do not move. */ - let firstToolIdx: number | undefined; + let firstToolKeyIdx: number | undefined; for (const { part, idx } of parts) { const toolCallId = getToolCallId(part); if (toolCallId) { return `tool:${toolCallId}`; } - if (firstToolIdx === undefined && part?.type === ContentTypes.TOOL_CALL) { - firstToolIdx = idx; + if (firstToolKeyIdx === undefined && part?.type === ContentTypes.TOOL_CALL) { + firstToolKeyIdx = getPartKeyIndex(part, idx); } } /** Same reasoning for id-less tool calls: anchor to the first TOOL entry's * index rather than the block's first part, which shifts when reasoning is * absorbed. Only a block with no tool call at all falls back to `parts[0]`. */ - return `fallback:${fallbackScope}:${firstToolIdx ?? firstPart.idx}`; + return `fallback:${fallbackScope}:${firstToolKeyIdx ?? getPartKeyIndex(firstPart.part, firstPart.idx)}`; }; type PartWithContextProps = { @@ -117,7 +122,7 @@ const PartWithContext = memo(function PartWithContext({ part={part} attachments={partAttachments} isSubmitting={isSubmitting} - key={`part-${messageId}-${idx}`} + key={`part-${messageId}-${getPartKeyIndex(part, idx)}`} isCreatedByUser={isCreatedByUser} isLast={isLastPart} showCursor={isLastPart && isLast} @@ -248,7 +253,7 @@ const ContentPartsBody = memo(function ContentPartsBody({ const indices = new Set(); for (const segment of phaseSegments ?? []) { if (segment.type === 'phase') { - indices.add(segment.labelIndex); + indices.add(getPartKeyIndex(segment.labelPart, segment.labelIndex)); } } return indices; @@ -332,7 +337,7 @@ const ContentPartsBody = memo(function ContentPartsBody({ const localIdx = localIndexByAbsolute?.get(idx) ?? idx - contentIndexOffset; return ( { + (idx: number, keyIdx: number = idx): ReactElement | null => { if (authorHeader == null || !postSteerAuthors.has(idx)) { return null; } const activeAgentId = postSteerAuthors.get(idx); if (activeAgentId != null) { - return ; + return ; } - return {authorHeader}; + return {authorHeader}; }, [authorHeader, postSteerAuthors, messageId], ); @@ -500,6 +505,23 @@ const ContentPartsBody = memo(function ContentPartsBody({ const relativeGlobalLastContentIdx = lastCursorContentIdx(content ?? []); const globalLastContentIdx = relativeGlobalLastContentIdx < 0 ? -1 : absoluteIndexAt(relativeGlobalLastContentIdx); + /** Segment keys anchor to their first defined part's stable index, never + * to the segment's ordinal: hole-only slots form phantom segments while + * a run streams and vanish from the compacted final content, so ordinal + * keys shift at settle and remount every segment body after them. */ + const segmentKeyIndex = (segment: { + content: Array; + contentIndices: number[]; + startIndex: number; + }): number => { + for (let i = 0; i < segment.content.length; i++) { + const part = segment.content[i]; + if (part != null) { + return getPartKeyIndex(part, absoluteIndexAt(segment.contentIndices[i])); + } + } + return absoluteIndexAt(segment.startIndex); + }; const renderSegment = ( segmentContent: Array, segmentStartIndex: number, @@ -538,17 +560,26 @@ const ContentPartsBody = memo(function ContentPartsBody({ )} {renderPendingSkills()} - {phaseSegments.map((segment, index) => - segment.type === 'phase' ? ( + {phaseSegments.map((segment) => { + if (segment.type !== 'phase') { + return renderSegment( + segment.content, + absoluteIndexAt(segment.startIndex), + segment.contentIndices.map(absoluteIndexAt), + `phase-adjacent-${segmentKeyIndex(segment)}`, + ); + } + const phaseKeyIndex = getPartKeyIndex(segment.labelPart, segment.labelIndex); + return ( part != null && hasPendingApprovalInPart(part), )} animateEntrance={ - previousPhaseIndices != null && !previousPhaseIndices.has(segment.labelIndex) + previousPhaseIndices != null && !previousPhaseIndices.has(phaseKeyIndex) } showCursor={ isLast && @@ -560,18 +591,11 @@ const ContentPartsBody = memo(function ContentPartsBody({ segment.content, absoluteIndexAt(segment.startIndex), segment.contentIndices.map(absoluteIndexAt), - `phase-content-${index}`, + `phase-content-${phaseKeyIndex}`, )} - ) : ( - renderSegment( - segment.content, - absoluteIndexAt(segment.startIndex), - segment.contentIndices.map(absoluteIndexAt), - `phase-adjacent-${index}`, - ) - ), - )} + ); + })} @@ -639,9 +663,13 @@ const ContentPartsBody = memo(function ContentPartsBody({ )} {!showEmptyCursor && groupedParts.flatMap((group) => { - const firstIdx = group.type === 'single' ? group.part.idx : (group.parts[0]?.idx ?? -1); + const first = group.type === 'single' ? group.part : group.parts[0]; + const firstIdx = first?.idx ?? -1; const nodes: ReactElement[] = []; - const attribution = renderResumeAttribution(firstIdx); + const attribution = renderResumeAttribution( + firstIdx, + first ? getPartKeyIndex(first.part, first.idx) : firstIdx, + ); if (attribution != null) { nodes.push(attribution); } diff --git a/client/src/components/Chat/Messages/Content/ParallelContent.tsx b/client/src/components/Chat/Messages/Content/ParallelContent.tsx index 6348e25870..27b961a3df 100644 --- a/client/src/components/Chat/Messages/Content/ParallelContent.tsx +++ b/client/src/components/Chat/Messages/Content/ParallelContent.tsx @@ -8,11 +8,11 @@ import { } from '~/utils/activityLabels'; import MemoryArtifacts from './MemoryArtifacts'; import Sources from '~/components/Web/Sources'; +import { cn, getPartKeyIndex } from '~/utils'; import { SearchContext } from '~/Providers'; import SiblingHeader from './SiblingHeader'; import { EmptyText } from './Parts'; import Container from './Container'; -import { cn } from '~/utils'; export type PartWithIndex = { part: TMessageContentParts; idx: number }; @@ -235,7 +235,7 @@ type ParallelContentRendererProps = { * sequential before/after stretches consult it: column content already * carries per-agent identity. */ - renderResumeAttribution?: (idx: number) => React.ReactNode; + renderResumeAttribution?: (idx: number, keyIdx?: number) => React.ReactNode; showDecorations?: boolean; /** Absolute transcript index represented by `content[0]` in a phase slice. */ contentIndexOffset?: number; @@ -302,7 +302,7 @@ export const ParallelContentRenderer = memo(function ParallelContentRenderer({ {/* Sequential content BEFORE parallel sections */} {before.flatMap(({ part, idx }) => { - const attribution = renderResumeAttribution?.(idx); + const attribution = renderResumeAttribution?.(idx, getPartKeyIndex(part, idx)); const rendered = renderPart(part, idx, false); return attribution != null ? [attribution, rendered] : [rendered]; })} @@ -324,7 +324,7 @@ export const ParallelContentRenderer = memo(function ParallelContentRenderer({ {/* Sequential content AFTER parallel sections */} {after.flatMap(({ part, idx }) => { - const attribution = renderResumeAttribution?.(idx); + const attribution = renderResumeAttribution?.(idx, getPartKeyIndex(part, idx)); const rendered = renderPart(part, idx, idx === lastContentIdx); return attribution != null ? [attribution, rendered] : [rendered]; })} diff --git a/client/src/components/Chat/Messages/Content/__tests__/ContentParts.test.tsx b/client/src/components/Chat/Messages/Content/__tests__/ContentParts.test.tsx index 33191b8ff9..0958825d8e 100644 --- a/client/src/components/Chat/Messages/Content/__tests__/ContentParts.test.tsx +++ b/client/src/components/Chat/Messages/Content/__tests__/ContentParts.test.tsx @@ -2,6 +2,7 @@ import React from 'react'; import { ContentTypes, Tools } from 'librechat-data-provider'; import { fireEvent, render, screen } from '@testing-library/react'; import type { TMessageContentParts, TAttachment } from 'librechat-data-provider'; +import { preserveStreamedContentIdentity } from '~/utils/messages'; import { groupSequentialToolCalls } from '~/utils'; jest.mock('~/utils', () => ({ @@ -10,6 +11,7 @@ jest.mock('~/utils', () => ({ filterAttachmentsForPart: (attachments: unknown) => attachments, groupSequentialToolCalls: jest.fn(), hasPendingApprovalInPart: jest.requireActual('~/utils/groupToolCalls').hasPendingApprovalInPart, + getPartKeyIndex: jest.requireActual('~/utils/messages').getPartKeyIndex, })); jest.mock('~/Providers', () => { @@ -651,3 +653,69 @@ describe('ContentParts — activity phase state', () => { ); }); }); + +describe('ContentParts — settled content identity across compaction', () => { + /** Mirrors a captured run: the aggregator leaves holes at the source indexes + * of steps that produced nothing, and `finalHandler` swaps in the server's + * compacted array. Without the streamed-index stamp every index-derived key + * shifts and the settled message remounts wholesale. */ + const toolPart = { + type: ContentTypes.TOOL_CALL, + [ContentTypes.TOOL_CALL]: { id: 'call_a', name: 'search', args: {}, output: 'one' }, + } as unknown as TMessageContentParts; + const batchLabel = { + type: ContentTypes.ACTIVITY_LABEL, + [ContentTypes.ACTIVITY_LABEL]: 'Recorded the fact', + tool_call_ids: ['call_a'], + } as unknown as TMessageContentParts; + const answer = { type: ContentTypes.TEXT, text: 'done' } as unknown as TMessageContentParts; + const phaseLabel = (bounds: { start: number; end: number }) => + ({ + type: ContentTypes.ACTIVITY_LABEL, + [ContentTypes.ACTIVITY_LABEL]: 'Researched the question', + activity_label_type: 'phase', + activity_start_index: bounds.start, + activity_end_index: bounds.end, + activity_count: 1, + pending: false, + }) as unknown as TMessageContentParts; + + const streamed: Array = [ + undefined, + toolPart, + batchLabel, + undefined, + answer, + phaseLabel({ start: 1, end: 4 }), + ]; + const compacted = [toolPart, batchLabel, answer, phaseLabel({ start: 0, end: 2 })]; + + const renderStreaming = () => + render(); + + it('keeps every part and the phase group mounted when the final content is stamped', () => { + const { rerender } = renderStreaming(); + const phaseNode = screen.getByTestId('activity-phase-group'); + const toolNode = screen.getByTestId('real-part-tool_call'); + const textNode = screen.getByTestId('real-part-text'); + + const finalContent = preserveStreamedContentIdentity(streamed, compacted); + rerender(); + + expect(screen.getByTestId('activity-phase-group')).toBe(phaseNode); + expect(screen.getByTestId('real-part-tool_call')).toBe(toolNode); + expect(screen.getByTestId('real-part-text')).toBe(textNode); + expect(phaseNode).toHaveAttribute('data-animate-entrance', 'false'); + }); + + it('remounts and replays the phase entrance without the stamp (regression control)', () => { + const { rerender } = renderStreaming(); + const phaseNode = screen.getByTestId('activity-phase-group'); + + rerender(); + + const settledPhase = screen.getByTestId('activity-phase-group'); + expect(settledPhase).not.toBe(phaseNode); + expect(settledPhase).toHaveAttribute('data-animate-entrance', 'true'); + }); +}); diff --git a/client/src/hooks/Chat/useChatFunctions.ts b/client/src/hooks/Chat/useChatFunctions.ts index 195d23a557..cb2bc929ad 100644 --- a/client/src/hooks/Chat/useChatFunctions.ts +++ b/client/src/hooks/Chat/useChatFunctions.ts @@ -33,6 +33,7 @@ import { isSubmittableMessage, createDualMessageContent, getRouteChatProjectId, + stripStreamedIndexStamps, } from '~/utils'; import useFocusRegeneratedResponse from '~/hooks/Chat/useFocusRegeneratedResponse'; import useSetFilesToDelete from '~/hooks/Files/useSetFilesToDelete'; @@ -626,7 +627,10 @@ export default function useChatFunctions({ initialResponse.text = ''; if (editedContent && latestMessage?.content) { - initialResponse.content = cloneDeep(latestMessage.content); + /** Stamps off: the rerun appends provider parts at the prefix LENGTH, + * and a retained `streamedIndex` at or above it would collide with an + * appended part's render key (see `stripStreamedIndexStamps`). */ + initialResponse.content = stripStreamedIndexStamps(cloneDeep(latestMessage.content)); /** Captured now, while it is still the retained prefix: a later resume * sync replaces this array with the server's completion-local * snapshot, after which its length no longer describes the offset. */ diff --git a/client/src/hooks/SSE/useEventHandlers.ts b/client/src/hooks/SSE/useEventHandlers.ts index 1550c24e06..de4b0d1d13 100644 --- a/client/src/hooks/SSE/useEventHandlers.ts +++ b/client/src/hooks/SSE/useEventHandlers.ts @@ -36,6 +36,7 @@ import { updateConvoInAllQueries, removeConvoFromAllQueries, findConversationInInfinite, + preserveStreamedContentIdentity, } from '~/utils'; import { startupConfigKey, @@ -870,19 +871,29 @@ export default function useEventHandlers({ finalMessages = [...messages, requestMessage, responseMessage]; } - /* Preserve files from current messages when server response lacks them */ + /* Preserve files and streamed content identity from current messages: + * files fill in when the server response lacks them, and the persisted + * (compacted) content is stamped with the indexes it streamed at so + * index-keyed renders don't remount the settled message. */ if (finalMessages.length > 0) { - const currentMsgMap = new Map( - currentMessages - .filter((m) => m.files && m.files.length > 0) - .map((m) => [m.messageId, m.files]), - ); + const currentMsgMap = new Map(currentMessages.map((m) => [m.messageId, m])); for (let i = 0; i < finalMessages.length; i++) { const msg = finalMessages[i]; - const preservedFiles = currentMsgMap.get(msg.messageId); - if (msg.files == null && preservedFiles) { - finalMessages[i] = { ...msg, files: preservedFiles }; + const currentMsg = currentMsgMap.get(msg.messageId); + if (!currentMsg) { + continue; } + const preservedFiles = + msg.files == null && currentMsg.files?.length ? currentMsg.files : undefined; + const content = preserveStreamedContentIdentity(currentMsg.content, msg.content); + if (preservedFiles == null && content === msg.content) { + continue; + } + finalMessages[i] = { + ...msg, + ...(preservedFiles != null ? { files: preservedFiles } : {}), + ...(content !== msg.content ? { content } : {}), + }; } } diff --git a/client/src/utils/messages.spec.ts b/client/src/utils/messages.spec.ts new file mode 100644 index 0000000000..b06f3e5a48 --- /dev/null +++ b/client/src/utils/messages.spec.ts @@ -0,0 +1,242 @@ +import { ContentTypes } from 'librechat-data-provider'; +import type { TMessage, TMessageContentParts } from 'librechat-data-provider'; +import { preserveStreamedContentIdentity, stripStreamedIndexStamps } from './messages'; + +const text = (value: string, extra: Record = {}): TMessageContentParts => + ({ type: ContentTypes.TEXT, text: value, ...extra }) as TMessageContentParts; + +const think = (value: string): TMessageContentParts => + ({ type: ContentTypes.THINK, think: value }) as TMessageContentParts; + +const tool = (id: string | undefined, name = 'search'): TMessageContentParts => + ({ + type: ContentTypes.TOOL_CALL, + tool_call: { id, name, args: '' }, + }) as TMessageContentParts; + +const label = (value: string, extra: Record = {}): TMessageContentParts => + ({ type: ContentTypes.ACTIVITY_LABEL, activity_label: value, ...extra }) as TMessageContentParts; + +const streamedIndexes = (content: TMessage['content']): Array => + (content ?? []).map((part) => part?.streamedIndex); + +describe('preserveStreamedContentIdentity', () => { + it('stamps every part shifted by compacted holes with its streamed index', () => { + const streamed = [ + undefined, + tool('call_a'), + label('first'), + undefined, + tool('call_b'), + label('second'), + text('answer'), + label('phase', { activity_label_type: 'phase' }), + ]; + const final = [ + tool('call_a'), + label('first'), + tool('call_b'), + label('second'), + text('answer'), + label('phase', { activity_label_type: 'phase' }), + ]; + + const result = preserveStreamedContentIdentity(streamed, final); + + expect(streamedIndexes(result)).toEqual([1, 2, 4, 5, 6, 7]); + expect(final.every((part) => part.streamedIndex === undefined)).toBe(true); + }); + + it('returns the final array untouched when no hole shifted anything', () => { + const streamed = [tool('call_a'), text('answer')]; + const final = [tool('call_a'), text('answer')]; + + expect(preserveStreamedContentIdentity(streamed, final)).toBe(final); + }); + + it('leaves aligned prefix parts unstamped while stamping the shifted tail', () => { + const streamed = [text('intro'), undefined, tool('call_a')]; + const final = [text('intro'), tool('call_a')]; + + expect(streamedIndexes(preserveStreamedContentIdentity(streamed, final))).toEqual([ + undefined, + 2, + ]); + }); + + it('skips streamed empty-text placeholders the compaction dropped', () => { + const streamed = [text(''), tool('call_a'), text('answer')]; + const final = [tool('call_a'), text('answer')]; + + expect(streamedIndexes(preserveStreamedContentIdentity(streamed, final))).toEqual([1, 2]); + }); + + it('skips streamed empty think parts and typeless placeholders', () => { + const streamed = [ + { type: '' } as unknown as TMessageContentParts, + think(''), + think('reasoned'), + text('answer'), + ]; + const final = [think('reasoned'), text('answer')]; + + expect(streamedIndexes(preserveStreamedContentIdentity(streamed, final))).toEqual([2, 3]); + }); + + it('matches by identity, not equality: richer final text keeps its streamed slot', () => { + const streamed = [undefined, text('partial ans')]; + const final = [text('partial answer, completed.')]; + + expect(streamedIndexes(preserveStreamedContentIdentity(streamed, final))).toEqual([1]); + }); + + it('pairs tool calls by id and abandons stamping on an id mismatch', () => { + const streamed = [undefined, tool('call_a')]; + const final = [tool('call_other')]; + + expect(preserveStreamedContentIdentity(streamed, final)).toBe(final); + }); + + it('abandons stamping when the server appended a part that never streamed', () => { + const streamed = [undefined, tool('call_a')]; + const final = [tool('call_a'), text('server-added')]; + + expect(preserveStreamedContentIdentity(streamed, final)).toBe(final); + }); + + it('abandons stamping on a type mismatch instead of mispairing', () => { + const streamed = [think('reasoned'), text('answer')]; + const final = [text('answer')]; + + expect(preserveStreamedContentIdentity(streamed, final)).toBe(final); + }); + + it('returns final content untouched when nothing streamed', () => { + const final = [text('answer')]; + + expect(preserveStreamedContentIdentity(undefined, final)).toBe(final); + expect(preserveStreamedContentIdentity([], final)).toBe(final); + }); + + it('abandons stamping when a filtered run retains only a same-type later part', () => { + const streamed = [text('intermediate agent output'), text('final agent answer')]; + const final = [text('final agent answer')]; + + expect(preserveStreamedContentIdentity(streamed, final)).toBe(final); + }); + + it('abandons stamping when an omitted intermediate is a prefix of the retained output', () => { + const streamed = [text('Answer:'), text('Answer: final details')]; + const final = [text('Answer: final details')]; + + expect(preserveStreamedContentIdentity(streamed, final)).toBe(final); + }); + + it('ignores trailing holes and empty slots when checking for removed content', () => { + const streamed = [undefined, tool('call_a'), text('answer'), text(''), undefined]; + const final = [tool('call_a'), text('answer')]; + + expect(streamedIndexes(preserveStreamedContentIdentity(streamed, final))).toEqual([1, 2]); + }); + + it('abandons stamping when streamed and final text diverge', () => { + const streamed = [undefined, text('answer A')]; + const final = [text('answer B')]; + + expect(preserveStreamedContentIdentity(streamed, final)).toBe(final); + }); + + it('never pairs a batch label with a phase label of the same text', () => { + const streamed = [undefined, label('Ran the tools')]; + const final = [label('Ran the tools', { activity_label_type: 'phase' })]; + + expect(preserveStreamedContentIdentity(streamed, final)).toBe(final); + }); + + it('pairs a blank label reservation with its filled final label', () => { + const streamed = [undefined, label('')]; + const final = [label('Recorded the fact')]; + + expect(streamedIndexes(preserveStreamedContentIdentity(streamed, final))).toEqual([1]); + }); + + it('never pairs text parts across different phases', () => { + const streamed = [undefined, text('note', { phase: 'commentary' })]; + const final = [text('note')]; + + expect(preserveStreamedContentIdentity(streamed, final)).toBe(final); + }); + + it('carries existing stamps forward when a settled message is re-delivered compact', () => { + const streamedSparse = [undefined, tool('call_a'), label('first'), undefined, text('answer')]; + const settled = preserveStreamedContentIdentity(streamedSparse, [ + tool('call_a'), + label('first'), + text('answer'), + ]); + expect(streamedIndexes(settled)).toEqual([1, 2, 4]); + + const redelivered = [tool('call_a'), label('first'), text('answer')]; + const result = preserveStreamedContentIdentity(settled, redelivered); + + expect(streamedIndexes(result)).toEqual([1, 2, 4]); + }); + + it('carries a partially stamped message forward without stamping its aligned prefix', () => { + const streamedSparse = [text('intro'), undefined, tool('call_a')]; + const settled = preserveStreamedContentIdentity(streamedSparse, [ + text('intro'), + tool('call_a'), + ]); + expect(streamedIndexes(settled)).toEqual([undefined, 2]); + + const result = preserveStreamedContentIdentity(settled, [text('intro'), tool('call_a')]); + + expect(streamedIndexes(result)).toEqual([undefined, 2]); + }); + + it('pairs id-less tool calls by name', () => { + const streamed = [undefined, tool(undefined, 'execute_code')]; + const final = [tool(undefined, 'execute_code')]; + + expect(streamedIndexes(preserveStreamedContentIdentity(streamed, final))).toEqual([1]); + }); + + it('abandons stamping when id-less tool call names differ', () => { + const streamed = [undefined, tool(undefined, 'execute_code')]; + const final = [tool(undefined, 'web_search')]; + + expect(preserveStreamedContentIdentity(streamed, final)).toBe(final); + }); +}); + +describe('stripStreamedIndexStamps', () => { + const tool = (id: string): TMessageContentParts => + ({ + type: ContentTypes.TOOL_CALL, + tool_call: { id, name: 'search', args: '' }, + }) as TMessageContentParts; + + it('drops every stamp from a settled content array', () => { + const settled = preserveStreamedContentIdentity( + [ + undefined, + tool('call_a'), + { type: ContentTypes.TEXT, text: 'answer' } as TMessageContentParts, + ], + [tool('call_a'), { type: ContentTypes.TEXT, text: 'answer' } as TMessageContentParts], + ); + expect((settled ?? []).some((part) => part?.streamedIndex !== undefined)).toBe(true); + + const stripped = stripStreamedIndexStamps(settled); + + expect((stripped ?? []).every((part) => part?.streamedIndex === undefined)).toBe(true); + }); + + it('returns the same reference when nothing is stamped', () => { + const plain = [tool('call_a')]; + + expect(stripStreamedIndexStamps(plain)).toBe(plain); + expect(stripStreamedIndexStamps(undefined)).toBeUndefined(); + }); +}); diff --git a/client/src/utils/messages.ts b/client/src/utils/messages.ts index aaa4444b18..e8afaac986 100644 --- a/client/src/utils/messages.ts +++ b/client/src/utils/messages.ts @@ -9,6 +9,7 @@ import { encodeEphemeralAgentId, } from 'librechat-data-provider'; import type { + Agents, TMessage, TConversation, TEndpointsConfig, @@ -192,6 +193,204 @@ export const getAllContentText = (message?: TMessage | null): string => { return ''; }; +const getPartTextValue = (value?: string | { value?: string }): string => + (typeof value === 'string' ? value : value?.value) ?? ''; + +const getPartToolCall = (part: TMessageContentParts): Agents.ToolCall | undefined => + part.type === ContentTypes.TOOL_CALL + ? (part[ContentTypes.TOOL_CALL] as Agents.ToolCall | undefined) + : undefined; + +/** Slots the persistence compaction leaves nothing behind for: the + * dual-message `type: ''` placeholders, text/think parts that never received a + * delta, and tool calls missing their `tool_call` payload. */ +const isEmptyContentPart = (part: TMessageContentParts): boolean => { + if (!part.type) { + return true; + } + if (part.type === ContentTypes.TEXT) { + return getPartTextValue(part.text).length === 0; + } + if (part.type === ContentTypes.THINK) { + return getPartTextValue(part.think).length === 0; + } + if (part.type === ContentTypes.TOOL_CALL) { + return getPartToolCall(part) == null; + } + return false; +}; + +/** One side extending the other is the same part observed at two moments — + * a flushed tail or a server-side trim — while divergent content is a + * different part that merely shares the type. */ +const isMutualPrefix = (streamed: string, final: string): boolean => + final.startsWith(streamed) || streamed.startsWith(final); + +/** Identity match, not equality: the persisted part may carry richer content + * (flushed text, tool output) than its streamed counterpart, and updating a + * kept identity in place is exactly the point. Content still has to agree as + * an extension of what streamed: a filtered run (`hide_sequential_outputs`) + * omits intermediate parts from the final array, and a type-only match would + * hand the retained output an omitted intermediate's identity. */ +const isSameStreamedPart = ( + streamed: TMessageContentParts, + final: TMessageContentParts, +): boolean => { + if (streamed.type !== final.type) { + return false; + } + if (streamed.type === ContentTypes.TOOL_CALL) { + const streamedCall = getPartToolCall(streamed); + const finalCall = getPartToolCall(final); + if (streamedCall?.id != null && finalCall?.id != null) { + return streamedCall.id === finalCall.id; + } + if (streamedCall?.name != null && finalCall?.name != null) { + return streamedCall.name === finalCall.name; + } + return true; + } + if (streamed.type === ContentTypes.TEXT && final.type === ContentTypes.TEXT) { + if ((streamed.phase ?? null) !== (final.phase ?? null)) { + return false; + } + return isMutualPrefix(getPartTextValue(streamed.text), getPartTextValue(final.text)); + } + if (streamed.type === ContentTypes.THINK && final.type === ContentTypes.THINK) { + return isMutualPrefix(getPartTextValue(streamed.think), getPartTextValue(final.think)); + } + if (streamed.type === ContentTypes.ACTIVITY_LABEL && final.type === ContentTypes.ACTIVITY_LABEL) { + if ((streamed.activity_label_type ?? null) !== (final.activity_label_type ?? null)) { + return false; + } + return isMutualPrefix( + getPartTextValue(streamed.activity_label), + getPartTextValue(final.activity_label), + ); + } + return true; +}; + +/** + * Stamps each part of a final (persisted, compacted) content array with the + * index it occupied while it streamed, pairing the two arrays in order. + * + * The aggregator writes parts at provider-source indexes, so the streamed + * array is sparse wherever a step produced nothing; persistence compacts the + * holes away and every later part shifts down. Adopting the compacted array + * verbatim re-keys every index-derived React identity at the final event — + * the settled message remounts wholesale, entrance animations replay, and the + * thread visibly jumps. The stamp (`streamedIndex`) lets renderers keep the + * streamed key while all coordinate logic uses the compacted positions the + * server persisted. + * + * Pairing is all-or-nothing: a partially stamped array could collide a + * streamed key with a compacted fallback key. When any final part has no + * streamed counterpart (server-enriched content), or any substantial streamed + * part has no final counterpart (a filtered run that dropped intermediate + * outputs — where in-order pairing could hand a retained part an omitted + * part's identity), the final array is returned untouched and the message + * re-keys as before. + */ +export const preserveStreamedContentIdentity = ( + streamedContent: Array | undefined, + finalContent: TMessage['content'], +): TMessage['content'] => { + if (!streamedContent?.length || !finalContent?.length) { + return finalContent; + } + + let cursor = 0; + let stamped: TMessageContentParts[] | null = null; + for (let index = 0; index < finalContent.length; index++) { + const finalPart = finalContent[index] as TMessageContentParts | undefined; + if (finalPart == null) { + return finalContent; + } + let matchedIndex = -1; + let matchedPart: TMessageContentParts | null = null; + while (cursor < streamedContent.length) { + const streamedPart = streamedContent[cursor]; + if (streamedPart == null) { + cursor += 1; + continue; + } + /** An empty streamed slot facing a filled final part was dropped by the + * compaction — never let it steal the match from the filled streamed + * part behind it (an empty THINK ahead of the real one, say). */ + if (isEmptyContentPart(streamedPart) && !isEmptyContentPart(finalPart)) { + cursor += 1; + continue; + } + if (isSameStreamedPart(streamedPart, finalPart)) { + matchedIndex = cursor; + matchedPart = streamedPart; + cursor += 1; + } + break; + } + if (matchedIndex === -1 || matchedPart == null) { + return finalContent; + } + /** A settled message can be re-delivered by a LATER final event (e.g. an + * Assistants run resyncing prior turns): both sides arrive compact, but + * the current parts already carry stamps from their own settle. Carrying + * them forward keeps their keys stable forever, instead of silently + * reverting the identity this stamp exists to preserve. */ + const stampIndex = matchedPart.streamedIndex ?? matchedIndex; + if (stampIndex !== index && stamped == null) { + stamped = [...finalContent]; + } + if (stamped != null && stampIndex !== index) { + stamped[index] = { ...finalPart, streamedIndex: stampIndex }; + } + } + /** Leftover substantial streamed parts mean the server REMOVED content + * (`hide_sequential_outputs`), so every pairing above is suspect — an + * omitted intermediate that happens to prefix the retained output would + * have claimed its identity. Only holes and empty slots may remain. */ + for (let rest = cursor; rest < streamedContent.length; rest++) { + const leftover = streamedContent[rest]; + if (leftover != null && !isEmptyContentPart(leftover)) { + return finalContent; + } + } + return stamped ?? finalContent; +}; + +/** + * Drops the client-only `streamedIndex` stamps from a content array. An + * edited resubmission retains the settled prefix and appends the rerun's + * parts at the prefix LENGTH — a stamp at or above that length would collide + * with an appended part's key — so the retained prefix reverts to physical + * identity for the rerun. Returns the input untouched when nothing is + * stamped. + */ +export function stripStreamedIndexStamps(content: TMessageContentParts[]): TMessageContentParts[]; +export function stripStreamedIndexStamps(content: TMessage['content']): TMessage['content']; +export function stripStreamedIndexStamps(content: TMessage['content']): TMessage['content'] { + if (!content?.length) { + return content; + } + let changed = false; + const next = content.map((part) => { + if (part == null || part.streamedIndex === undefined) { + return part; + } + changed = true; + const { streamedIndex: _streamedIndex, ...rest } = part; + return rest as TMessageContentParts; + }); + return changed ? next : content; +} + +/** Render-identity index for content-part keys: the streamed position stamped + * by the final handler survives the sparse→compact swap; everything else keys + * by the live index. Coordinate logic (edit indexes, phase bounds, cursor) + * must keep using the live index. */ +export const getPartKeyIndex = (part: TMessageContentParts | undefined, idx: number): number => + part?.streamedIndex ?? idx; + /** * Whether a draft message has enough content to submit: non-whitespace * text, or at least one attached file. Lets users send a file without diff --git a/packages/data-provider/src/types/assistants.ts b/packages/data-provider/src/types/assistants.ts index 97752e545b..9d309086f0 100644 --- a/packages/data-provider/src/types/assistants.ts +++ b/packages/data-provider/src/types/assistants.ts @@ -638,10 +638,20 @@ export type PartMetadata = { * as dispatch time rather than the task's runtime. */ backgrounded?: boolean; + /** + * Content index this part occupied while its run streamed. The aggregator + * writes parts at provider-source indexes, so the streamed array is sparse; + * persistence compacts it and every part after a hole shifts down. The + * client's final handler stamps the streamed position onto the compacted + * parts it adopts, so index-derived render identity survives the swap + * instead of remounting the settled message. Client-only and absent + * everywhere else — persisted content never carries it. + */ + streamedIndex?: number; }; /** Metadata for parallel content rendering - subset of PartMetadata */ -export type ContentMetadata = Pick; +export type ContentMetadata = Pick; export type ContentPart = ( | CodeToolCall