⏭️ fix: Avoid False Resume Submission Stale Detection (#13297)

* fix: avoid false stale resume submission detection

* style: format resume on load spec

* test: cover delayed resume message identity

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 }) => (
+    <RecoilRoot initializeState={initializeState}>
+      <SubmissionProbe />
+      {children}
+    </RecoilRoot>
+  );
+
+  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<TSubmission | null> = [];
+    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);
+  });
+});

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<string | null>(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,