diff --git a/client/src/hooks/Chat/__tests__/transport.spec.tsx b/client/src/hooks/Chat/__tests__/transport.spec.tsx index b850423fdc6..4d9fe04608a 100644 --- a/client/src/hooks/Chat/__tests__/transport.spec.tsx +++ b/client/src/hooks/Chat/__tests__/transport.spec.tsx @@ -750,6 +750,24 @@ describe('chat transport boundary', () => { await waitFor(() => expect(result.current.status).toBe('error')); expect(result.current.error?.message).toContain('Generation failed'); }); + + it('reports a failure on a stream reattached without resume state', async () => { + const fake = createFakeTransport(); + const { result } = renderPane(fake.transport); + await waitFor(() => expect(statusReads()).toBe(1)); + /** An older server answers a running job without the turn it belongs to. */ + status = { active: true, streamId: 'convo-1', status: 'running', createdAt: 2000 }; + await act(async () => { + await result.current.resumeStream(); + }); + await waitFor(() => expect(fake.streams).toHaveLength(1)); + + status = { active: false, status: 'error', createdAt: 2000 }; + act(() => fake.streams[0].options.onEvent({ type: 'error', data: { message: 'x' } })); + + await waitFor(() => expect(result.current.status).toBe('error')); + expect(result.current.error?.message).toContain('x'); + }); }); describe('useChat', () => { diff --git a/client/src/hooks/SSE/useResumableSSE.ts b/client/src/hooks/SSE/useResumableSSE.ts index 2c3eaefb811..43a9589526c 100644 --- a/client/src/hooks/SSE/useResumableSSE.ts +++ b/client/src/hooks/SSE/useResumableSSE.ts @@ -26,37 +26,43 @@ import { } from 'librechat-data-provider'; import { MAX_RETRIES, + getStepEventId, + toStartGenerationError, + isOAuthStepEvent, getReconnectDelay, - START_GENERATION_NETWORK_RETRIES, - START_GENERATION_READINESS_TIMEOUT_MS, + waitForRetryDelay, + applyPendingAction, getRecoverySteerId, - getPredecessorMismatchStartResponse, - isUnnegotiatedV2StartControl, - getStartGenerationCreatedAt, - getStartGenerationConversationId, - getStartGenerationStreamId, + getRetryAfterDelay, + mergeResumeMessages, + preferDefinedString, + isServerNotReadyError, + getOptimisticMessages, isResumedStartResponse, isSettledStartResponse, - isReplacedStartResponse, isRetryableNetworkError, - isServerNotReadyError, - getRetryAfterDelay, - waitForRetryDelay, - getStreamStartFailureText, - isOAuthStepEvent, - getStepEventId, - toStartGenerationError, - GENERATION_PROTOCOL_VERSION, - getGenerationProtocolVersion, - supportsGenerationProtocolV2, + isReplacedStartResponse, + getResumeMessageIndexes, + countTaggedApprovalParts, isInitialNewConversation, + getStreamStartFailureText, + keepLocalCodeApprovalMode, + hydrateSubmissionMessages, + buildResumeEventSubmission, completedResponseMessageId, - preferDefinedString, - getOptimisticMessages, + getStartGenerationStreamId, buildOptimisticConversation, - hydrateSubmissionMessages, buildCreatedInitialResponse, - keepLocalCodeApprovalMode, + getGenerationProtocolVersion, + GENERATION_PROTOCOL_VERSION, + getStartGenerationCreatedAt, + isUnnegotiatedV2StartControl, + supportsGenerationProtocolV2, + findPendingActionMessageIndex, + START_GENERATION_NETWORK_RETRIES, + getStartGenerationConversationId, + getPredecessorMismatchStartResponse, + START_GENERATION_READINESS_TIMEOUT_MS, } from '@librechat/chat'; import type { Agents, @@ -86,7 +92,6 @@ import { logger, clearComposerDrafts, applySteerPart, - applyPendingAction, resolveRunEndTarget, findSteerMessageIndex, applyActivityLabelPart, @@ -99,12 +104,9 @@ import { removeConvoFromAllQueries, upsertConvoInAllQueries, invalidateConversationLists, - countTaggedApprovalParts, countTrailingOutputChars, markStreamStartFailedMetadata, - findPendingActionMessageIndex, insertQueuedOrigin, - isCompactionAnchorProjection, } from '~/utils'; import { resolveDetachedRunEnd, @@ -186,217 +188,6 @@ const replaceNewConversationUrl = (conversationId: string) => { ); }; -const buildResumeEventSubmission = ( - currentSubmission: TSubmission, - currentUserMessage: TMessage, - resumeState?: Agents.ResumeState | null, -): EventSubmission => { - if (!resumeState) { - return { ...currentSubmission, userMessage: currentUserMessage } as EventSubmission; - } - - const conversationId = - resumeState.conversationId ?? - resumeState.userMessage?.conversationId ?? - currentUserMessage.conversationId ?? - currentSubmission.conversation?.conversationId; - - /** - * A compaction submits no user turn: its user-message slot names the LEAF it - * summarizes up to, and the server projects that anchor as identity only - * (`projectCompactionAnchor` — no parent, no author). Adopting the projection - * as a user row would rewrite the leaf into an empty, parentless message, so - * an anchored run keeps the identity it resumed with. - */ - /** Read from the projection as well as the flag: a re-attach whose submission - * never learned it was a compaction still has to recognize the anchor. */ - const anchoredUserMessage = - currentSubmission.compact === true || isCompactionAnchorProjection(resumeState.userMessage); - - const userMessage = ( - anchoredUserMessage - ? { ...currentUserMessage, conversationId } - : { - ...currentUserMessage, - ...resumeState.userMessage, - conversationId, - isCreatedByUser: true, - } - ) as TMessage; - - const responseMessageId = - resumeState.responseMessageId ?? - currentSubmission.initialResponse?.messageId ?? - `${userMessage.messageId}_`; - - const initialResponse = { - ...(currentSubmission.initialResponse as TMessage), - messageId: responseMessageId, - parentMessageId: userMessage.messageId, - conversationId, - content: - resumeState.aggregatedContent ?? - (currentSubmission.initialResponse as TMessage | undefined)?.content, - sender: resumeState.sender ?? currentSubmission.initialResponse?.sender, - iconURL: preferDefinedString(currentSubmission.initialResponse?.iconURL, resumeState.iconURL), - model: preferDefinedString(currentSubmission.initialResponse?.model, resumeState.model), - isCreatedByUser: false, - } as TMessage; - - return { - ...currentSubmission, - conversation: { - ...currentSubmission.conversation, - conversationId, - }, - userMessage, - initialResponse, - /** Carried so every consumer of the resumed submission — the merges below, - * the abort-error write — reads the same answer this resolved. */ - ...(anchoredUserMessage && { compact: true }), - } as EventSubmission; -}; - -type ResumeMessageIndexes = { - userIndex: number; - responseIndex: number; - preliminaryUserIndex: number; - preliminaryResponseIndex: number; -}; - -const getResumeMessageIndexes = ( - messages: TMessage[], - userMessageId: string, - responseMessageId: string, - preliminaryUserMessageId?: string, - preliminaryResponseMessageId?: string, -): ResumeMessageIndexes => { - let userIndex = -1; - let responseIndex = -1; - let preliminaryUserIndex = -1; - let preliminaryResponseIndex = -1; - const hasPreliminaryResponse = preliminaryResponseMessageId?.endsWith('_') === true; - const eligiblePreliminaryUserId = - hasPreliminaryResponse && preliminaryUserMessageId && preliminaryUserMessageId !== userMessageId - ? preliminaryUserMessageId - : undefined; - const eligiblePreliminaryResponseId = - hasPreliminaryResponse && preliminaryResponseMessageId !== responseMessageId - ? preliminaryResponseMessageId - : undefined; - - for (let index = 0; index < messages.length; index++) { - const messageId = messages[index]?.messageId; - if (userIndex < 0 && messageId === userMessageId) { - userIndex = index; - } - if (responseIndex < 0 && messageId === responseMessageId) { - responseIndex = index; - } - if ( - preliminaryUserIndex < 0 && - eligiblePreliminaryUserId && - messageId === eligiblePreliminaryUserId - ) { - preliminaryUserIndex = index; - } - if ( - preliminaryResponseIndex < 0 && - eligiblePreliminaryResponseId && - messageId === eligiblePreliminaryResponseId - ) { - preliminaryResponseIndex = index; - } - } - - return { userIndex, responseIndex, preliminaryUserIndex, preliminaryResponseIndex }; -}; - -const mergeResumeMessages = ( - messages: TMessage[], - userMessage: TMessage, - responseMessage: TMessage, - indexes: ResumeMessageIndexes, - /** - * An anchored run — a compaction — owns no user row. Its user-message slot - * holds the leaf the summary hangs off, a row the transcript already has: - * merging the slot onto it would replace that answer with an empty user - * message, and inserting it would duplicate its id. Either way the row loses - * its parent and `buildTree` files it as a phantom root, folding the thread. - */ - anchoredUserMessage = false, -): TMessage[] => { - const nextMessages = [...messages]; - let { userIndex, responseIndex, preliminaryResponseIndex } = indexes; - const { preliminaryUserIndex } = indexes; - - if (preliminaryUserIndex >= 0 && !anchoredUserMessage) { - if (userIndex >= 0) { - nextMessages.splice(preliminaryUserIndex, 1); - if (userIndex > preliminaryUserIndex) { - userIndex -= 1; - } - if (responseIndex > preliminaryUserIndex) { - responseIndex -= 1; - } - if (preliminaryResponseIndex > preliminaryUserIndex) { - preliminaryResponseIndex -= 1; - } - } else { - nextMessages[preliminaryUserIndex] = { - ...nextMessages[preliminaryUserIndex], - ...userMessage, - }; - userIndex = preliminaryUserIndex; - } - } - - if (preliminaryResponseIndex >= 0) { - if (responseIndex >= 0) { - nextMessages.splice(preliminaryResponseIndex, 1); - if (userIndex > preliminaryResponseIndex) { - userIndex -= 1; - } - if (responseIndex > preliminaryResponseIndex) { - responseIndex -= 1; - } - } else { - nextMessages[preliminaryResponseIndex] = { - ...nextMessages[preliminaryResponseIndex], - ...responseMessage, - }; - responseIndex = preliminaryResponseIndex; - } - } - - if (userIndex >= 0 && !anchoredUserMessage) { - nextMessages[userIndex] = { ...nextMessages[userIndex], ...userMessage }; - } - - if (responseIndex >= 0) { - nextMessages[responseIndex] = { ...nextMessages[responseIndex], ...responseMessage }; - } - - if (anchoredUserMessage) { - return responseIndex >= 0 ? nextMessages : [...nextMessages, responseMessage]; - } - - if (userIndex >= 0 && responseIndex >= 0) { - return nextMessages; - } - - if (userIndex >= 0) { - return [...nextMessages, responseMessage]; - } - - if (responseIndex >= 0) { - nextMessages.splice(responseIndex, 0, userMessage); - return nextMessages; - } - - return [...nextMessages, userMessage, responseMessage]; -}; - export { ABORT_SWEEP_STATUSES, selectLocalSteersForQueue }; /** diff --git a/client/src/hooks/SSE/useResumeOnLoad.ts b/client/src/hooks/SSE/useResumeOnLoad.ts index 78b518794b2..8b8c82e6a32 100644 --- a/client/src/hooks/SSE/useResumeOnLoad.ts +++ b/client/src/hooks/SSE/useResumeOnLoad.ts @@ -9,13 +9,19 @@ import { import { Constants, QueryKeys, - tMessageSchema, isAssistantsEndpoint, isForcedTemporaryRetention, } from 'librechat-data-provider'; -import type { TMessage, TConversation, TSubmission, Agents } from 'librechat-data-provider'; -import type { GenerationProtocolVersion } from '~/data-provider/SSE/protocol'; -import type { StreamStatusResponse } from '~/data-provider'; +import { + hasSubmissionUserMessage, + getGenerationProtocolVersion, + supportsGenerationProtocolV2, + buildMinimalResumeSubmission, + resumeStateMatchesSubmission, + getResumeBranchTargetMessageId, + buildSubmissionFromResumeState, +} from '@librechat/chat'; +import type { TMessage, Agents } from 'librechat-data-provider'; import { useStreamStatus, useActiveJobs, @@ -26,24 +32,15 @@ import { extendActiveJobsGrace, ACTIVE_JOBS_SUCCESSOR_GRACE_MS, } from '~/data-provider'; -import { - dedupeSteersById, - applyPendingAction, - getBranchSiblingIndexesForTarget, - isCompactionAnchorProjection, -} from '~/utils'; import { isDetachedRunManual, resolveDetachedRunEnd, pendingRunEndByConvoId, detachedRunByConvoId, } from '~/hooks/Chat/queue'; -import { - getGenerationProtocolVersion, - supportsGenerationProtocolV2, -} from '~/data-provider/SSE/protocol'; import { siblingIdxFamily, siblingKey } from '~/components/Chat/Messages/Thread/state'; import { pendingApprovalActionFamily } from '~/components/Chat/approval/state'; +import { dedupeSteersById, getBranchSiblingIndexesForTarget } from '~/utils'; import { agentQueuedTurnsQueryKey } from '~/data-provider/SSE/queuedTurns'; import { submissionStartFamily } from '~/store/generation'; import { resumeRequestsAtom } from '~/hooks/Chat/resume'; @@ -56,229 +53,6 @@ import store from '~/store'; */ const ACTIVE_JOB_REARM_INTERVAL_MS = 5_000; -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; -} - -/** - * The row that names the branch a resumed run belongs to when its own response - * is not in the loaded history yet. A compaction's user-message slot is the leaf - * it summarizes up to and carries no parent (`projectCompactionAnchor`), so - * there the anchor itself names the branch — without this the pane restores no - * sibling selection and a compaction started on an older branch comes back on - * whichever branch the default lands on. - */ -function getResumeBranchFallbackMessageId( - resumeState: Agents.ResumeState, -): string | null | undefined { - return resumeState.userMessage?.parentMessageId ?? resumeState.userMessage?.messageId; -} - -function getResumeBranchTargetMessageId( - resumeState: Agents.ResumeState, - messages: TMessage[], -): string | null | undefined { - const responseMessageId = resumeState.responseMessageId; - if (!responseMessageId) { - return getResumeBranchFallbackMessageId(resumeState); - } - - const unpaddedResponseMessageId = responseMessageId.replace(/_+$/, ''); - let hasResponseMessage = false; - let hasUnpaddedResponseMessage = false; - - for (const message of messages) { - if (message.messageId === responseMessageId) { - hasResponseMessage = true; - break; - } - - if (message.messageId === unpaddedResponseMessageId) { - hasUnpaddedResponseMessage = true; - } - } - - if (hasResponseMessage) { - return responseMessageId; - } - - if (hasUnpaddedResponseMessage) { - return unpaddedResponseMessageId; - } - - return getResumeBranchFallbackMessageId(resumeState); -} - -function preferDefinedString(value?: string | null, fallback?: string): string | undefined { - return value != null && value !== '' ? value : fallback; -} - -/** - * Build a submission object from resume state for reconnected streams. - * This provides the minimum data needed for useResumableSSE to subscribe. - */ -function buildSubmissionFromResumeState( - resumeState: Agents.ResumeState, - streamId: string, - messages: TMessage[], - conversationId: string, - generationCreatedAt?: number, - generationProtocolVersion: GenerationProtocolVersion = 1, - isTemporary = false, -): TSubmission { - const userMessageData = resumeState.userMessage; - const responseMessageId = - resumeState.responseMessageId ?? `${userMessageData?.messageId ?? 'resume'}_`; - - /** - * The run's user-message slot as the loaded history already holds it. Message - * ids are unique per row, so a match is that row whoever wrote it: a - * compaction submits no user turn and puts the LEAF it summarizes up to in - * this slot (`useCompactConversation` client-side, `projectCompactionAnchor` - * server-side), which is usually the assistant answer. Adopting the row keeps - * the anchor's own identity — synthesizing an empty, parentless USER row over - * it rewrites that answer into a phantom root and folds the thread. - */ - const existingSlotMessage = messages.find((m) => m.messageId === userMessageData?.messageId); - /** An anchored run — a compaction — created no user turn: the slot names a row - * the transcript already holds. Judged from the projection's shape, never from - * the anchor's author, which is a user message as often as an answer. Either - * way the run is regenerate-shaped for every consumer: no user turn of its - * own, the response parented onto an existing message. */ - const isAnchoredRun = isCompactionAnchorProjection(userMessageData); - - // A trailing underscore distinguishes an in-flight regeneration from the persisted - // response it replaces. Only the exact response id proves generation ownership. - const existingResponseMessage = messages.find( - (m) => !m.isCreatedByUser && m.messageId === responseMessageId, - ); - // The persisted row may seed display metadata, but never identity or deduplication. - const unpaddedResponseMessageId = responseMessageId.replace(/_+$/, ''); - const persistedRegenerationResponse = - unpaddedResponseMessageId !== responseMessageId - ? messages.find((m) => !m.isCreatedByUser && m.messageId === unpaddedResponseMessageId) - : undefined; - const responseMetadataMessage = existingResponseMessage ?? persistedRegenerationResponse; - const isRegenerateResume = - resumeState.isRegenerate === true || persistedRegenerationResponse != null || isAnchoredRun; - let regenerateMessages: TMessage[] | undefined; - if (isRegenerateResume) { - regenerateMessages = - unpaddedResponseMessageId === responseMessageId - ? [...messages] - : messages.filter((message) => message.messageId !== responseMessageId); - } - - // Create or use the row the slot already names - const userMessage: TMessage = - existingSlotMessage ?? - (userMessageData - ? (tMessageSchema.parse({ - messageId: userMessageData.messageId, - parentMessageId: userMessageData.parentMessageId ?? Constants.NO_PARENT, - conversationId: userMessageData.conversationId ?? conversationId, - text: userMessageData.text ?? '', - isCreatedByUser: true, - role: 'user', - }) as TMessage) - : (messages[messages.length - 2] ?? - ({ - messageId: 'resume_user_msg', - conversationId, - text: '', - isCreatedByUser: true, - } as TMessage))); - - // ALWAYS use aggregatedContent from resumeState - it has the latest content from the running job. - // DB content may be stale (saved at disconnect, but generation continued). - let initialResponse: TMessage = { - messageId: existingResponseMessage?.messageId ?? responseMessageId, - parentMessageId: existingResponseMessage?.parentMessageId ?? userMessage.messageId, - conversationId, - text: '', - // aggregatedContent is authoritative - it reflects actual job state - content: (resumeState.aggregatedContent as TMessage['content']) ?? [], - isCreatedByUser: false, - role: 'assistant', - sender: responseMetadataMessage?.sender ?? resumeState.sender, - model: preferDefinedString(responseMetadataMessage?.model, resumeState.model), - iconURL: preferDefinedString(responseMetadataMessage?.iconURL, resumeState.iconURL), - } as TMessage; - - // Re-paused turn: seed the approval / ask-user controls straight onto the - // placeholder so they render on load without waiting for the SSE sync replay. - if (resumeState.pendingAction) { - initialResponse = applyPendingAction(initialResponse, resumeState.pendingAction); - } - - const conversation: TConversation = { - conversationId, - title: 'Resumed Chat', - endpoint: null, - } as TConversation; - - // Non-regenerate resumes strip the persisted request/response pair before handlers - // re-supply it. A regeneration keeps the original branch for early-abort rollback; - // explicit resume metadata covers edited regenerations that reuse the exact response id. - const dedupedMessages = messages.filter( - (m) => - m.messageId !== initialResponse.messageId && - (isRegenerateResume || m.messageId !== userMessage.messageId), - ); - - return { - messages: dedupedMessages, - userMessage, - initialResponse, - conversation, - isRegenerate: isRegenerateResume, - ...(isAnchoredRun && { compact: true }), - ...(regenerateMessages && { regenerateMessages }), - isTemporary, - endpointOption: {}, - // Signal to useResumableSSE to subscribe to existing stream instead of starting new - resumeStreamId: streamId, - ...(generationCreatedAt != null && { resumeGenerationCreatedAt: generationCreatedAt }), - resumeGenerationProtocolVersion: generationProtocolVersion, - } as TSubmission & { - resumeStreamId: string; - resumeGenerationCreatedAt?: number; - resumeGenerationProtocolVersion: GenerationProtocolVersion; - }; -} - /** * Hook to resume streaming if navigating to a conversation with active generation. * Checks stream status via React Query and sets submission if active job found. @@ -1000,34 +774,17 @@ export default function useResumeOnLoad( ); setSubmission(submission); } else { - // Minimal submission without resume state - const lastUserMessage = [...messages].reverse().find((m) => m.isCreatedByUser); - const submission = { - messages, - userMessage: - lastUserMessage ?? ({ messageId: 'resume', conversationId, text: '' } as TMessage), - initialResponse: { - messageId: 'resume_', + setSubmission( + buildMinimalResumeSubmission({ + messages, conversationId, - text: '', - content: streamStatus.aggregatedContent ?? [{ type: 'text', text: '' }], - } as TMessage, - conversation: { conversationId, title: 'Resumed Chat' } as TConversation, - isRegenerate: false, - isTemporary: streamStatus.isTemporary === true || isRetentionForced, - endpointOption: {}, - // Signal to useResumableSSE to subscribe to existing stream instead of starting new - resumeStreamId: streamStatus.streamId, - ...(streamStatus.createdAt != null && { - resumeGenerationCreatedAt: streamStatus.createdAt, + streamId: streamStatus.streamId, + createdAt: streamStatus.createdAt, + aggregatedContent: streamStatus.aggregatedContent, + isTemporary: streamStatus.isTemporary === true || isRetentionForced, + generationProtocolVersion, }), - resumeGenerationProtocolVersion: generationProtocolVersion, - } as TSubmission & { - resumeStreamId: string; - resumeGenerationCreatedAt?: number; - resumeGenerationProtocolVersion: GenerationProtocolVersion; - }; - setSubmission(submission); + ); } }, [ conversationId, diff --git a/client/src/utils/approval.ts b/client/src/utils/approval.ts index 8ee018fcea5..9859476b909 100644 --- a/client/src/utils/approval.ts +++ b/client/src/utils/approval.ts @@ -15,6 +15,11 @@ export { findLiveAskUserQuestion, isAnsweredAskUserQuestionPart, }; +export { + applyPendingAction, + countTaggedApprovalParts, + findPendingActionMessageIndex, +} from '@librechat/chat'; export type { AskUserQuestionPart } from '@librechat/chat'; /** @@ -27,164 +32,6 @@ const ASK_USER_QUESTION_ID_PATTERN = /^[A-Za-z][A-Za-z0-9_-]{0,63}$/; const MAX_ASK_USER_QUESTIONS = 4; const MAX_ASK_USER_QUESTION_HEADER_LENGTH = 80; -const getToolCallId = (part: TMessageContentParts | undefined): string => - (part?.[ContentTypes.TOOL_CALL] as Agents.ToolCall | undefined)?.id ?? ''; - -/** A tool-call value that may carry approval state and/or nested subagent content. */ -type ToolCallWithApproval = Agents.ToolCall & { - approval?: unknown; - subagent_content?: TMessageContentParts[]; -}; - -/** - * Tags one tool-call part with the pending action's approval when it matches an - * `action_request` (joined by `tool_call_id`, NOT position) and is still unresolved. - * - * Recurses into a subagent's `subagent_content`: a tool paused INSIDE a subagent - * lives there, not as a top-level part, so without this the approval never attaches - * and the user gets no controls. Returns a NEW part only when something changed. - */ -function tagApprovalOnPart( - part: TMessageContentParts, - actionId: string, - requestByToolCallId: Map, - reviewByToolCallId: Map, -): { part: TMessageContentParts; changed: boolean } { - if (part?.type !== ContentTypes.TOOL_CALL) { - return { part, changed: false }; - } - const toolCall = part[ContentTypes.TOOL_CALL] as ToolCallWithApproval | undefined; - if (!toolCall) { - return { part, changed: false }; - } - - let nextToolCall = toolCall; - let changed = false; - - // Descend into nested subagent tool calls first. - if (Array.isArray(toolCall.subagent_content) && toolCall.subagent_content.length > 0) { - let nestedChanged = false; - const nextNested = toolCall.subagent_content.map((nestedPart) => { - const res = tagApprovalOnPart(nestedPart, actionId, requestByToolCallId, reviewByToolCallId); - if (res.changed) { - nestedChanged = true; - } - return res.part; - }); - if (nestedChanged) { - nextToolCall = { ...nextToolCall, subagent_content: nextNested }; - changed = true; - } - } - - // Tag this call itself when it's one of the paused requests and still unresolved - // (an `output` means the pause already resolved — leave it alone). - const toolCallId = getToolCallId(part); - const request = toolCallId ? requestByToolCallId.get(toolCallId) : undefined; - if (request && (nextToolCall.output?.length ?? 0) === 0) { - const reviewConfig = reviewByToolCallId.get(toolCallId); - nextToolCall = { - ...nextToolCall, - // A PreToolUse hook may replace the model's original args before asking. - // The interrupt payload is authoritative so the reviewer sees, edits, and - // approves the same arguments the resumed tool will actually execute. - args: request.arguments, - approval: { - actionId, - allowed_decisions: reviewConfig?.allowed_decisions ?? [], - remember_scope: reviewConfig?.remember_scope, - remember_unavailable: reviewConfig?.remember_unavailable, - description: request.description, - ...(reviewConfig?.allow_always === true && { allow_always: true }), - }, - }; - changed = true; - } - - if (!changed) { - return { part, changed: false }; - } - return { - part: { ...part, [ContentTypes.TOOL_CALL]: nextToolCall } as TMessageContentParts, - changed: true, - }; -} - -/** - * Maps a tool-approval pending action onto a message's tool-call content parts, - * including tool calls nested inside subagents (see {@link tagApprovalOnPart}). - * - * Returns a NEW message only when something changed (referential stability lets - * React bail out of needless re-renders); otherwise the original is returned. - */ -function applyToolApproval( - message: TMessage, - actionId: string, - payload: Agents.ToolApprovalInterruptPayload, -): TMessage { - const content = message.content; - if (!Array.isArray(content) || content.length === 0) { - return message; - } - - const reviewByToolCallId = new Map(); - for (const config of payload.review_configs) { - reviewByToolCallId.set(config.tool_call_id, config); - } - const requestByToolCallId = new Map(); - for (const requestItem of payload.action_requests) { - requestByToolCallId.set(requestItem.tool_call_id, requestItem); - } - - let changed = false; - const nextContent = content.map((part) => { - const res = tagApprovalOnPart(part, actionId, requestByToolCallId, reviewByToolCallId); - if (res.changed) { - changed = true; - } - return res.part; - }); - - if (!changed) { - return message; - } - return { ...message, content: nextContent }; -} - -/** - * Appends (or refreshes) an ask-user-question content part for the pending - * action. Idempotent: replaces an existing part with the same `actionId` rather - * than stacking duplicates on reconnect/replay. - */ -function applyAskUserQuestion( - message: TMessage, - actionId: string, - payload: Agents.AskUserQuestionInterruptPayload, -): TMessage { - const content = Array.isArray(message.content) ? message.content : []; - const askPart = { - type: ASK_USER_QUESTION, - [ASK_USER_QUESTION]: { - actionId, - question: payload.question, - ...(payload.questions != null && { questions: payload.questions }), - ...(payload.tool_call_id != null && { tool_call_id: payload.tool_call_id }), - }, - } as unknown as TMessageContentParts; - - const existingIdx = content.findIndex( - (part) => - isAskUserQuestionPart(part) && - (part as unknown as AskUserQuestionPart)[ASK_USER_QUESTION].actionId === actionId, - ); - if (existingIdx >= 0) { - const nextContent = [...content]; - nextContent[existingIdx] = askPart; - return { ...message, content: nextContent }; - } - return { ...message, content: [...content, askPart] }; -} - /** * Parse an `ask_user_question` tool call's args into the question request shape. * Args arrive as a JSON string on persisted messages (or an object mid-stream); @@ -484,98 +331,9 @@ export function collectLiveAskToolCallIds(messages: TMessage[] | null | undefine return { ids, hasUnattributed }; } -/** - * Applies a {@link Agents.PendingAction} onto the target response message, - * dispatching on the interrupt type. Pure — returns a new message only when the - * mapping actually changed something. - */ -export function applyPendingAction( - message: TMessage, - pendingAction: Agents.PendingAction, -): TMessage { - const { payload, actionId } = pendingAction; - if (payload.type === 'tool_approval') { - return applyToolApproval(message, actionId, payload); - } - if (payload.type === 'ask_user_question') { - return applyAskUserQuestion(message, actionId, payload); - } - return message; -} - -/** - * Counts the tool-call content parts already tagged with this action's approval — - * i.e. how many of a tool-approval pending action's `action_requests` have rendered - * and been mapped. A multi-tool pause can render its sibling cards across several - * frames, so the SSE retry compares this against `action_requests.length` to know - * whether EVERY paused call is tagged yet. - */ -export function countTaggedApprovalParts(message: TMessage, actionId: string): number { - const content = message.content; - if (!Array.isArray(content)) { - return 0; - } - const countIn = (parts: TMessageContentParts[]): number => { - let count = 0; - for (const part of parts) { - if (part?.type !== ContentTypes.TOOL_CALL) { - continue; - } - const toolCall = part[ContentTypes.TOOL_CALL] as ToolCallWithApproval | undefined; - if ((toolCall?.approval as { actionId?: string } | undefined)?.actionId === actionId) { - count += 1; - } - // Nested subagent tool calls count too, so the retry loop's "all tagged" check - // is reachable for a tool paused inside a subagent. - if (Array.isArray(toolCall?.subagent_content)) { - count += countIn(toolCall.subagent_content); - } - } - return count; - }; - return countIn(content); -} - /** Returns the ask-user-question synthetic part when `part` is one, else undefined. */ export function getAskUserQuestionPart( part: TMessageContentParts | undefined, ): AskUserQuestionPart | undefined { return isAskUserQuestionPart(part) ? (part as unknown as AskUserQuestionPart) : undefined; } - -/** - * Resolves the assistant response message a pending action targets within - * `messages`. Returns the index, or -1 when the assistant placeholder isn't present - * yet (the caller retries on the next frame). - * - * Only ever matches an ASSISTANT message. The `responseMessageId` for a fresh turn - * is the user message id with a trailing underscore (`_`), so a naive - * underscore-strip would resolve to the just-created USER message before the - * assistant placeholder exists — appending the prompt to the wrong bubble and never - * triggering the retry. Matching strictly on assistant messages avoids that. - */ -export function findPendingActionMessageIndex( - messages: TMessage[], - pendingAction: Agents.PendingAction, -): number { - const isAssistant = (message: TMessage | undefined) => message?.isCreatedByUser === false; - const { responseMessageId } = pendingAction; - if (responseMessageId) { - // When the id is provided, ONLY an exact assistant match counts. A miss means the - // assistant placeholder for this turn hasn't been inserted yet — return -1 so the - // caller retries on the next frame. Falling back to the last assistant here would - // attach the prompt/approval to a PRIOR reply (applyAskUserQuestion always appends), - // and the retry would never run. The id is the in-flight response, so once it renders - // the retry resolves it. - return messages.findIndex( - (message) => message.messageId === responseMessageId && isAssistant(message), - ); - } - /** No responseMessageId: best-effort to the last assistant (the in-flight placeholder). */ - for (let i = messages.length - 1; i >= 0; i--) { - if (isAssistant(messages[i])) { - return i; - } - } - return -1; -} diff --git a/client/src/utils/messages.ts b/client/src/utils/messages.ts index 9e085ae4754..0236e588d38 100644 --- a/client/src/utils/messages.ts +++ b/client/src/utils/messages.ts @@ -99,34 +99,6 @@ export const getMessageBranchSiblingParentIds = ( return Array.from(parentIds); }; -/** - * True when a resumed run's user-message slot is a compaction ANCHOR rather than - * a turn the run created. - * - * A compaction submits no user turn: the slot names the LEAF it summarizes up - * to, which the server projects as identity only (`projectCompactionAnchor` — - * id, conversation, empty text, and no parent). Every other run publishes the - * turn it created, parent included (`getPreliminaryUserMessage`), so the absent - * parent is what separates the two. - * - * The anchor's author cannot separate them: Compact runs on whatever leaf the - * branch ends with and `canCompact` does not restrict its author, so the anchor - * is an assistant answer on one branch and a user message on the next (see - * `isUserInitiatedCompaction`). Testing for a non-user row recognized only the - * first kind and left the second rewritten on every re-attach. - */ -export const isCompactionAnchorProjection = ( - userMessage?: { - messageId?: string; - parentMessageId?: string | null; - text?: string | null; - } | null, -): boolean => - userMessage?.messageId != null && - userMessage.messageId !== '' && - userMessage.parentMessageId == null && - (userMessage.text == null || userMessage.text === ''); - export const getBranchSiblingIndexesForTarget = ( messages: TMessage[] | null | undefined, targetMessageId: string | null | undefined, diff --git a/packages/chat/src/core/__tests__/resume.spec.ts b/packages/chat/src/core/__tests__/resume.spec.ts new file mode 100644 index 00000000000..bcc7fd1d60c --- /dev/null +++ b/packages/chat/src/core/__tests__/resume.spec.ts @@ -0,0 +1,198 @@ +import { Constants, StepEvents, tMessageSchema } from 'librechat-data-provider'; +import type { Agents, TMessage, TSubmission } from 'librechat-data-provider'; +import { + getStepEventId, + isOAuthStepEvent, + mergeResumeMessages, + getResumeMessageIndexes, + hasSubmissionUserMessage, + buildResumeEventSubmission, + completedResponseMessageId, + isCompactionAnchorProjection, + buildMinimalResumeSubmission, + resumeStateMatchesSubmission, + buildSubmissionFromResumeState, + getResumeBranchTargetMessageId, +} from '../resumable'; + +const message = (fields: Partial) => + ({ conversationId: 'c-1', text: '', ...fields }) as TMessage; + +const user = message({ + messageId: 'u-1', + parentMessageId: Constants.NO_PARENT as string, + isCreatedByUser: true, +}); +const response = message({ messageId: 'r-1', parentMessageId: 'u-1', isCreatedByUser: false }); + +const resumeState = (fields: Partial = {}) => + ({ + runSteps: [], + aggregatedContent: [], + userMessage: { + messageId: 'u-1', + parentMessageId: Constants.NO_PARENT as string, + text: 'Hello', + }, + responseMessageId: 'r-1', + conversationId: 'c-1', + ...fields, + }) as Agents.ResumeState; + +describe('isCompactionAnchorProjection', () => { + it('recognizes an identity-only slot, whoever authored it', () => { + expect(isCompactionAnchorProjection({ messageId: 'leaf' })).toBe(true); + expect(isCompactionAnchorProjection({ messageId: 'leaf', parentMessageId: 'p' })).toBe(false); + expect(isCompactionAnchorProjection({ messageId: 'leaf', text: 'hi' })).toBe(false); + expect(isCompactionAnchorProjection(null)).toBe(false); + }); +}); + +describe('buildMinimalResumeSubmission', () => { + const build = (messages: TMessage[]) => + buildMinimalResumeSubmission({ + messages, + streamId: 'c-1', + createdAt: 2000, + isTemporary: false, + conversationId: 'c-1', + generationProtocolVersion: 1, + }); + + it('parents an assistant placeholder under the last user turn', () => { + const submission = build([user, response]); + expect(submission.userMessage).toBe(user); + expect(submission.initialResponse).toMatchObject({ + messageId: 'resume_', + parentMessageId: 'u-1', + isCreatedByUser: false, + }); + expect(submission).toMatchObject({ resumeStreamId: 'c-1', resumeGenerationCreatedAt: 2000 }); + }); + + it('builds a placeholder an error turn can parse when history holds no user turn', () => { + const submission = build([]); + expect(submission.userMessage).toMatchObject({ messageId: 'resume', isCreatedByUser: true }); + expect(() => + tMessageSchema.parse({ ...submission.initialResponse, error: true }), + ).not.toThrow(); + }); +}); + +describe('buildSubmissionFromResumeState', () => { + it('strips the persisted pair a fresh resume re-supplies', () => { + const submission = buildSubmissionFromResumeState( + resumeState(), + 'c-1', + [user, response], + 'c-1', + 2000, + 2, + ); + expect(submission.messages).toEqual([]); + expect(submission.isRegenerate).toBe(false); + expect(submission.initialResponse).toMatchObject({ messageId: 'r-1', parentMessageId: 'u-1' }); + expect(submission).toMatchObject({ + resumeStreamId: 'c-1', + resumeGenerationCreatedAt: 2000, + resumeGenerationProtocolVersion: 2, + }); + }); + + it('keeps an anchored run on the row its slot names', () => { + const leaf = message({ messageId: 'leaf', parentMessageId: 'u-1', isCreatedByUser: false }); + const submission = buildSubmissionFromResumeState( + resumeState({ userMessage: { messageId: 'leaf' }, responseMessageId: 'sum-1' }), + 'c-1', + [user, leaf], + 'c-1', + ); + expect(submission.userMessage).toBe(leaf); + expect(submission.isRegenerate).toBe(true); + expect(submission.compact).toBe(true); + }); +}); + +describe('resume matching', () => { + it('matches a submission to resume state by either message id', () => { + const submission = { userMessage: user, initialResponse: response } as TSubmission; + expect(resumeStateMatchesSubmission({ resumeState: resumeState() }, submission)).toBe(true); + expect( + resumeStateMatchesSubmission( + { resumeState: resumeState({ userMessage: { messageId: 'x' } }) }, + submission, + ), + ).toBe(true); + expect(resumeStateMatchesSubmission(undefined, submission)).toBe(false); + expect(hasSubmissionUserMessage(submission, [user], 'c-1')).toBe(true); + expect(hasSubmissionUserMessage(submission, [user], 'c-2')).toBe(false); + }); + + it('targets the padded response, then the persisted one, then the user parent', () => { + const padded = resumeState({ responseMessageId: 'r-1_' }); + expect(getResumeBranchTargetMessageId(padded, [message({ messageId: 'r-1_' })])).toBe('r-1_'); + expect(getResumeBranchTargetMessageId(padded, [response])).toBe('r-1'); + expect(getResumeBranchTargetMessageId(padded, [])).toBe(Constants.NO_PARENT); + }); +}); + +describe('resume message merge', () => { + it('adopts the resumed identity onto the current user turn', () => { + const submission = buildResumeEventSubmission( + { userMessage: user, initialResponse: response, conversation: {} } as TSubmission, + user, + resumeState({ conversationId: 'c-2' }), + ); + expect(submission.userMessage).toMatchObject({ messageId: 'u-1', conversationId: 'c-2' }); + expect(submission.initialResponse).toMatchObject({ + messageId: 'r-1', + parentMessageId: 'u-1', + isCreatedByUser: false, + }); + }); + + it('replaces the preliminary pair with the resumed one', () => { + const preliminaryUser = message({ messageId: 'tmp', isCreatedByUser: true }); + const preliminaryResponse = message({ messageId: 'tmp_', parentMessageId: 'tmp' }); + const messages = [preliminaryUser, preliminaryResponse]; + const indexes = getResumeMessageIndexes(messages, 'u-1', 'r-1', 'tmp', 'tmp_'); + expect(indexes).toEqual({ + userIndex: -1, + responseIndex: -1, + preliminaryUserIndex: 0, + preliminaryResponseIndex: 1, + }); + expect(mergeResumeMessages(messages, user, response, indexes).map((m) => m.messageId)).toEqual([ + 'u-1', + 'r-1', + ]); + }); + + it('identifies a completed response only when it is unambiguous', () => { + const sibling = message({ messageId: 'r-2', parentMessageId: 'u-1', isCreatedByUser: false }); + expect(completedResponseMessageId([user, response], 'u-1', 'u-1')).toBe('r-1'); + expect(completedResponseMessageId([user, response, sibling], 'u-1', 'u-1')).toBeUndefined(); + expect(completedResponseMessageId([user, response, sibling], 'u-1', 'r-2_')).toBe('r-2'); + }); +}); + +describe('step events', () => { + it('flags the OAuth steps and reads their ids', () => { + const oauthStep = { + event: StepEvents.ON_RUN_STEP, + data: { + id: 'step-1', + stepDetails: { tool_calls: [{ name: `oauth${Constants.mcp_delimiter}github` }] }, + }, + }; + expect(isOAuthStepEvent(oauthStep)).toBe(true); + expect(getStepEventId(oauthStep)).toBe('step-1'); + expect( + isOAuthStepEvent({ event: StepEvents.ON_RUN_STEP_DELTA, data: { delta: { auth: 'url' } } }), + ).toBe(true); + expect(isOAuthStepEvent({ event: StepEvents.ON_RUN_STEP, data: { stepDetails: {} } })).toBe( + false, + ); + expect(getStepEventId({ data: { result: { id: 'step-2' } } })).toBe('step-2'); + }); +}); diff --git a/packages/chat/src/core/index.ts b/packages/chat/src/core/index.ts index 8950a72d072..0644676ea76 100644 --- a/packages/chat/src/core/index.ts +++ b/packages/chat/src/core/index.ts @@ -1,4 +1,5 @@ export * from './ask'; +export * from './pending'; export * from './decision'; export * from './resumable'; export * from './steps'; diff --git a/packages/chat/src/core/pending.ts b/packages/chat/src/core/pending.ts new file mode 100644 index 00000000000..3c0f63984df --- /dev/null +++ b/packages/chat/src/core/pending.ts @@ -0,0 +1,253 @@ +import { ContentTypes } from 'librechat-data-provider'; +import type { Agents, TMessage, TMessageContentParts } from 'librechat-data-provider'; +import type { AskUserQuestionPart } from './ask'; +import { ASK_USER_QUESTION, isAskUserQuestionPart } from './ask'; + +const getToolCallId = (part: TMessageContentParts | undefined): string => + part?.type === ContentTypes.TOOL_CALL + ? ((part[ContentTypes.TOOL_CALL] as Agents.ToolCall | undefined)?.id ?? '') + : ''; + +/** A tool-call value that may carry approval state and/or nested subagent content. */ +type ToolCallWithApproval = Agents.ToolCall & { + approval?: unknown; + subagent_content?: TMessageContentParts[]; +}; + +/** + * Tags one tool-call part with the pending action's approval when it matches an + * `action_request` (joined by `tool_call_id`, NOT position) and is still unresolved. + * + * Recurses into a subagent's `subagent_content`: a tool paused INSIDE a subagent + * lives there, not as a top-level part, so without this the approval never attaches + * and the user gets no controls. Returns a NEW part only when something changed. + */ +function tagApprovalOnPart( + part: TMessageContentParts, + actionId: string, + requestByToolCallId: Map, + reviewByToolCallId: Map, +): { part: TMessageContentParts; changed: boolean } { + if (part?.type !== ContentTypes.TOOL_CALL) { + return { part, changed: false }; + } + const toolCall = part[ContentTypes.TOOL_CALL] as ToolCallWithApproval | undefined; + if (!toolCall) { + return { part, changed: false }; + } + + let nextToolCall = toolCall; + let changed = false; + + // Descend into nested subagent tool calls first. + if (Array.isArray(toolCall.subagent_content) && toolCall.subagent_content.length > 0) { + let nestedChanged = false; + const nextNested = toolCall.subagent_content.map((nestedPart) => { + const res = tagApprovalOnPart(nestedPart, actionId, requestByToolCallId, reviewByToolCallId); + if (res.changed) { + nestedChanged = true; + } + return res.part; + }); + if (nestedChanged) { + nextToolCall = { ...nextToolCall, subagent_content: nextNested }; + changed = true; + } + } + + // Tag this call itself when it's one of the paused requests and still unresolved + // (an `output` means the pause already resolved, so leave it alone). + const toolCallId = getToolCallId(part); + const request = toolCallId ? requestByToolCallId.get(toolCallId) : undefined; + if (request && (nextToolCall.output?.length ?? 0) === 0) { + const reviewConfig = reviewByToolCallId.get(toolCallId); + nextToolCall = { + ...nextToolCall, + // A PreToolUse hook may replace the model's original args before asking. + // The interrupt payload is authoritative so the reviewer sees, edits, and + // approves the same arguments the resumed tool will actually execute. + args: request.arguments, + approval: { + actionId, + allowed_decisions: reviewConfig?.allowed_decisions ?? [], + remember_scope: reviewConfig?.remember_scope, + remember_unavailable: reviewConfig?.remember_unavailable, + description: request.description, + ...(reviewConfig?.allow_always === true && { allow_always: true }), + }, + }; + changed = true; + } + + if (!changed) { + return { part, changed: false }; + } + return { + part: { ...part, [ContentTypes.TOOL_CALL]: nextToolCall } as TMessageContentParts, + changed: true, + }; +} + +/** + * Maps a tool-approval pending action onto a message's tool-call content parts, + * including tool calls nested inside subagents (see {@link tagApprovalOnPart}). + * + * Returns a NEW message only when something changed (referential stability lets + * React bail out of needless re-renders); otherwise the original is returned. + */ +function applyToolApproval( + message: TMessage, + actionId: string, + payload: Agents.ToolApprovalInterruptPayload, +): TMessage { + const content = message.content; + if (!Array.isArray(content) || content.length === 0) { + return message; + } + + const reviewByToolCallId = new Map(); + for (const config of payload.review_configs) { + reviewByToolCallId.set(config.tool_call_id, config); + } + const requestByToolCallId = new Map(); + for (const requestItem of payload.action_requests) { + requestByToolCallId.set(requestItem.tool_call_id, requestItem); + } + + let changed = false; + const nextContent = content.map((part) => { + const res = tagApprovalOnPart(part, actionId, requestByToolCallId, reviewByToolCallId); + if (res.changed) { + changed = true; + } + return res.part; + }); + + if (!changed) { + return message; + } + return { ...message, content: nextContent }; +} + +/** + * Appends (or refreshes) an ask-user-question content part for the pending + * action. Idempotent: replaces an existing part with the same `actionId` rather + * than stacking duplicates on reconnect/replay. + */ +function applyAskUserQuestion( + message: TMessage, + actionId: string, + payload: Agents.AskUserQuestionInterruptPayload, +): TMessage { + const content = Array.isArray(message.content) ? message.content : []; + const askPart = { + type: ASK_USER_QUESTION, + [ASK_USER_QUESTION]: { + actionId, + question: payload.question, + ...(payload.questions != null && { questions: payload.questions }), + ...(payload.tool_call_id != null && { tool_call_id: payload.tool_call_id }), + }, + } as unknown as TMessageContentParts; + + const existingIdx = content.findIndex( + (part) => + isAskUserQuestionPart(part) && + (part as unknown as AskUserQuestionPart)[ASK_USER_QUESTION].actionId === actionId, + ); + if (existingIdx >= 0) { + const nextContent = [...content]; + nextContent[existingIdx] = askPart; + return { ...message, content: nextContent }; + } + return { ...message, content: [...content, askPart] }; +} + +/** + * Applies a {@link Agents.PendingAction} onto the target response message, + * dispatching on the interrupt type. Pure: returns a new message only when the + * mapping actually changed something. + */ +export function applyPendingAction( + message: TMessage, + pendingAction: Agents.PendingAction, +): TMessage { + const { payload, actionId } = pendingAction; + if (payload.type === 'tool_approval') { + return applyToolApproval(message, actionId, payload); + } + if (payload.type === 'ask_user_question') { + return applyAskUserQuestion(message, actionId, payload); + } + return message; +} + +/** + * Counts the tool-call content parts already tagged with this action's approval, + * i.e. how many of a tool-approval pending action's `action_requests` have rendered + * and been mapped. A multi-tool pause can render its sibling cards across several + * frames, so the SSE retry compares this against `action_requests.length` to know + * whether EVERY paused call is tagged yet. + */ +export function countTaggedApprovalParts(message: TMessage, actionId: string): number { + const content = message.content; + if (!Array.isArray(content)) { + return 0; + } + const countIn = (parts: TMessageContentParts[]): number => { + let count = 0; + for (const part of parts) { + if (part?.type !== ContentTypes.TOOL_CALL) { + continue; + } + const toolCall = part[ContentTypes.TOOL_CALL] as ToolCallWithApproval | undefined; + if ((toolCall?.approval as { actionId?: string } | undefined)?.actionId === actionId) { + count += 1; + } + // Nested subagent tool calls count too, so the retry loop's "all tagged" check + // is reachable for a tool paused inside a subagent. + if (Array.isArray(toolCall?.subagent_content)) { + count += countIn(toolCall.subagent_content); + } + } + return count; + }; + return countIn(content); +} + +/** + * Resolves the assistant response message a pending action targets within + * `messages`. Returns the index, or -1 when the assistant placeholder isn't present + * yet (the caller retries on the next frame). + * + * Only ever matches an ASSISTANT message. The `responseMessageId` for a fresh turn + * is the user message id with a trailing underscore (`_`), so a naive + * underscore-strip would resolve to the just-created USER message before the + * assistant placeholder exists, appending the prompt to the wrong bubble and never + * triggering the retry. Matching strictly on assistant messages avoids that. + */ +export function findPendingActionMessageIndex( + messages: TMessage[], + pendingAction: Agents.PendingAction, +): number { + const isAssistant = (message: TMessage | undefined) => message?.isCreatedByUser === false; + const { responseMessageId } = pendingAction; + if (responseMessageId) { + // When the id is provided, ONLY an exact assistant match counts. A miss means the + // assistant placeholder for this turn hasn't been inserted yet; return -1 so the + // caller retries on the next frame. Falling back to the last assistant here would + // attach the prompt/approval to a PRIOR reply (applyAskUserQuestion always appends), + // and the retry would never run. The id is the in-flight response, so once it renders + // the retry resolves it. + return messages.findIndex( + (message) => message.messageId === responseMessageId && isAssistant(message), + ); + } + /** No responseMessageId: best-effort to the last assistant (the in-flight placeholder). */ + for (let i = messages.length - 1; i >= 0; i--) { + if (isAssistant(messages[i])) { + return i; + } + } + return -1; +} diff --git a/packages/chat/src/core/resumable/index.ts b/packages/chat/src/core/resumable/index.ts index c05bda03293..d519a3e6047 100644 --- a/packages/chat/src/core/resumable/index.ts +++ b/packages/chat/src/core/resumable/index.ts @@ -1,4 +1,5 @@ export * from './start'; export * from './errors'; export * from './events'; +export * from './resume'; export * from './messages'; diff --git a/packages/chat/src/core/resumable/resume.ts b/packages/chat/src/core/resumable/resume.ts new file mode 100644 index 00000000000..9d058374d66 --- /dev/null +++ b/packages/chat/src/core/resumable/resume.ts @@ -0,0 +1,522 @@ +import { Constants, tMessageSchema } from 'librechat-data-provider'; +import type { + Agents, + TMessage, + TSubmission, + TConversation, + EventSubmission, +} from 'librechat-data-provider'; +import type { GenerationProtocolVersion } from '../protocol'; +import { preferDefinedString } from './messages'; +import { applyPendingAction } from '../pending'; + +/** A submission that attaches to a generation already running on the server instead of + * starting one. */ +export type ResumeSubmission = TSubmission & { + resumeStreamId: string; + resumeGenerationCreatedAt?: number; + resumeGenerationProtocolVersion: GenerationProtocolVersion; +}; + +/** + * True when a resumed run's user-message slot is a compaction ANCHOR rather than + * a turn the run created. + * + * A compaction submits no user turn: the slot names the LEAF it summarizes up + * to, which the server projects as identity only (`projectCompactionAnchor`: + * id, conversation, empty text, and no parent). Every other run publishes the + * turn it created, parent included (`getPreliminaryUserMessage`), so the absent + * parent is what separates the two. + * + * The anchor's author cannot separate them: Compact runs on whatever leaf the + * branch ends with and `canCompact` does not restrict its author, so the anchor + * is an assistant answer on one branch and a user message on the next (see + * `isUserInitiatedCompaction`). Testing for a non-user row recognized only the + * first kind and left the second rewritten on every re-attach. + */ +export const isCompactionAnchorProjection = ( + userMessage?: { + messageId?: string; + parentMessageId?: string | null; + text?: string | null; + } | null, +): boolean => + userMessage?.messageId != null && + userMessage.messageId !== '' && + userMessage.parentMessageId == null && + (userMessage.text == null || userMessage.text === ''); + +export const buildResumeEventSubmission = ( + currentSubmission: TSubmission, + currentUserMessage: TMessage, + resumeState?: Agents.ResumeState | null, +): EventSubmission => { + if (!resumeState) { + return { ...currentSubmission, userMessage: currentUserMessage } as EventSubmission; + } + + const conversationId = + resumeState.conversationId ?? + resumeState.userMessage?.conversationId ?? + currentUserMessage.conversationId ?? + currentSubmission.conversation?.conversationId; + + /** + * A compaction submits no user turn: its user-message slot names the LEAF it + * summarizes up to, and the server projects that anchor as identity only + * (`projectCompactionAnchor`: no parent, no author). Adopting the projection + * as a user row would rewrite the leaf into an empty, parentless message, so + * an anchored run keeps the identity it resumed with. + */ + /** Read from the projection as well as the flag: a re-attach whose submission + * never learned it was a compaction still has to recognize the anchor. */ + const anchoredUserMessage = + currentSubmission.compact === true || isCompactionAnchorProjection(resumeState.userMessage); + + const userMessage = ( + anchoredUserMessage + ? { ...currentUserMessage, conversationId } + : { + ...currentUserMessage, + ...resumeState.userMessage, + conversationId, + isCreatedByUser: true, + } + ) as TMessage; + + const responseMessageId = + resumeState.responseMessageId ?? + currentSubmission.initialResponse?.messageId ?? + `${userMessage.messageId}_`; + + const initialResponse = { + ...(currentSubmission.initialResponse as TMessage), + messageId: responseMessageId, + parentMessageId: userMessage.messageId, + conversationId, + content: + resumeState.aggregatedContent ?? + (currentSubmission.initialResponse as TMessage | undefined)?.content, + sender: resumeState.sender ?? currentSubmission.initialResponse?.sender, + iconURL: preferDefinedString(currentSubmission.initialResponse?.iconURL, resumeState.iconURL), + model: preferDefinedString(currentSubmission.initialResponse?.model, resumeState.model), + isCreatedByUser: false, + } as TMessage; + + return { + ...currentSubmission, + conversation: { + ...currentSubmission.conversation, + conversationId, + }, + userMessage, + initialResponse, + /** Carried so every consumer of the resumed submission (the merges below, + * the abort-error write) reads the same answer this resolved. */ + ...(anchoredUserMessage && { compact: true }), + } as EventSubmission; +}; + +export type ResumeMessageIndexes = { + userIndex: number; + responseIndex: number; + preliminaryUserIndex: number; + preliminaryResponseIndex: number; +}; + +export const getResumeMessageIndexes = ( + messages: TMessage[], + userMessageId: string, + responseMessageId: string, + preliminaryUserMessageId?: string, + preliminaryResponseMessageId?: string, +): ResumeMessageIndexes => { + let userIndex = -1; + let responseIndex = -1; + let preliminaryUserIndex = -1; + let preliminaryResponseIndex = -1; + const hasPreliminaryResponse = preliminaryResponseMessageId?.endsWith('_') === true; + const eligiblePreliminaryUserId = + hasPreliminaryResponse && preliminaryUserMessageId && preliminaryUserMessageId !== userMessageId + ? preliminaryUserMessageId + : undefined; + const eligiblePreliminaryResponseId = + hasPreliminaryResponse && preliminaryResponseMessageId !== responseMessageId + ? preliminaryResponseMessageId + : undefined; + + for (let index = 0; index < messages.length; index++) { + const messageId = messages[index]?.messageId; + if (userIndex < 0 && messageId === userMessageId) { + userIndex = index; + } + if (responseIndex < 0 && messageId === responseMessageId) { + responseIndex = index; + } + if ( + preliminaryUserIndex < 0 && + eligiblePreliminaryUserId && + messageId === eligiblePreliminaryUserId + ) { + preliminaryUserIndex = index; + } + if ( + preliminaryResponseIndex < 0 && + eligiblePreliminaryResponseId && + messageId === eligiblePreliminaryResponseId + ) { + preliminaryResponseIndex = index; + } + } + + return { userIndex, responseIndex, preliminaryUserIndex, preliminaryResponseIndex }; +}; + +export const mergeResumeMessages = ( + messages: TMessage[], + userMessage: TMessage, + responseMessage: TMessage, + indexes: ResumeMessageIndexes, + /** + * An anchored run (a compaction) owns no user row. Its user-message slot + * holds the leaf the summary hangs off, a row the transcript already has: + * merging the slot onto it would replace that answer with an empty user + * message, and inserting it would duplicate its id. Either way the row loses + * its parent and `buildTree` files it as a phantom root, folding the thread. + */ + anchoredUserMessage = false, +): TMessage[] => { + const nextMessages = [...messages]; + let { userIndex, responseIndex, preliminaryResponseIndex } = indexes; + const { preliminaryUserIndex } = indexes; + + if (preliminaryUserIndex >= 0 && !anchoredUserMessage) { + if (userIndex >= 0) { + nextMessages.splice(preliminaryUserIndex, 1); + if (userIndex > preliminaryUserIndex) { + userIndex -= 1; + } + if (responseIndex > preliminaryUserIndex) { + responseIndex -= 1; + } + if (preliminaryResponseIndex > preliminaryUserIndex) { + preliminaryResponseIndex -= 1; + } + } else { + nextMessages[preliminaryUserIndex] = { + ...nextMessages[preliminaryUserIndex], + ...userMessage, + }; + userIndex = preliminaryUserIndex; + } + } + + if (preliminaryResponseIndex >= 0) { + if (responseIndex >= 0) { + nextMessages.splice(preliminaryResponseIndex, 1); + if (userIndex > preliminaryResponseIndex) { + userIndex -= 1; + } + if (responseIndex > preliminaryResponseIndex) { + responseIndex -= 1; + } + } else { + nextMessages[preliminaryResponseIndex] = { + ...nextMessages[preliminaryResponseIndex], + ...responseMessage, + }; + responseIndex = preliminaryResponseIndex; + } + } + + if (userIndex >= 0 && !anchoredUserMessage) { + nextMessages[userIndex] = { ...nextMessages[userIndex], ...userMessage }; + } + + if (responseIndex >= 0) { + nextMessages[responseIndex] = { ...nextMessages[responseIndex], ...responseMessage }; + } + + if (anchoredUserMessage) { + return responseIndex >= 0 ? nextMessages : [...nextMessages, responseMessage]; + } + + if (userIndex >= 0 && responseIndex >= 0) { + return nextMessages; + } + + if (userIndex >= 0) { + return [...nextMessages, responseMessage]; + } + + if (responseIndex >= 0) { + nextMessages.splice(responseIndex, 0, userMessage); + return nextMessages; + } + + return [...nextMessages, userMessage, responseMessage]; +}; + +export 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, + ); +} + +export function resumeStateMatchesSubmission( + streamStatus: { resumeState?: Agents.ResumeState } | 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; +} + +/** + * The row that names the branch a resumed run belongs to when its own response + * is not in the loaded history yet. A compaction's user-message slot is the leaf + * it summarizes up to and carries no parent (`projectCompactionAnchor`), so + * there the anchor itself names the branch; without this the pane restores no + * sibling selection and a compaction started on an older branch comes back on + * whichever branch the default lands on. + */ +function getResumeBranchFallbackMessageId( + resumeState: Agents.ResumeState, +): string | null | undefined { + return resumeState.userMessage?.parentMessageId ?? resumeState.userMessage?.messageId; +} + +export function getResumeBranchTargetMessageId( + resumeState: Agents.ResumeState, + messages: TMessage[], +): string | null | undefined { + const responseMessageId = resumeState.responseMessageId; + if (!responseMessageId) { + return getResumeBranchFallbackMessageId(resumeState); + } + + const unpaddedResponseMessageId = responseMessageId.replace(/_+$/, ''); + let hasResponseMessage = false; + let hasUnpaddedResponseMessage = false; + + for (const message of messages) { + if (message.messageId === responseMessageId) { + hasResponseMessage = true; + break; + } + + if (message.messageId === unpaddedResponseMessageId) { + hasUnpaddedResponseMessage = true; + } + } + + if (hasResponseMessage) { + return responseMessageId; + } + + if (hasUnpaddedResponseMessage) { + return unpaddedResponseMessageId; + } + + return getResumeBranchFallbackMessageId(resumeState); +} + +/** + * Build a submission object from resume state for reconnected streams. + * This provides the minimum data needed for useResumableSSE to subscribe. + */ +export function buildSubmissionFromResumeState( + resumeState: Agents.ResumeState, + streamId: string, + messages: TMessage[], + conversationId: string, + generationCreatedAt?: number, + generationProtocolVersion: GenerationProtocolVersion = 1, + isTemporary = false, +): ResumeSubmission { + const userMessageData = resumeState.userMessage; + const responseMessageId = + resumeState.responseMessageId ?? `${userMessageData?.messageId ?? 'resume'}_`; + + /** + * The run's user-message slot as the loaded history already holds it. Message + * ids are unique per row, so a match is that row whoever wrote it: a + * compaction submits no user turn and puts the LEAF it summarizes up to in + * this slot (`useCompactConversation` client-side, `projectCompactionAnchor` + * server-side), which is usually the assistant answer. Adopting the row keeps + * the anchor's own identity; synthesizing an empty, parentless USER row over + * it rewrites that answer into a phantom root and folds the thread. + */ + const existingSlotMessage = messages.find((m) => m.messageId === userMessageData?.messageId); + /** An anchored run (a compaction) created no user turn: the slot names a row + * the transcript already holds. Judged from the projection's shape, never from + * the anchor's author, which is a user message as often as an answer. Either + * way the run is regenerate-shaped for every consumer: no user turn of its + * own, the response parented onto an existing message. */ + const isAnchoredRun = isCompactionAnchorProjection(userMessageData); + + // A trailing underscore distinguishes an in-flight regeneration from the persisted + // response it replaces. Only the exact response id proves generation ownership. + const existingResponseMessage = messages.find( + (m) => !m.isCreatedByUser && m.messageId === responseMessageId, + ); + // The persisted row may seed display metadata, but never identity or deduplication. + const unpaddedResponseMessageId = responseMessageId.replace(/_+$/, ''); + const persistedRegenerationResponse = + unpaddedResponseMessageId !== responseMessageId + ? messages.find((m) => !m.isCreatedByUser && m.messageId === unpaddedResponseMessageId) + : undefined; + const responseMetadataMessage = existingResponseMessage ?? persistedRegenerationResponse; + const isRegenerateResume = + resumeState.isRegenerate === true || persistedRegenerationResponse != null || isAnchoredRun; + let regenerateMessages: TMessage[] | undefined; + if (isRegenerateResume) { + regenerateMessages = + unpaddedResponseMessageId === responseMessageId + ? [...messages] + : messages.filter((message) => message.messageId !== responseMessageId); + } + + // Create or use the row the slot already names + const userMessage: TMessage = + existingSlotMessage ?? + (userMessageData + ? (tMessageSchema.parse({ + messageId: userMessageData.messageId, + parentMessageId: userMessageData.parentMessageId ?? Constants.NO_PARENT, + conversationId: userMessageData.conversationId ?? conversationId, + text: userMessageData.text ?? '', + isCreatedByUser: true, + role: 'user', + }) as TMessage) + : (messages[messages.length - 2] ?? + ({ + messageId: 'resume_user_msg', + conversationId, + text: '', + isCreatedByUser: true, + } as TMessage))); + + // ALWAYS use aggregatedContent from resumeState - it has the latest content from the running job. + // DB content may be stale (saved at disconnect, but generation continued). + let initialResponse: TMessage = { + messageId: existingResponseMessage?.messageId ?? responseMessageId, + parentMessageId: existingResponseMessage?.parentMessageId ?? userMessage.messageId, + conversationId, + text: '', + // aggregatedContent is authoritative - it reflects actual job state + content: (resumeState.aggregatedContent as TMessage['content']) ?? [], + isCreatedByUser: false, + role: 'assistant', + sender: responseMetadataMessage?.sender ?? resumeState.sender, + model: preferDefinedString(responseMetadataMessage?.model, resumeState.model), + iconURL: preferDefinedString(responseMetadataMessage?.iconURL, resumeState.iconURL), + } as TMessage; + + // Re-paused turn: seed the approval / ask-user controls straight onto the + // placeholder so they render on load without waiting for the SSE sync replay. + if (resumeState.pendingAction) { + initialResponse = applyPendingAction(initialResponse, resumeState.pendingAction); + } + + const conversation: TConversation = { + conversationId, + title: 'Resumed Chat', + endpoint: null, + } as TConversation; + + // Non-regenerate resumes strip the persisted request/response pair before handlers + // re-supply it. A regeneration keeps the original branch for early-abort rollback; + // explicit resume metadata covers edited regenerations that reuse the exact response id. + const dedupedMessages = messages.filter( + (m) => + m.messageId !== initialResponse.messageId && + (isRegenerateResume || m.messageId !== userMessage.messageId), + ); + + return { + messages: dedupedMessages, + userMessage, + initialResponse, + conversation, + isRegenerate: isRegenerateResume, + ...(isAnchoredRun && { compact: true }), + ...(regenerateMessages && { regenerateMessages }), + isTemporary, + endpointOption: {}, + // Signal to useResumableSSE to subscribe to existing stream instead of starting new + resumeStreamId: streamId, + ...(generationCreatedAt != null && { resumeGenerationCreatedAt: generationCreatedAt }), + resumeGenerationProtocolVersion: generationProtocolVersion, + } as ResumeSubmission; +} + +/** + * The submission for a running generation whose status carries no resume state (an older + * server): the loaded history's last user turn and an empty placeholder response under it. + * The placeholder is a parented assistant row, so a failure the reattached stream reports + * still parses into the error turn. + */ +export function buildMinimalResumeSubmission({ + messages, + streamId, + createdAt, + isTemporary, + conversationId, + aggregatedContent, + generationProtocolVersion, +}: { + messages: TMessage[]; + streamId: string; + createdAt?: number; + isTemporary: boolean; + conversationId: string; + aggregatedContent?: Array<{ type: string; text?: string }>; + generationProtocolVersion: GenerationProtocolVersion; +}): ResumeSubmission { + const lastUserMessage = [...messages].reverse().find((m) => m.isCreatedByUser); + const userMessage = + lastUserMessage ?? + ({ messageId: 'resume', conversationId, text: '', isCreatedByUser: true } as TMessage); + return { + messages, + userMessage, + initialResponse: { + messageId: 'resume_', + parentMessageId: userMessage.messageId, + conversationId, + text: '', + content: aggregatedContent ?? [{ type: 'text', text: '' }], + isCreatedByUser: false, + } as TMessage, + conversation: { conversationId, title: 'Resumed Chat' } as TConversation, + isRegenerate: false, + isTemporary, + endpointOption: {}, + // Signal to useResumableSSE to subscribe to existing stream instead of starting new + resumeStreamId: streamId, + ...(createdAt != null && { resumeGenerationCreatedAt: createdAt }), + resumeGenerationProtocolVersion: generationProtocolVersion, + } as ResumeSubmission; +}