mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-08-29 05:20:49 +00:00
📌 fix: Keep the Settled Turn Mounted Through Final Content Compaction (#15186)
* 📌 fix: Keep the Settled Turn Mounted Through Final Content Compaction The agent aggregator writes content parts at provider-source indexes, so the streamed array is sparse wherever a step produced nothing; the final SSE event carries the persisted, compacted array. Adopting it verbatim shifted every part after a hole, re-keying every index-derived React identity: the settled message remounted wholesale, activity-phase groups replayed their fold-in entrance, code panes re-highlighted, and the thread visibly snapped up and down at the end of every tool-calling run. finalHandler now pairs the compacted parts with their streamed counterparts in order and stamps each with the index it streamed at (`streamedIndex`, client-only); render keys read the stamp while all coordinate logic (edit indexes, phase bounds, cursor) stays on the live compacted positions the server persisted. Phase-segment keys also anchor to their first defined part instead of the segment ordinal, since phantom hole-only segments vanish at compaction and shifted every segment after them. * 🔁 fix: Carry Identity Stamps Through Re-Delivered Finals and Parallel Attribution Codex round 1, both real: - P1: a later final event can re-deliver an already-settled message as a fresh compact array (Assistants runMessages resync); index-aligned pairing returned it unstamped, wiping the previous settle's stamps and re-keying the older turn all over again. The pairing now carries the matched current part's stamp forward, so a settled turn keeps its keys through every subsequent final. - P2: ParallelContentRenderer's sequential stretches invoked renderResumeAttribution with only the live index, so steer attribution nodes in parallel content still re-keyed at the swap. The stable key index now threads through both call sites; getPartKeyIndex moves to utils/messages beside the stamp writer it reads. * 🧿 fix: Require Content Agreement Before Pairing Streamed Identity Codex round 2 (P2, real): hide_sequential_outputs runs omit intermediate parts from the final array, so a type-only match could hand the retained output an omitted intermediate's identity — transferring its key and any UI state. Non-tool pairing now requires content agreement: mutual-prefix text for TEXT/THINK/ACTIVITY_LABEL (one side extending the other is the same part observed at two moments), the Open Responses phase for TEXT, and the label kind for activity labels — a blank reservation still pairs with its filled label. Ambiguous shapes fall back to the pre-stamp full re-key, which is honest for a final that visibly removes parts. * 🪢 fix: Refuse Stamping When the Server Removed Content; Strip Stamps on Edited Reruns Codex round 3, two of three real: - Prefix agreement alone still mis-paired when an omitted intermediate happened to prefix the retained output. Pairing now also requires that no substantial streamed part is left over: leftovers mean the server removed content (hide_sequential_outputs), so every in-order pairing is suspect and the message re-keys plainly instead. - An edited resubmission clones the settled (stamped) prefix and appends the rerun's parts at the prefix length; a retained stamp at or above that length collides with an appended part's key. The clone now strips the client-only stamps, reverting the retained prefix to physical identity for the rerun. The third finding (content-segment keys under late-phase recovery) is declined with rationale on the PR: user expansion overrides survive via the message-wide expansion map with stable group ids, recovery is a genuine restructure at the moment a phase materializes, and first-child anchoring is the only choice stable under the two high-frequency events (streaming appends and final compaction).
This commit is contained in:
Danny Avila
GitHub
parent
ac2aef00f6
commit
a9d99b3771
8 changed files with 605 additions and 43 deletions
|
|
@ -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<number>();
|
||||
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 (
|
||||
<PartWithContext
|
||||
key={`provider-${messageId}-${idx}`}
|
||||
key={`provider-${messageId}-${getPartKeyIndex(part, idx)}`}
|
||||
idx={idx}
|
||||
part={part}
|
||||
isLast={isLast}
|
||||
|
|
@ -369,7 +374,7 @@ const ContentPartsBody = memo(function ContentPartsBody({
|
|||
const localIdx = localIndexByAbsolute?.get(idx) ?? idx - contentIndexOffset;
|
||||
return (
|
||||
<PartWithContext
|
||||
key={`provider-${messageId}-${idx}`}
|
||||
key={`provider-${messageId}-${getPartKeyIndex(part, idx)}`}
|
||||
idx={idx}
|
||||
part={part}
|
||||
isLast={isLast}
|
||||
|
|
@ -456,15 +461,15 @@ const ContentPartsBody = memo(function ContentPartsBody({
|
|||
/** The re-attribution node for a part resuming after a steer block, shared
|
||||
* by the sequential path and the parallel renderer's sequential stretches. */
|
||||
const renderResumeAttribution = useCallback(
|
||||
(idx: number): ReactElement | null => {
|
||||
(idx: number, keyIdx: number = idx): ReactElement | null => {
|
||||
if (authorHeader == null || !postSteerAuthors.has(idx)) {
|
||||
return null;
|
||||
}
|
||||
const activeAgentId = postSteerAuthors.get(idx);
|
||||
if (activeAgentId != null) {
|
||||
return <AgentUpdate key={`author-${messageId}-${idx}`} currentAgentId={activeAgentId} />;
|
||||
return <AgentUpdate key={`author-${messageId}-${keyIdx}`} currentAgentId={activeAgentId} />;
|
||||
}
|
||||
return <Fragment key={`author-${messageId}-${idx}`}>{authorHeader}</Fragment>;
|
||||
return <Fragment key={`author-${messageId}-${keyIdx}`}>{authorHeader}</Fragment>;
|
||||
},
|
||||
[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<TMessageContentParts | undefined>;
|
||||
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<TMessageContentParts | undefined>,
|
||||
segmentStartIndex: number,
|
||||
|
|
@ -538,17 +560,26 @@ const ContentPartsBody = memo(function ContentPartsBody({
|
|||
<Sources messageId={messageId} conversationId={conversationId || undefined} />
|
||||
)}
|
||||
{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 (
|
||||
<ActivityPhaseGroup
|
||||
key={`activity-phase-${messageId}-${segment.labelIndex}`}
|
||||
key={`activity-phase-${messageId}-${phaseKeyIndex}`}
|
||||
labelPart={segment.labelPart}
|
||||
hasContent={segment.hasContent}
|
||||
hasPendingApproval={segment.content.some(
|
||||
(part) => 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}`,
|
||||
)}
|
||||
</ActivityPhaseGroup>
|
||||
) : (
|
||||
renderSegment(
|
||||
segment.content,
|
||||
absoluteIndexAt(segment.startIndex),
|
||||
segment.contentIndices.map(absoluteIndexAt),
|
||||
`phase-adjacent-${index}`,
|
||||
)
|
||||
),
|
||||
)}
|
||||
);
|
||||
})}
|
||||
<WorkspaceChanges attachments={workspaceChanges} />
|
||||
</SearchContext.Provider>
|
||||
</ApprovalProvider>
|
||||
|
|
@ -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);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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];
|
||||
})}
|
||||
|
|
|
|||
|
|
@ -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<TMessageContentParts | undefined> = [
|
||||
undefined,
|
||||
toolPart,
|
||||
batchLabel,
|
||||
undefined,
|
||||
answer,
|
||||
phaseLabel({ start: 1, end: 4 }),
|
||||
];
|
||||
const compacted = [toolPart, batchLabel, answer, phaseLabel({ start: 0, end: 2 })];
|
||||
|
||||
const renderStreaming = () =>
|
||||
render(<ContentParts {...baseProps} content={streamed} isLast isSubmitting isLatestMessage />);
|
||||
|
||||
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(<ContentParts {...baseProps} content={finalContent} isLast />);
|
||||
|
||||
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(<ContentParts {...baseProps} content={compacted} isLast />);
|
||||
|
||||
const settledPhase = screen.getByTestId('activity-phase-group');
|
||||
expect(settledPhase).not.toBe(phaseNode);
|
||||
expect(settledPhase).toHaveAttribute('data-animate-entrance', 'true');
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -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. */
|
||||
|
|
|
|||
|
|
@ -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 } : {}),
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
242
client/src/utils/messages.spec.ts
Normal file
242
client/src/utils/messages.spec.ts
Normal file
|
|
@ -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<string, unknown> = {}): 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<string, unknown> = {}): TMessageContentParts =>
|
||||
({ type: ContentTypes.ACTIVITY_LABEL, activity_label: value, ...extra }) as TMessageContentParts;
|
||||
|
||||
const streamedIndexes = (content: TMessage['content']): Array<number | undefined> =>
|
||||
(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();
|
||||
});
|
||||
});
|
||||
|
|
@ -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<TMessageContentParts | undefined> | 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
|
||||
|
|
|
|||
|
|
@ -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<PartMetadata, 'agentId' | 'groupId'>;
|
||||
export type ContentMetadata = Pick<PartMetadata, 'agentId' | 'groupId' | 'streamedIndex'>;
|
||||
|
||||
export type ContentPart = (
|
||||
| CodeToolCall
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue