From 6f7dc1b289776c6f7ae97334a4cb8ff9b04a8702 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Sun, 24 May 2026 19:31:42 -0400 Subject: [PATCH] =?UTF-8?q?=E2=8F=AD=EF=B8=8F=20fix:=20Avoid=20False=20Res?= =?UTF-8?q?ume=20Submission=20Stale=20Detection=20(#13297)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix: avoid false stale resume submission detection * style: format resume on load spec * test: cover delayed resume message identity --- .../SSE/__tests__/useResumeOnLoad.spec.tsx | 194 ++++++++++++++++++ client/src/hooks/SSE/useResumeOnLoad.ts | 68 +++++- 2 files changed, 258 insertions(+), 4 deletions(-) create mode 100644 client/src/hooks/SSE/__tests__/useResumeOnLoad.spec.tsx diff --git a/client/src/hooks/SSE/__tests__/useResumeOnLoad.spec.tsx b/client/src/hooks/SSE/__tests__/useResumeOnLoad.spec.tsx new file mode 100644 index 0000000000..9b785b9655 --- /dev/null +++ b/client/src/hooks/SSE/__tests__/useResumeOnLoad.spec.tsx @@ -0,0 +1,194 @@ +import { RecoilRoot, useRecoilValue } from 'recoil'; +import { Constants } from 'librechat-data-provider'; +import { renderHook, act } from '@testing-library/react'; + +import type { TMessage, TConversation, TSubmission } from 'librechat-data-provider'; +import type { MutableSnapshot } from 'recoil'; +import type { ReactNode } from 'react'; + +import useResumeOnLoad from '../useResumeOnLoad'; +import store from '~/store'; + +const mockUseStreamStatus = jest.fn(); + +jest.mock('~/data-provider', () => ({ + useStreamStatus: (conversationId: string | undefined, enabled: boolean) => + mockUseStreamStatus(conversationId, enabled), +})); + +const CONVERSATION_ID = 'conv-current'; +const STALE_CONVERSATION_ID = 'conv-stale'; +const USER_MESSAGE_ID = 'user-message-1'; +const RESPONSE_MESSAGE_ID = 'user-message-1_'; + +function buildConversation(conversationId = CONVERSATION_ID): TConversation { + return { + conversationId, + endpoint: 'agents', + } as TConversation; +} + +function buildUserMessage( + conversationId: string | null = CONVERSATION_ID, + messageId = USER_MESSAGE_ID, +): TMessage { + return { + text: 'Hello', + sender: 'User', + messageId, + conversationId, + isCreatedByUser: true, + parentMessageId: Constants.NO_PARENT, + } as TMessage; +} + +function buildSubmission(conversationId: string | null | undefined): TSubmission { + return { + messages: [], + isTemporary: false, + endpointOption: { endpoint: 'agents' }, + conversation: { conversationId }, + userMessage: buildUserMessage(null), + initialResponse: { + text: '', + sender: 'Assistant', + messageId: RESPONSE_MESSAGE_ID, + conversationId, + isCreatedByUser: false, + parentMessageId: USER_MESSAGE_ID, + } as TMessage, + } as unknown as TSubmission; +} + +function renderUseResumeOnLoad({ + messages = [], + getMessages: getMessagesOverride, + submission = null, + conversationId = CONVERSATION_ID, + messagesLoaded = true, + onSubmission, +}: { + messages?: TMessage[]; + getMessages?: () => TMessage[] | undefined; + submission?: TSubmission | null; + conversationId?: string; + messagesLoaded?: boolean; + onSubmission?: (submission: TSubmission | null) => void; +}) { + const getMessages = jest.fn(getMessagesOverride ?? (() => messages)); + const initializeState = (snapshot: MutableSnapshot) => { + snapshot.set(store.conversationByIndex(0), buildConversation(conversationId)); + snapshot.set(store.submissionByIndex(0), submission); + }; + + const SubmissionProbe = () => { + const currentSubmission = useRecoilValue(store.submissionByIndex(0)); + onSubmission?.(currentSubmission); + return null; + }; + + const wrapper = ({ children }: { children: ReactNode }) => ( + + + {children} + + ); + + return { + getMessages, + ...renderHook(() => useResumeOnLoad(conversationId, getMessages, 0, messagesLoaded), { + wrapper, + }), + }; +} + +describe('useResumeOnLoad', () => { + beforeEach(() => { + jest.spyOn(console, 'log').mockImplementation(() => undefined); + mockUseStreamStatus.mockReset(); + mockUseStreamStatus.mockReturnValue({ + data: undefined, + isSuccess: false, + isFetching: false, + }); + }); + + afterEach(() => { + jest.restoreAllMocks(); + }); + + it('does not check for resume when a null-conversation submission matches a loaded user message', () => { + renderUseResumeOnLoad({ + submission: buildSubmission(null), + messages: [buildUserMessage(CONVERSATION_ID)], + }); + + expect(mockUseStreamStatus).toHaveBeenCalledWith(CONVERSATION_ID, false); + }); + + it('checks for resume when the active submission belongs to a different conversation', () => { + renderUseResumeOnLoad({ + submission: buildSubmission(STALE_CONVERSATION_ID), + messages: [buildUserMessage(CONVERSATION_ID)], + }); + + expect(mockUseStreamStatus).toHaveBeenCalledWith(CONVERSATION_ID, true); + }); + + it('checks for resume when a null-conversation submission cannot be matched to loaded messages', () => { + renderUseResumeOnLoad({ + submission: buildSubmission(null), + messages: [buildUserMessage(CONVERSATION_ID, 'different-user-message')], + }); + + expect(mockUseStreamStatus).toHaveBeenCalledWith(CONVERSATION_ID, true); + }); + + it('stops checking for resume after loaded messages prove a null-conversation submission belongs to the route', () => { + const submission = buildSubmission(null); + let messages: TMessage[] = []; + const { rerender } = renderUseResumeOnLoad({ + submission, + getMessages: () => messages, + }); + + expect(mockUseStreamStatus).toHaveBeenLastCalledWith(CONVERSATION_ID, true); + + messages = [buildUserMessage(CONVERSATION_ID)]; + rerender(); + + expect(mockUseStreamStatus).toHaveBeenLastCalledWith(CONVERSATION_ID, false); + }); + + it('does not replace a null-conversation submission when stream status matches its resume state', async () => { + const submission = buildSubmission(null); + const observedSubmissions: Array = []; + mockUseStreamStatus.mockReturnValue({ + isSuccess: true, + isFetching: false, + data: { + active: true, + status: 'running', + streamId: 'stream-1', + resumeState: { + aggregatedContent: [], + responseMessageId: RESPONSE_MESSAGE_ID, + userMessage: { messageId: USER_MESSAGE_ID }, + }, + }, + }); + + renderUseResumeOnLoad({ + submission, + messages: [], + onSubmission: (currentSubmission) => observedSubmissions.push(currentSubmission), + }); + + await act(async () => { + await Promise.resolve(); + }); + + expect(mockUseStreamStatus).toHaveBeenCalledWith(CONVERSATION_ID, true); + expect(observedSubmissions[observedSubmissions.length - 1]).toBe(submission); + }); +}); diff --git a/client/src/hooks/SSE/useResumeOnLoad.ts b/client/src/hooks/SSE/useResumeOnLoad.ts index 5f0f691787..a0d0b6c4d5 100644 --- a/client/src/hooks/SSE/useResumeOnLoad.ts +++ b/client/src/hooks/SSE/useResumeOnLoad.ts @@ -2,9 +2,46 @@ import { useEffect, useRef } from 'react'; import { useSetRecoilState, useRecoilValue } from 'recoil'; import { Constants, tMessageSchema, isAssistantsEndpoint } from 'librechat-data-provider'; import type { TMessage, TConversation, TSubmission, Agents } from 'librechat-data-provider'; +import type { StreamStatusResponse } from '~/data-provider'; import { useStreamStatus } from '~/data-provider'; import store from '~/store'; +function hasSubmissionUserMessage( + submission: TSubmission | null, + messages: TMessage[] | undefined, + conversationId: string | undefined, +): boolean { + const userMessageId = submission?.userMessage?.messageId; + if (!userMessageId || !conversationId || !messages?.length) { + return false; + } + + return messages.some( + (message) => + message.isCreatedByUser === true && + message.messageId === userMessageId && + message.conversationId === conversationId, + ); +} + +function resumeStateMatchesSubmission( + streamStatus: StreamStatusResponse | undefined, + submission: TSubmission | null, +): boolean { + const resumeState = streamStatus?.resumeState; + if (!resumeState || !submission) { + return false; + } + + const userMessageId = submission.userMessage?.messageId; + if (userMessageId && resumeState.userMessage?.messageId === userMessageId) { + return true; + } + + const responseMessageId = submission.initialResponse?.messageId; + return !!responseMessageId && resumeState.responseMessageId === responseMessageId; +} + /** * Build a submission object from resume state for reconnected streams. * This provides the minimum data needed for useResumableSSE to subscribe. @@ -113,14 +150,21 @@ export default function useResumeOnLoad( const processedConvoRef = useRef(null); // Check for active stream when conversation changes - // Allow check if no submission OR submission is for a different conversation (stale) const submissionConvoId = currentSubmission?.conversation?.conversationId; - const hasActiveSubmissionForThisConvo = currentSubmission && submissionConvoId === conversationId; + const loadedMessages = messagesLoaded ? getMessages() : undefined; + const hasExplicitSubmissionMatch = !!conversationId && submissionConvoId === conversationId; + const hasHydratedMessageMatch = + submissionConvoId == null && + hasSubmissionUserMessage(currentSubmission, loadedMessages, conversationId); + const hasActiveSubmissionForThisConvo = + !!currentSubmission && (hasExplicitSubmissionMatch || hasHydratedMessageMatch); + const hasStaleSubmissionForDifferentConvo = + !!currentSubmission && submissionConvoId != null && submissionConvoId !== conversationId; const shouldCheck = resumableEnabled && messagesLoaded && // Wait for messages to load before checking - !hasActiveSubmissionForThisConvo && // Allow if no submission or stale submission + !hasActiveSubmissionForThisConvo && // Allow if no submission or a confirmed stale submission !!conversationId && conversationId !== Constants.NEW_CONVO && processedConvoRef.current !== conversationId; // Don't re-check processed convos @@ -166,7 +210,7 @@ export default function useResumeOnLoad( } // If there's a stale submission for a different conversation, log it but continue - if (currentSubmission && submissionConvoId !== conversationId) { + if (hasStaleSubmissionForDifferentConvo) { console.log( '[ResumeOnLoad] Found stale submission for different conversation, will check for resume', { @@ -183,6 +227,21 @@ export default function useResumeOnLoad( return; } + if ( + streamStatus.active && + streamStatus.streamId && + submissionConvoId == null && + resumeStateMatchesSubmission(streamStatus, currentSubmission) + ) { + console.log('[ResumeOnLoad] Skipping - active submission matches stream status', { + streamId: streamStatus.streamId, + currentConvoId: conversationId, + userMessageId: currentSubmission?.userMessage?.messageId, + }); + processedConvoRef.current = conversationId; + return; + } + // Don't process the same conversation twice if (processedConvoRef.current === conversationId) { console.log('[ResumeOnLoad] Skipping - already processed this conversation'); @@ -242,6 +301,7 @@ export default function useResumeOnLoad( messagesLoaded, hasActiveSubmissionForThisConvo, submissionConvoId, + hasStaleSubmissionForDifferentConvo, currentSubmission, isSuccess, isFetching,