From 3d2f3a6a18c9df7c264eefc060de1b712cfd9663 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Tue, 28 Jul 2026 22:26:20 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=A7=A9=20fix:=20Apply=20Every=20Entry=20o?= =?UTF-8?q?f=20Multi-Part=20Stream=20Deltas=20(#14502)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The message and reasoning delta handlers read only delta.content[0], silently dropping the remaining entries of any multi-part delta — reachable today via Google server-side tool chunks, which dispatch several reasoning parts in one event. Every entry now chains through updateContent in order, with the per-part content index computed against the progressively updated response so later entries observe earlier state. Companion to the identical fix in the @librechat/agents aggregator (danny-avila/agents#350), which keeps streamed and saved content consistent. Adds multi-entry coverage for both delta types. --- .../SSE/__tests__/useStepHandler.spec.ts | 80 ++++++++++++++ client/src/hooks/SSE/useStepHandler.ts | 101 ++++++++++-------- 2 files changed, 139 insertions(+), 42 deletions(-) diff --git a/client/src/hooks/SSE/__tests__/useStepHandler.spec.ts b/client/src/hooks/SSE/__tests__/useStepHandler.spec.ts index 288e82ad3c..ac8c7c6240 100644 --- a/client/src/hooks/SSE/__tests__/useStepHandler.spec.ts +++ b/client/src/hooks/SSE/__tests__/useStepHandler.spec.ts @@ -1197,6 +1197,45 @@ describe('useStepHandler', () => { ); }); + it('applies every entry of a multi-part text delta in order', () => { + const responseMessage = createResponseMessage(); + mockGetMessages.mockReturnValue([responseMessage]); + + const { result } = renderHook(() => useStepHandler(createHookParams())); + + const runStep = createRunStep(); + const submission = createSubmission(); + + act(() => { + result.current.stepHandler({ event: StepEvents.ON_RUN_STEP, data: runStep }, submission); + }); + + act(() => { + result.current.stepHandler( + { + event: StepEvents.ON_MESSAGE_DELTA, + data: { + id: 'step-1', + delta: { + content: [ + { type: ContentTypes.TEXT, text: 'Hello ' }, + { type: ContentTypes.TEXT, text: 'streaming ' }, + { type: ContentTypes.TEXT, text: 'world' }, + ], + }, + }, + }, + submission, + ); + }); + + const lastCall = mockSetMessages.mock.calls[mockSetMessages.mock.calls.length - 1][0]; + const responseMsg = lastCall[lastCall.length - 1]; + expect(responseMsg.content).toContainEqual( + expect.objectContaining({ type: ContentTypes.TEXT, text: 'Hello streaming world' }), + ); + }); + it('coalesces multiple deltas into a single flush per animation frame', () => { const rafQueue: FrameRequestCallback[] = []; (window.requestAnimationFrame as jest.Mock).mockImplementation((cb: FrameRequestCallback) => { @@ -1405,6 +1444,47 @@ describe('useStepHandler', () => { expect.objectContaining({ type: ContentTypes.THINK, think: 'First thought' }), ); }); + + it('applies every entry of a multi-part reasoning delta in order', () => { + const responseMessage = createResponseMessage(); + mockGetMessages.mockReturnValue([responseMessage]); + + const { result } = renderHook(() => useStepHandler(createHookParams())); + + const runStep = createRunStep(); + const submission = createSubmission(); + + act(() => { + result.current.stepHandler({ event: StepEvents.ON_RUN_STEP, data: runStep }, submission); + }); + + act(() => { + result.current.stepHandler( + { + event: StepEvents.ON_REASONING_DELTA, + data: { + id: 'step-1', + delta: { + content: [ + { type: ContentTypes.THINK, think: 'First reasoning block. ' }, + { type: ContentTypes.THINK, think: 'Second reasoning block.' }, + ], + }, + }, + }, + submission, + ); + }); + + const lastCall = mockSetMessages.mock.calls[mockSetMessages.mock.calls.length - 1][0]; + const responseMsg = lastCall[lastCall.length - 1]; + expect(responseMsg.content).toContainEqual( + expect.objectContaining({ + type: ContentTypes.THINK, + think: 'First reasoning block. Second reasoning block.', + }), + ); + }); }); describe('on_run_step_delta event', () => { diff --git a/client/src/hooks/SSE/useStepHandler.ts b/client/src/hooks/SSE/useStepHandler.ts index 2969501a62..0055ee9664 100644 --- a/client/src/hooks/SSE/useStepHandler.ts +++ b/client/src/hooks/SSE/useStepHandler.ts @@ -894,29 +894,38 @@ export default function useStepHandler({ const response = messageMap.current.get(responseMessageId); if (response && messageDelta.delta.content) { - const contentPart = Array.isArray(messageDelta.delta.content) - ? messageDelta.delta.content[0] - : messageDelta.delta.content; + /** A delta may carry several parts (e.g. Google server-side tool + * chunks) — every entry must be applied, in order, or streamed + * text is silently dropped. */ + const contentParts = Array.isArray(messageDelta.delta.content) + ? messageDelta.delta.content + : [messageDelta.delta.content]; - if (contentPart == null) { - return; + let updatedResponse = response; + let hasUpdate = false; + for (const contentPart of contentParts) { + if (contentPart == null) { + continue; + } + const currentIndex = calculateContentIndex( + runStep.index, + initialContent, + contentPart.type || '', + updatedResponse.content, + ); + updatedResponse = updateContent( + updatedResponse, + currentIndex, + contentPart, + false, + getStepMetadata(runStep), + ); + hasUpdate = true; + } + if (hasUpdate) { + messageMap.current.set(responseMessageId, updatedResponse); + scheduleCoalescedMessagesFlush(responseMessageId); } - - const currentIndex = calculateContentIndex( - runStep.index, - initialContent, - contentPart.type || '', - response.content, - ); - const updatedResponse = updateContent( - response, - currentIndex, - contentPart, - false, - getStepMetadata(runStep), - ); - messageMap.current.set(responseMessageId, updatedResponse); - scheduleCoalescedMessagesFlush(responseMessageId); } } else if (stepEvent.event === StepEvents.ON_REASONING_DELTA) { const reasoningDelta = stepEvent.data; @@ -936,29 +945,37 @@ export default function useStepHandler({ const response = messageMap.current.get(responseMessageId); if (response && reasoningDelta.delta.content != null) { - const contentPart = Array.isArray(reasoningDelta.delta.content) - ? reasoningDelta.delta.content[0] - : reasoningDelta.delta.content; + /** Same multi-part contract as message deltas: Google server-side + * tool chunks emit several think entries in one delta. */ + const contentParts = Array.isArray(reasoningDelta.delta.content) + ? reasoningDelta.delta.content + : [reasoningDelta.delta.content]; - if (contentPart == null) { - return; + let updatedResponse = response; + let hasUpdate = false; + for (const contentPart of contentParts) { + if (contentPart == null) { + continue; + } + const currentIndex = calculateContentIndex( + runStep.index, + initialContent, + contentPart.type || '', + updatedResponse.content, + ); + updatedResponse = updateContent( + updatedResponse, + currentIndex, + contentPart, + false, + getStepMetadata(runStep), + ); + hasUpdate = true; + } + if (hasUpdate) { + messageMap.current.set(responseMessageId, updatedResponse); + scheduleCoalescedMessagesFlush(responseMessageId); } - - const currentIndex = calculateContentIndex( - runStep.index, - initialContent, - contentPart.type || '', - response.content, - ); - const updatedResponse = updateContent( - response, - currentIndex, - contentPart, - false, - getStepMetadata(runStep), - ); - messageMap.current.set(responseMessageId, updatedResponse); - scheduleCoalescedMessagesFlush(responseMessageId); } } else if (stepEvent.event === StepEvents.ON_RUN_STEP_DELTA) { const runStepDelta = stepEvent.data;