From 8f31f214235b247e9cfbb110edeae94eef5cc5a3 Mon Sep 17 00:00:00 2001 From: Marco Beretta <81851188+berry-13@users.noreply.github.com> Date: Thu, 15 Oct 2026 09:19:29 +0200 Subject: [PATCH] refactor: Split useEventHandlers by Responsibility --- .../SSE/__tests__/useEventHandlers.spec.ts | 2 +- client/src/hooks/SSE/handlers.ts | 47 + client/src/hooks/SSE/turns.ts | 281 ++++ client/src/hooks/SSE/useAbortHandlers.ts | 234 ++++ client/src/hooks/SSE/useEventHandlers.ts | 1151 +---------------- client/src/hooks/SSE/useFinalHandler.ts | 364 ++++++ client/src/hooks/SSE/useTurnHandlers.ts | 335 +++++ 7 files changed, 1299 insertions(+), 1115 deletions(-) create mode 100644 client/src/hooks/SSE/handlers.ts create mode 100644 client/src/hooks/SSE/turns.ts create mode 100644 client/src/hooks/SSE/useAbortHandlers.ts create mode 100644 client/src/hooks/SSE/useFinalHandler.ts create mode 100644 client/src/hooks/SSE/useTurnHandlers.ts diff --git a/client/src/hooks/SSE/__tests__/useEventHandlers.spec.ts b/client/src/hooks/SSE/__tests__/useEventHandlers.spec.ts index ddc2caeb0c0..d8fd6ea0cb0 100644 --- a/client/src/hooks/SSE/__tests__/useEventHandlers.spec.ts +++ b/client/src/hooks/SSE/__tests__/useEventHandlers.spec.ts @@ -9,8 +9,8 @@ import type { import type { TResData } from '~/common'; import { getFailedCodeDecisionRequest } from '~/hooks/Agents/codeDecision'; import { stripStreamedIndexStamps, getPartKeyIndex } from '~/utils'; -import { resolveErrorTurn } from '~/hooks/SSE/useEventHandlers'; import buildDefaultConvo from '~/utils/buildDefaultConvo'; +import { resolveErrorTurn } from '~/hooks/SSE/turns'; describe('resolveErrorTurn', () => { const userMessage = { diff --git a/client/src/hooks/SSE/handlers.ts b/client/src/hooks/SSE/handlers.ts new file mode 100644 index 00000000000..6695fa51b2c --- /dev/null +++ b/client/src/hooks/SSE/handlers.ts @@ -0,0 +1,47 @@ +import type { TMessage, EventSubmission } from 'librechat-data-provider'; +import type { Location, NavigateFunction } from 'react-router-dom'; +import type { QueryClient } from '@tanstack/react-query'; +import type { MutableRefObject } from 'react'; +import type { SetterOrUpdater } from 'recoil'; +import type useFocusRegeneratedResponse from '~/hooks/Chat/useFocusRegeneratedResponse'; +import type useAttachmentHandler from './useAttachmentHandler'; +import type { EventHandlerParams } from './useEventHandlers'; +import type { useApplyAgentTemplate } from '~/hooks/Agents'; +import type { useLiveAnnouncer } from '~/Providers'; + +/** What the stream event handlers share: the pane they write to and the app around it. */ +export type EventHandlerContext = Required< + Pick +> & + Pick< + EventHandlerParams, + | 'setMessages' + | 'getMessages' + | 'setCompleted' + | 'setConversation' + | 'setIsSubmitting' + | 'setShowStopButton' + > & { + queryClient: QueryClient; + setAbortScroll: SetterOrUpdater; + /** Clears the elapsed anchor on every terminal path. */ + setSubmissionStart: (start: number | null) => void; + announcePolite: ReturnType['announcePolite']; + lastAnnouncementTimeRef: MutableRefObject; + applyAgentTemplate: ReturnType; + focusRegeneratedResponse: ReturnType; + location: Location; + navigate: NavigateFunction; + paramId?: string; + /** Re-stages request context when an early abort restores the draft. */ + restorePendingContext: ( + convoId: string, + quotes?: string[], + reasoningOverride?: TMessage['reasoningOverride'], + ) => void; + reconcileFailedCodeDecision: (submission: EventSubmission, conversationId?: string) => void; + recoverConversation: (conversationId: string, submission: EventSubmission) => void; + attachmentHandler: ReturnType; + /** The saved id a new chat's own run moves to, so its subagent atoms survive the move. */ + preserveSubagentAtomsForNewConvoIdRef: MutableRefObject; + }; diff --git a/client/src/hooks/SSE/turns.ts b/client/src/hooks/SSE/turns.ts new file mode 100644 index 00000000000..5b503234cbe --- /dev/null +++ b/client/src/hooks/SSE/turns.ts @@ -0,0 +1,281 @@ +import { v4 } from 'uuid'; +import { useQueryClient } from '@tanstack/react-query'; +import { responseParentMessageId } from '@librechat/chat'; +import { QueryKeys, ContentTypes, tMessageSchema } from 'librechat-data-provider'; +import type { + TMessage, + TConversation, + EventSubmission, + TMessageContentParts, +} from 'librechat-data-provider'; +import type { ConvoTitleState } from '~/utils/convos'; +import type { TResData } from '~/common'; +import { + hasRealTitle, + findConvoInAllQueries, + findManualConvoTitleInAllQueries, + preserveStreamedContentIdentity, + isEmptyContentPart, + getPartKeyIndex, +} from '~/utils'; + +const CONNECTION_ERROR_TEXT = 'Error connecting to server, try refreshing the page.'; + +/** + * The parts the in-flight response has streamed so far: the transcript's tail (a user row there + * means no response was placed), without the holes an interrupted stream leaves. Whether anything + * streamed is judged without the slots that never received content (a comparison run's + * `type: ''` placeholders, a text or think part opened before its first delta), but once something + * did, those slots stay: a placeholder is what keeps a comparison lane's layout and attribution + * when only the other lane produced output. The parts carry the render identity they streamed + * under, as the final path stamps it, so the settled row does not remount. + */ +const getStreamedContent = (message?: TMessage): TMessageContentParts[] => { + if (message == null || message.isCreatedByUser === true) { + return []; + } + const streamed = message.content ?? []; + const parts = streamed.filter((part): part is TMessageContentParts => part != null); + if (!parts.some((part) => !isEmptyContentPart(part))) { + return []; + } + return preserveStreamedContentIdentity(streamed, parts) ?? parts; +}; + +/** Keys the appended failure past every key the kept parts render under, physical or stamped. */ +const appendErrorPart = ( + content: TMessageContentParts[], + errorText: string, +): TMessageContentParts[] => { + const keyIndex = content.reduce( + (next, part, idx) => Math.max(next, getPartKeyIndex(part, idx) + 1), + 0, + ); + const errorPart: TMessageContentParts = { type: ContentTypes.ERROR, error: errorText }; + return [ + ...content, + keyIndex === content.length ? errorPart : { ...errorPart, streamedIndex: keyIndex }, + ]; +}; + +/** + * A failure that lands after the run has streamed keeps what streamed and takes the failure as + * one more part, the way the server records a failure inside a run; one that lands before + * anything streamed is the whole row. + */ +export const createErrorMessage = ({ + errorMetadata, + getMessages, + submission, + error, +}: { + getMessages: () => TMessage[] | undefined; + errorMetadata?: Partial; + submission: EventSubmission; + error?: Error | unknown; +}): TMessage => { + const currentMessages = getMessages(); + const latestMessage = currentMessages?.[currentMessages.length - 1]; + const text = submission.initialResponse.text.length > 45 ? submission.initialResponse.text : ''; + const errorText = + (errorMetadata?.text || text || (error as Error | undefined)?.message) ?? + 'Error cancelling request'; + const streamedContent = getStreamedContent(latestMessage); + if (latestMessage?.conversationId && latestMessage.messageId && streamedContent.length > 0) { + /** The row keeps the envelope it streamed under (author, model, icon, creation time) and + * takes from the failure only what describes the failure: a server payload names `System` + * as its sender and the schema dates a fresh envelope now, neither of which applies to a + * response that is merely gaining a part. */ + const errorMessage: TMessage = { + ...latestMessage, + ...(errorMetadata?.metadata != null + ? { metadata: { ...latestMessage.metadata, ...errorMetadata.metadata } } + : {}), + error: undefined, + text: '', + content: appendErrorPart(streamedContent, errorText), + }; + const parentMessageId = responseParentMessageId( + submission.initialResponse, + submission.userMessage, + submission.isRegenerate, + ); + if (parentMessageId && parentMessageId !== errorMessage.parentMessageId) { + errorMessage.parentMessageId = parentMessageId; + } + return errorMessage; + } + if (errorMetadata) { + return errorMetadata as TMessage; + } + return tMessageSchema.parse({ + ...submission, + ...submission.initialResponse, + text: errorText, + unfinished: !!text.length, + error: true, + }) as TMessage; +}; + +export interface ErrorTurn { + conversationId: string; + errorResponse: TMessage; + /** The conversation must be rebuilt: it never learned its id, the transport failed, or the + * server named a conversation the new-chat route has not navigated to. */ + recover: boolean; +} + +/** Builds the row a failed turn leaves in the transcript and names the conversation it belongs to. */ +type SettledErrorReadState = { + lastResponseAt: string; + lastResponseMessageId?: string; + updatedAt?: string; +}; + +/** + * A persisted terminal error still carries the server's winning reply stamp in its nested + * conversation snapshot. Pairing the error messages cache object with that stamp before the list + * update lets `useConversationSeen` acknowledge the reply the user just watched fail, instead of + * leaving a dot on a conversation they are looking at. + */ +export const resolveSettledErrorReadState = ({ + data, + submission, + conversationId, +}: { + data?: TResData; + submission: EventSubmission; + conversationId: string; +}): SettledErrorReadState | undefined => { + if (submission.isTemporary === true || !data || !conversationId) { + return undefined; + } + const settledConversation = data.conversation; + const lastResponseAt = settledConversation?.lastResponseAt; + if (typeof lastResponseAt !== 'string' || lastResponseAt.length === 0) { + return undefined; + } + return { + lastResponseAt, + lastResponseMessageId: + typeof settledConversation.lastResponseMessageId === 'string' + ? settledConversation.lastResponseMessageId + : undefined, + updatedAt: + typeof settledConversation.updatedAt === 'string' ? settledConversation.updatedAt : undefined, + }; +}; + +export const resolveErrorTurn = ({ + data, + submission, + getMessages, + isNewConversationRoute, +}: { + data?: TResData; + submission: EventSubmission; + getMessages: () => TMessage[] | undefined; + isNewConversationRoute: boolean; +}): ErrorTurn => { + const { userMessage, initialResponse } = submission; + const parentMessageId = responseParentMessageId( + initialResponse, + userMessage, + submission.isRegenerate, + ); + const conversationId = + userMessage.conversationId ?? submission.conversation?.conversationId ?? ''; + + const parseErrorResponse = (payload: TResData | Partial): TMessage => { + const metadata = payload['responseMessage'] ?? payload; + const errorMessage: Partial = { + ...initialResponse, + ...metadata, + error: true, + parentMessageId, + }; + + if (errorMessage.messageId === undefined || errorMessage.messageId === '') { + errorMessage.messageId = v4(); + } + + return tMessageSchema.parse(errorMessage) as TMessage; + }; + const build = (errorMetadata: TMessage): TMessage => + createErrorMessage({ errorMetadata, getMessages, submission }); + + if (!data) { + const convoId = conversationId || `_${v4()}`; + return { + conversationId: convoId, + recover: true, + errorResponse: build( + parseErrorResponse({ text: CONNECTION_ERROR_TEXT, ...submission, conversationId: convoId }), + ), + }; + } + + const receivedConvoId = data.conversationId ?? ''; + if (!conversationId && !receivedConvoId) { + return { + conversationId: `_${v4()}`, + recover: true, + errorResponse: build(parseErrorResponse(data)), + }; + } + if (!receivedConvoId) { + return { conversationId, recover: false, errorResponse: build(parseErrorResponse(data)) }; + } + return { + conversationId: receivedConvoId, + recover: isNewConversationRoute, + errorResponse: build( + tMessageSchema.parse({ + ...data, + error: true, + parentMessageId, + }) as TMessage, + ), + }; +}; + +export const getConvoTitle = ({ + queryClient, + currentTitle, + conversationId, + titleSetByUser, + titleRevision, + authoritative = false, +}: { + queryClient: ReturnType; + currentTitle?: string | null; + conversationId?: string | null; + titleSetByUser?: boolean; + titleRevision?: number; + authoritative?: boolean; +}): ConvoTitleState => { + const incoming = { title: currentTitle ?? null, titleSetByUser, titleRevision }; + if (!conversationId) { + return incoming; + } + const cached = queryClient.getQueryData([QueryKeys.conversation, conversationId]); + const owned = findManualConvoTitleInAllQueries(queryClient, conversationId, incoming); + const listed = owned ? undefined : findConvoInAllQueries(queryClient, conversationId); + let selected = owned; + if (!selected) { + if (authoritative && hasRealTitle(incoming.title)) { + selected = incoming; + } else if (hasRealTitle(cached?.title)) { + selected = cached; + } else if (hasRealTitle(listed?.title)) { + selected = listed; + } else { + selected = incoming; + } + } + return { + title: selected?.title ?? null, + titleSetByUser: selected?.titleSetByUser, + titleRevision: selected?.titleRevision, + }; +}; diff --git a/client/src/hooks/SSE/useAbortHandlers.ts b/client/src/hooks/SSE/useAbortHandlers.ts new file mode 100644 index 00000000000..845cf915b40 --- /dev/null +++ b/client/src/hooks/SSE/useAbortHandlers.ts @@ -0,0 +1,234 @@ +import { useCallback } from 'react'; +import { v4 } from 'uuid'; +import { mergeErrorMessages, keepLocalCodeApprovalMode } from '@librechat/chat'; +import { QueryKeys, Constants, isAssistantsEndpoint } from 'librechat-data-provider'; +import type { TMessage, TConversation, EventSubmission } from 'librechat-data-provider'; +import type { TResData, TFinalResData } from '~/common'; +import type { EventHandlerContext } from './handlers'; +import { + logger, + applyServerReplyStamp, + markLocallyCommittedReply, + removeConvoFromAllQueries, +} from '~/utils'; +import { createErrorMessage, resolveErrorTurn, resolveSettledErrorReadState } from './turns'; +import { useChatTransport } from '~/Providers/ChatTransportContext'; +import { useAuthContext } from '~/hooks/AuthContext'; + +/** Writes a failed or stopped turn: the error card, a cancelled run, and an abort that settles from what streamed. */ +export default function useAbortHandlers( + ctx: EventHandlerContext, + finalHandler: (data: TFinalResData, submission: EventSubmission) => void, +) { + const { token } = useAuthContext(); + const transport = useChatTransport(); + const { + setCompleted, + setMessages, + paramId, + setIsSubmitting, + setShowStopButton, + setSubmissionStart, + getMessages, + queryClient, + recoverConversation, + reconcileFailedCodeDecision, + setConversation, + isAddedRequest, + } = ctx; + + const cancelHandler = useCallback( + (data: TResData, submission: EventSubmission) => { + const { requestMessage, responseMessage, conversation } = data; + const { messages, isRegenerate = false } = submission; + const convoUpdate = + (conversation as TConversation | null) ?? (submission.conversation as TConversation); + + // update the messages + if (isRegenerate) { + const messagesUpdate = ( + [...messages, responseMessage] as Array + ).filter((msg) => msg); + setMessages(messagesUpdate as TMessage[]); + } else { + const messagesUpdate = ( + [...messages, requestMessage, responseMessage] as Array + ).filter((msg) => msg); + setMessages(messagesUpdate as TMessage[]); + } + + const isNewConvo = conversation.conversationId !== submission.conversation.conversationId; + if (isNewConvo) { + removeConvoFromAllQueries(queryClient, submission.conversation.conversationId as string); + } + + if (setConversation && !isAddedRequest) { + setConversation((prevState) => + keepLocalCodeApprovalMode( + { ...prevState, ...convoUpdate }, + prevState, + convoUpdate.conversationId, + ), + ); + } + + setIsSubmitting(false); + }, + [setMessages, setConversation, isAddedRequest, queryClient, setIsSubmitting], + ); + + const errorHandler = useCallback( + ({ data, submission }: { data?: TResData; submission: EventSubmission }) => { + setCompleted((prev) => new Set(prev.add(submission.initialResponse.messageId))); + setSubmissionStart(null); + + const { conversationId, errorResponse, recover } = resolveErrorTurn({ + data, + submission, + getMessages, + isNewConversationRoute: paramId === Constants.NEW_CONVO, + }); + const finalMessages = mergeErrorMessages({ ...submission, errorMessage: errorResponse }); + setMessages(finalMessages); + queryClient.setQueryData([QueryKeys.messages, conversationId], finalMessages); + const settled = resolveSettledErrorReadState({ data, submission, conversationId }); + if (settled) { + markLocallyCommittedReply(queryClient, conversationId, settled.lastResponseAt); + applyServerReplyStamp(queryClient, conversationId, settled); + } + if (recover) { + recoverConversation(conversationId, submission); + } + reconcileFailedCodeDecision(submission, conversationId); + setIsSubmitting(false); + }, + [ + setCompleted, + setMessages, + paramId, + setIsSubmitting, + setSubmissionStart, + getMessages, + queryClient, + recoverConversation, + reconcileFailedCodeDecision, + ], + ); + + const abortConversation = useCallback( + async (conversationId = '', submission: EventSubmission, messages?: TMessage[]) => { + const runAbortKey = `${conversationId}:${messages?.[messages.length - 1]?.messageId ?? ''}`; + const { endpoint: _endpoint, endpointType } = + (submission.conversation as TConversation | null) ?? {}; + const endpoint = endpointType ?? _endpoint; + if ( + !isAssistantsEndpoint(endpoint) && + messages?.[messages.length - 1] != null && + messages[messages.length - 2] != null + ) { + let requestMessage = messages[messages.length - 2]; + const _responseMessage = messages[messages.length - 1]; + if (requestMessage.messageId !== _responseMessage.parentMessageId) { + // the request message is the parent of response, which we search for backwards + for (let i = messages.length - 3; i >= 0; i--) { + if (messages[i].messageId === _responseMessage.parentMessageId) { + requestMessage = messages[i]; + break; + } + } + } + /** Sanitize content array to remove undefined parts from interrupted streaming */ + const responseMessage = { + ..._responseMessage, + content: _responseMessage.content?.filter((part) => part != null), + }; + try { + finalHandler( + { + conversation: { + conversationId, + }, + requestMessage, + responseMessage, + }, + submission, + ); + } catch (error) { + console.error('Error in finalHandler during abort:', error); + setShowStopButton(false); + setIsSubmitting(false); + setSubmissionStart(null); + } + return; + } else if (!isAssistantsEndpoint(endpoint)) { + const convoId = conversationId || `_${v4()}`; + logger.log('conversation', 'Aborted conversation with minimal messages, ID: ' + convoId); + recoverConversation(convoId, submission); + setIsSubmitting(false); + return; + } + + try { + const response = await transport.abortRun( + { endpoint: endpoint ?? '', abortKey: runAbortKey }, + { token }, + ); + + // Check if the response is JSON + const contentType = response.headers.get('content-type'); + if (contentType != null && contentType.includes('application/json')) { + const data = await response.json(); + if (response.status === 404) { + setIsSubmitting(false); + return; + } + if (data.final === true) { + finalHandler(data, submission); + } else { + cancelHandler(data, submission); + } + } else if (response.status === 204 || response.status === 200) { + setIsSubmitting(false); + } else { + throw new Error( + 'Unexpected response from server; Status: ' + + response.status + + ' ' + + response.statusText, + ); + } + } catch (error) { + const errorResponse = createErrorMessage({ + getMessages, + submission, + error, + }); + /** A compaction has no user row: its `userMessage` slot names the leaf + * the turn hangs off, which `messages` already holds. Writing it here + * would duplicate that id as an empty, self-parented user message, + * a phantom root the thread then folds into. */ + setMessages( + submission.compact === true + ? [...submission.messages, errorResponse] + : [...submission.messages, submission.userMessage, errorResponse], + ); + recoverConversation(conversationId || errorResponse.conversationId || v4(), submission); + setIsSubmitting(false); + } + }, + [ + token, + transport, + getMessages, + setMessages, + finalHandler, + cancelHandler, + setIsSubmitting, + setShowStopButton, + setSubmissionStart, + recoverConversation, + ], + ); + + return { cancelHandler, errorHandler, abortConversation }; +} diff --git a/client/src/hooks/SSE/useEventHandlers.ts b/client/src/hooks/SSE/useEventHandlers.ts index b5d1aa6f429..19f24c41552 100644 --- a/client/src/hooks/SSE/useEventHandlers.ts +++ b/client/src/hooks/SSE/useEventHandlers.ts @@ -1,101 +1,34 @@ import { useCallback, useEffect, useRef } from 'react'; -import { v4 } from 'uuid'; import { useSetRecoilState } from 'recoil'; import { useStore, useSetAtom } from 'jotai'; +import { QueryKeys } from 'librechat-data-provider'; +import { buildRecoveryPreset } from '@librechat/chat'; import { useQueryClient } from '@tanstack/react-query'; import { submissionStartFamily } from '@librechat/chat/react'; import { useParams, useNavigate, useLocation } from 'react-router-dom'; -import { - QueryKeys, - Constants, - ContentTypes, - tMessageSchema, - tConvoUpdateSchema, - isAssistantsEndpoint, -} from 'librechat-data-provider'; -import { - mergeErrorMessages, - buildRecoveryPreset, - responseParentMessageId, - keepLocalCodeApprovalMode, - buildCreatedInitialResponse, - mergeRegenerateFinalMessages, - isInitialNewConversationSubmission, - getExistingConversationAbortMessages, -} from '@librechat/chat'; -import type { - TMessage, - TConversation, - TStartupConfig, - EventSubmission, - ChatCreatedFrame, - TMessageContentParts, -} from 'librechat-data-provider'; +import type { TMessage, TConversation, EventSubmission } from 'librechat-data-provider'; import type { SetterOrUpdater } from 'recoil'; -import type { TResData, TFinalResData, ConvoGenerator } from '~/common'; -import type { ConvoTitleState } from '~/utils/convos'; -import { - logger, - setDraft, - getConversationDraftId, - scrollToEnd, - hasRealTitle, - withoutListFlags, - setDocumentTitle, - requestChatFocus, - getAllContentText, - upsertConvoInAllQueries, - updateConvoInAllQueries, - applyServerReplyStamp, - markLocallyCommittedReply, - removeConvoFromAllQueries, - findConvoInAllQueries, - findManualConvoTitleInAllQueries, - preserveStreamedContentIdentity, - isEmptyContentPart, - getPartKeyIndex, -} from '~/utils'; -import { - startupConfigKey, - queueTitleGeneration, - markTitleGenerationProcessed, - useReconcileConversationCodeEnvironmentMutation, -} from '~/data-provider'; +import type { EventHandlerContext } from './handlers'; +import type { ConvoGenerator } from '~/common'; import { getReasoningStateKey, pendingQuotesByConvoId, pendingReasoningOverrideFamily, } from '~/components/Chat/Input/Composer/state'; import useFocusRegeneratedResponse from '~/hooks/Chat/useFocusRegeneratedResponse'; +import { useReconcileConversationCodeEnvironmentMutation } from '~/data-provider'; import { getFailedCodeDecisionRequest } from '~/hooks/Agents/codeDecision'; import { shouldResetSubagentAtomsOnConversationChange } from './cleanup'; import useAttachmentHandler from '~/hooks/SSE/useAttachmentHandler'; -import { useChatTransport } from '~/Providers/ChatTransportContext'; import useContentHandler from '~/hooks/SSE/useContentHandler'; import useStepHandler from '~/hooks/SSE/useStepHandler'; import { useApplyAgentTemplate } from '~/hooks/Agents'; -import { useAuthContext } from '~/hooks/AuthContext'; -import { MESSAGE_UPDATE_INTERVAL } from '~/common'; +import useAbortHandlers from './useAbortHandlers'; +import useFinalHandler from './useFinalHandler'; +import useTurnHandlers from './useTurnHandlers'; import { useLiveAnnouncer } from '~/Providers'; import store from '~/store'; -type TSyncData = { - sync: boolean; - thread_id: string; - messages?: TMessage[]; - requestMessage: TMessage; - responseMessage: TMessage; - conversationId: string; -}; - -type TTitleEvent = { - event: 'title'; - data?: { - conversationId?: string; - title?: string; - }; -}; - /** Skill caches refreshed when a chat turn authors a skill via `create_file`/`edit_file`. */ const SKILL_QUERY_KEYS = [ QueryKeys.skills, @@ -106,25 +39,6 @@ const SKILL_QUERY_KEYS = [ QueryKeys.skillNodeContent, ] as const; -/** Apply the resolved request choice only after acknowledgement. A saved decision may have - * moved since this event was emitted, so replay must never replace it. */ -function acknowledgedCodeEnvironment( - current: TConversation | null, - submission: EventSubmission, -): Pick { - const { codeEnvironmentMode, codeWorkspaces } = submission; - if (codeEnvironmentMode == null && !codeWorkspaces?.length) return {}; - const saved = - current?.conversationId != null && - current.conversationId !== Constants.NEW_CONVO && - current.conversationId !== Constants.PENDING_CONVO; - if (saved && (current.codeEnvironmentMode != null || current.codeWorkspaces?.length)) return {}; - return { - codeEnvironmentMode, - codeWorkspaces: codeEnvironmentMode === 'without_attached' ? undefined : codeWorkspaces, - }; -} - export type EventHandlerParams = { isAddedRequest?: boolean; runIndex?: number; @@ -137,267 +51,6 @@ export type EventHandlerParams = { setShowStopButton: SetterOrUpdater; }; -const CONNECTION_ERROR_TEXT = 'Error connecting to server, try refreshing the page.'; - -/** - * The parts the in-flight response has streamed so far: the transcript's tail (a user row there - * means no response was placed), without the holes an interrupted stream leaves. Whether anything - * streamed is judged without the slots that never received content (a comparison run's - * `type: ''` placeholders, a text or think part opened before its first delta), but once something - * did, those slots stay: a placeholder is what keeps a comparison lane's layout and attribution - * when only the other lane produced output. The parts carry the render identity they streamed - * under, as the final path stamps it, so the settled row does not remount. - */ -const getStreamedContent = (message?: TMessage): TMessageContentParts[] => { - if (message == null || message.isCreatedByUser === true) { - return []; - } - const streamed = message.content ?? []; - const parts = streamed.filter((part): part is TMessageContentParts => part != null); - if (!parts.some((part) => !isEmptyContentPart(part))) { - return []; - } - return preserveStreamedContentIdentity(streamed, parts) ?? parts; -}; - -/** Keys the appended failure past every key the kept parts render under, physical or stamped. */ -const appendErrorPart = ( - content: TMessageContentParts[], - errorText: string, -): TMessageContentParts[] => { - const keyIndex = content.reduce( - (next, part, idx) => Math.max(next, getPartKeyIndex(part, idx) + 1), - 0, - ); - const errorPart: TMessageContentParts = { type: ContentTypes.ERROR, error: errorText }; - return [ - ...content, - keyIndex === content.length ? errorPart : { ...errorPart, streamedIndex: keyIndex }, - ]; -}; - -/** - * A failure that lands after the run has streamed keeps what streamed and takes the failure as - * one more part, the way the server records a failure inside a run; one that lands before - * anything streamed is the whole row. - */ -const createErrorMessage = ({ - errorMetadata, - getMessages, - submission, - error, -}: { - getMessages: () => TMessage[] | undefined; - errorMetadata?: Partial; - submission: EventSubmission; - error?: Error | unknown; -}): TMessage => { - const currentMessages = getMessages(); - const latestMessage = currentMessages?.[currentMessages.length - 1]; - const text = submission.initialResponse.text.length > 45 ? submission.initialResponse.text : ''; - const errorText = - (errorMetadata?.text || text || (error as Error | undefined)?.message) ?? - 'Error cancelling request'; - const streamedContent = getStreamedContent(latestMessage); - if (latestMessage?.conversationId && latestMessage.messageId && streamedContent.length > 0) { - /** The row keeps the envelope it streamed under (author, model, icon, creation time) and - * takes from the failure only what describes the failure: a server payload names `System` - * as its sender and the schema dates a fresh envelope now, neither of which applies to a - * response that is merely gaining a part. */ - const errorMessage: TMessage = { - ...latestMessage, - ...(errorMetadata?.metadata != null - ? { metadata: { ...latestMessage.metadata, ...errorMetadata.metadata } } - : {}), - error: undefined, - text: '', - content: appendErrorPart(streamedContent, errorText), - }; - const parentMessageId = responseParentMessageId( - submission.initialResponse, - submission.userMessage, - submission.isRegenerate, - ); - if (parentMessageId && parentMessageId !== errorMessage.parentMessageId) { - errorMessage.parentMessageId = parentMessageId; - } - return errorMessage; - } - if (errorMetadata) { - return errorMetadata as TMessage; - } - return tMessageSchema.parse({ - ...submission, - ...submission.initialResponse, - text: errorText, - unfinished: !!text.length, - error: true, - }) as TMessage; -}; - -export interface ErrorTurn { - conversationId: string; - errorResponse: TMessage; - /** The conversation must be rebuilt: it never learned its id, the transport failed, or the - * server named a conversation the new-chat route has not navigated to. */ - recover: boolean; -} - -/** Builds the row a failed turn leaves in the transcript and names the conversation it belongs to. */ -type SettledErrorReadState = { - lastResponseAt: string; - lastResponseMessageId?: string; - updatedAt?: string; -}; - -/** - * A persisted terminal error still carries the server's winning reply stamp in its nested - * conversation snapshot. Pairing the error messages cache object with that stamp before the list - * update lets `useConversationSeen` acknowledge the reply the user just watched fail, instead of - * leaving a dot on a conversation they are looking at. - */ -const resolveSettledErrorReadState = ({ - data, - submission, - conversationId, -}: { - data?: TResData; - submission: EventSubmission; - conversationId: string; -}): SettledErrorReadState | undefined => { - if (submission.isTemporary === true || !data || !conversationId) { - return undefined; - } - const settledConversation = data.conversation; - const lastResponseAt = settledConversation?.lastResponseAt; - if (typeof lastResponseAt !== 'string' || lastResponseAt.length === 0) { - return undefined; - } - return { - lastResponseAt, - lastResponseMessageId: - typeof settledConversation.lastResponseMessageId === 'string' - ? settledConversation.lastResponseMessageId - : undefined, - updatedAt: - typeof settledConversation.updatedAt === 'string' ? settledConversation.updatedAt : undefined, - }; -}; - -export const resolveErrorTurn = ({ - data, - submission, - getMessages, - isNewConversationRoute, -}: { - data?: TResData; - submission: EventSubmission; - getMessages: () => TMessage[] | undefined; - isNewConversationRoute: boolean; -}): ErrorTurn => { - const { userMessage, initialResponse } = submission; - const parentMessageId = responseParentMessageId( - initialResponse, - userMessage, - submission.isRegenerate, - ); - const conversationId = - userMessage.conversationId ?? submission.conversation?.conversationId ?? ''; - - const parseErrorResponse = (payload: TResData | Partial): TMessage => { - const metadata = payload['responseMessage'] ?? payload; - const errorMessage: Partial = { - ...initialResponse, - ...metadata, - error: true, - parentMessageId, - }; - - if (errorMessage.messageId === undefined || errorMessage.messageId === '') { - errorMessage.messageId = v4(); - } - - return tMessageSchema.parse(errorMessage) as TMessage; - }; - const build = (errorMetadata: TMessage): TMessage => - createErrorMessage({ errorMetadata, getMessages, submission }); - - if (!data) { - const convoId = conversationId || `_${v4()}`; - return { - conversationId: convoId, - recover: true, - errorResponse: build( - parseErrorResponse({ text: CONNECTION_ERROR_TEXT, ...submission, conversationId: convoId }), - ), - }; - } - - const receivedConvoId = data.conversationId ?? ''; - if (!conversationId && !receivedConvoId) { - return { - conversationId: `_${v4()}`, - recover: true, - errorResponse: build(parseErrorResponse(data)), - }; - } - if (!receivedConvoId) { - return { conversationId, recover: false, errorResponse: build(parseErrorResponse(data)) }; - } - return { - conversationId: receivedConvoId, - recover: isNewConversationRoute, - errorResponse: build( - tMessageSchema.parse({ - ...data, - error: true, - parentMessageId, - }) as TMessage, - ), - }; -}; - -export const getConvoTitle = ({ - queryClient, - currentTitle, - conversationId, - titleSetByUser, - titleRevision, - authoritative = false, -}: { - queryClient: ReturnType; - currentTitle?: string | null; - conversationId?: string | null; - titleSetByUser?: boolean; - titleRevision?: number; - authoritative?: boolean; -}): ConvoTitleState => { - const incoming = { title: currentTitle ?? null, titleSetByUser, titleRevision }; - if (!conversationId) { - return incoming; - } - const cached = queryClient.getQueryData([QueryKeys.conversation, conversationId]); - const owned = findManualConvoTitleInAllQueries(queryClient, conversationId, incoming); - const listed = owned ? undefined : findConvoInAllQueries(queryClient, conversationId); - let selected = owned; - if (!selected) { - if (authoritative && hasRealTitle(incoming.title)) { - selected = incoming; - } else if (hasRealTitle(cached?.title)) { - selected = cached; - } else if (hasRealTitle(listed?.title)) { - selected = listed; - } else { - selected = incoming; - } - } - return { - title: selected?.title ?? null, - titleSetByUser: selected?.titleSetByUser, - titleRevision: selected?.titleRevision, - }; -}; - export default function useEventHandlers({ setMessages, getMessages, @@ -481,8 +134,6 @@ export default function useEventHandlers({ const lastAnnouncementTimeRef = useRef(Date.now()); const { conversationId: paramId } = useParams(); - const { token } = useAuthContext(); - const transport = useChatTransport(); const { contentHandler, resetContentHandler } = useContentHandler({ setMessages, getMessages }); /** `refetchType: 'all'` so cached-but-unmounted skill queries refresh too: @@ -571,764 +222,36 @@ export default function useEventHandlers({ [resetSubagentAtoms, resetPtcAtoms], ); - const messageHandler = useCallback( - (data: string | undefined, submission: EventSubmission) => { - const { messages, userMessage, initialResponse, isRegenerate = false } = submission; - const text = data ?? ''; - setIsSubmitting(true); - - const currentTime = Date.now(); - if (currentTime - lastAnnouncementTimeRef.current > MESSAGE_UPDATE_INTERVAL) { - announcePolite({ message: 'composing', isStatus: true }); - lastAnnouncementTimeRef.current = currentTime; - } - - if (isRegenerate) { - setMessages([ - ...messages, - { - ...initialResponse, - text, - }, - ]); - } else { - setMessages([ - ...messages, - userMessage, - { - ...initialResponse, - text, - }, - ]); - } - }, - [setMessages, announcePolite, setIsSubmitting], - ); - - const cancelHandler = useCallback( - (data: TResData, submission: EventSubmission) => { - const { requestMessage, responseMessage, conversation } = data; - const { messages, isRegenerate = false } = submission; - const convoUpdate = - (conversation as TConversation | null) ?? (submission.conversation as TConversation); - - // update the messages - if (isRegenerate) { - const messagesUpdate = ( - [...messages, responseMessage] as Array - ).filter((msg) => msg); - setMessages(messagesUpdate as TMessage[]); - } else { - const messagesUpdate = ( - [...messages, requestMessage, responseMessage] as Array - ).filter((msg) => msg); - setMessages(messagesUpdate as TMessage[]); - } - - const isNewConvo = conversation.conversationId !== submission.conversation.conversationId; - if (isNewConvo) { - removeConvoFromAllQueries(queryClient, submission.conversation.conversationId as string); - } - - if (setConversation && !isAddedRequest) { - setConversation((prevState) => - keepLocalCodeApprovalMode( - { ...prevState, ...convoUpdate }, - prevState, - convoUpdate.conversationId, - ), - ); - } - - setIsSubmitting(false); - }, - [setMessages, setConversation, isAddedRequest, queryClient, setIsSubmitting], - ); - - const syncHandler = useCallback( - (data: TSyncData, submission: EventSubmission) => { - const { conversationId, thread_id, responseMessage, requestMessage } = data; - const { initialResponse, messages: _messages, userMessage, isTemporary = false } = submission; - /** Swap the optimistic user row for the server-stamped one IN PLACE. - * Filtering it out and re-appending at the tail would order any of its - * already-present children (abandoned responses from preempted - * attempts) before their parent, and the message tree hoists such rows - * into phantom root branches, a folded thread. */ - const userIndex = _messages.findIndex((msg) => msg.messageId === userMessage.messageId); - const messages = - userIndex >= 0 - ? _messages.map((msg, i) => (i === userIndex ? requestMessage : msg)) - : [..._messages, requestMessage]; - - const nextResponseMessage = { - ...initialResponse, - ...responseMessage, - }; - - setMessages([...messages, nextResponseMessage]); - - announcePolite({ - message: 'start', - isStatus: true, - }); - - let update = {} as TConversation; - if (setConversation && !isAddedRequest) { - setConversation((prevState) => { - const titleState = getConvoTitle({ - queryClient, - conversationId, - currentTitle: prevState?.title, - titleSetByUser: prevState?.titleSetByUser, - titleRevision: prevState?.titleRevision, - }); - update = tConvoUpdateSchema.parse({ - ...prevState, - ...acknowledgedCodeEnvironment(prevState, submission), - conversationId, - thread_id, - ...titleState, - messages: [requestMessage.messageId, responseMessage.messageId], - }) as TConversation; - return update; - }); - - if (!isTemporary) { - const sidebarUpdate = withoutListFlags(update); - if (requestMessage.parentMessageId === Constants.NO_PARENT) { - upsertConvoInAllQueries(queryClient, sidebarUpdate); - } else { - updateConvoInAllQueries(queryClient, update.conversationId!, () => sidebarUpdate, true); - } - if (update.chatProjectId) { - queryClient.invalidateQueries([QueryKeys.projects]); - queryClient.invalidateQueries([QueryKeys.project, update.chatProjectId]); - } - } - } else if (setConversation) { - setConversation((prevState) => { - update = tConvoUpdateSchema.parse({ - ...prevState, - ...acknowledgedCodeEnvironment(prevState, submission), - conversationId, - thread_id, - messages: [requestMessage.messageId, responseMessage.messageId], - }) as TConversation; - return update; - }); - } - - setShowStopButton(true); - }, - [queryClient, setMessages, isAddedRequest, announcePolite, setConversation, setShowStopButton], - ); - const focusRegeneratedResponse = useFocusRegeneratedResponse(); - const createdHandler = useCallback( - (data: ChatCreatedFrame, submission: EventSubmission) => { - const { messages, userMessage, isRegenerate = false, isTemporary = false } = submission; - /** - * The spread carries `manualSkills` through from - * `submission.initialResponse`: `useChatFunctions` seeds the field - * there at construction so the assistant placeholder already has it - * by the time this handler fires. Subsequent `useStepHandler` - * spreads and `updateContent` spreads preserve it, and - * `finalHandler`'s server-backed `responseMessage` replacement - * drops it, which is the right behavior: by finalize the real - * `skill` tool_call is in `content` and takes over rendering. - */ - const initialResponse = buildCreatedInitialResponse({ - initialResponse: submission.initialResponse, - userMessage, - isRegenerate, - }); - if (isRegenerate) { - setMessages([...messages, initialResponse]); - focusRegeneratedResponse(initialResponse.parentMessageId); - } else { - setMessages([...messages, userMessage, initialResponse]); - } - - const { conversationId, parentMessageId } = userMessage; - lastAnnouncementTimeRef.current = Date.now(); - announcePolite({ - message: 'start', - isStatus: true, - }); - - let update = {} as TConversation; - if (setConversation && !isAddedRequest) { - setConversation((prevState) => { - const titleState = getConvoTitle({ - queryClient, - conversationId, - currentTitle: prevState?.title, - titleSetByUser: prevState?.titleSetByUser, - titleRevision: prevState?.titleRevision, - }); - update = tConvoUpdateSchema.parse({ - ...prevState, - ...acknowledgedCodeEnvironment(prevState, submission), - conversationId, - ...titleState, - }) as TConversation; - return update; - }); - - if (!isTemporary) { - const sidebarUpdate = withoutListFlags(update); - if (parentMessageId === Constants.NO_PARENT) { - upsertConvoInAllQueries(queryClient, sidebarUpdate); - } else { - updateConvoInAllQueries(queryClient, update.conversationId!, () => sidebarUpdate, true); - } - if (update.chatProjectId) { - queryClient.invalidateQueries([QueryKeys.projects]); - queryClient.invalidateQueries([QueryKeys.project, update.chatProjectId]); - } - } - } else if (setConversation) { - setConversation((prevState) => { - update = tConvoUpdateSchema.parse({ - ...prevState, - ...acknowledgedCodeEnvironment(prevState, submission), - conversationId, - }) as TConversation; - return update; - }); - } - - if (conversationId) { - applyAgentTemplate({ - targetId: conversationId, - sourceId: submission.conversation?.conversationId, - ephemeralAgent: submission.ephemeralAgent, - specName: submission.conversation?.spec, - startupConfig: queryClient.getQueryData(startupConfigKey(true)), - }); - } - - scrollToEnd(() => setAbortScroll(false)); - }, - [ - setMessages, - queryClient, - setAbortScroll, - isAddedRequest, - announcePolite, - setConversation, - applyAgentTemplate, - focusRegeneratedResponse, - ], - ); - - const titleHandler = useCallback( - (event: TTitleEvent) => { - const { conversationId, title: generatedTitle } = event.data ?? {}; - if (!conversationId || !hasRealTitle(generatedTitle)) { - return; - } - const titleState = getConvoTitle({ - queryClient, - conversationId, - currentTitle: generatedTitle, - }); - const title = titleState.title ?? generatedTitle; - - queryClient.setQueryData([QueryKeys.conversation, conversationId], (convo) => - convo ? { ...convo, ...titleState, title } : convo, - ); - updateConvoInAllQueries(queryClient, conversationId, (convo) => ({ - ...convo, - ...titleState, - title, - })); - markTitleGenerationProcessed(conversationId); - - if (location.pathname.includes(conversationId)) { - setDocumentTitle(title); - } - - if (setConversation && !isAddedRequest) { - setConversation((prevState) => { - if (!prevState) { - return prevState; - } - if (prevState.conversationId && prevState.conversationId !== conversationId) { - return prevState; - } - return { - ...prevState, - conversationId, - ...titleState, - title, - }; - }); - } - }, - [queryClient, location.pathname, setConversation, isAddedRequest], - ); - - const finalHandler = useCallback( - (data: TFinalResData, submission: EventSubmission) => { - const { requestMessage, responseMessage, conversation, runMessages } = data; - const { - messages, - conversation: submissionConvo, - isRegenerate = false, - isTemporary: _isTemporary = false, - } = submission; - const serverConversation = conversation as TConversation; - setSubmissionStart(null); - - try { - // Handle early abort - aborted before any response message was saved. - if ((data as Record).earlyAbort) { - console.log('[finalHandler] Early abort detected - no response message saved'); - setShowStopButton(false); - setIsSubmitting(false); - - const currentConvoId = submissionConvo.conversationId; - const isInitialNewConvo = isInitialNewConversationSubmission(submission); - const isExistingConvo = - currentConvoId && currentConvoId !== Constants.NEW_CONVO && !isInitialNewConvo; - if (isExistingConvo) { - reconcileFailedCodeDecision(submission, currentConvoId); - const abortMessages = getExistingConversationAbortMessages({ - messages, - isRegenerate, - currentMessages: getMessages(), - regenerateMessages: submission.regenerateMessages, - }); - setMessages(abortMessages); - queryClient.setQueryData( - [QueryKeys.messages, currentConvoId], - abortMessages, - ); - setDraft({ id: currentConvoId, value: requestMessage?.text }); - restorePendingContext( - currentConvoId, - requestMessage?.quotes, - requestMessage?.reasoningOverride, - ); - return; - } - - if (currentConvoId && currentConvoId !== Constants.NEW_CONVO) { - removeConvoFromAllQueries(queryClient, currentConvoId); - queryClient.removeQueries({ queryKey: [QueryKeys.conversation, currentConvoId] }); - queryClient.removeQueries({ queryKey: [QueryKeys.messages, currentConvoId] }); - } - setMessages([]); - queryClient.setQueryData([QueryKeys.messages, Constants.NEW_CONVO], []); - setDraft({ - id: getConversationDraftId(runIndex, Constants.NEW_CONVO), - value: requestMessage?.text, - }); - restorePendingContext( - String(Constants.NEW_CONVO), - requestMessage?.quotes, - requestMessage?.reasoningOverride, - ); - if (location.pathname !== `/c/${Constants.NEW_CONVO}`) { - navigate(`/c/${Constants.NEW_CONVO}`, { replace: true }); - } - return; - } - - if (responseMessage?.attachments && responseMessage.attachments.length > 0) { - // Process each attachment through the attachmentHandler - responseMessage.attachments.forEach((attachment) => { - const attachmentData = { - ...attachment, - messageId: responseMessage.messageId, - }; - - attachmentHandler({ - data: attachmentData, - submission: submission as EventSubmission, - }); - }); - } - - setCompleted((prev) => new Set(prev.add(submission.initialResponse.messageId))); - - const currentMessages = getMessages(); - /* Early return if messages are empty; i.e., the user navigated away */ - if (!currentMessages || currentMessages.length === 0) { - return; - } - - /* a11y announcements */ - announcePolite({ message: 'end', isStatus: true }); - announcePolite({ message: getAllContentText(responseMessage) }); - - const isNewConvo = conversation.conversationId !== submissionConvo.conversationId; - - // Skip temporary conversations; the server never generates titles for them. - if (isNewConvo && conversation.conversationId && !_isTemporary) { - queueTitleGeneration(conversation.conversationId); - } - - const setFinalMessages = (id: string | null, _messages: TMessage[]) => { - setMessages(_messages); - queryClient.setQueryData([QueryKeys.messages, id], _messages); - }; - - const hasNoResponse = - responseMessage?.content?.[0]?.['text']?.value === - submission.initialResponse?.content?.[0]?.['text']?.value || - !!responseMessage?.content?.[0]?.['tool_call']?.auth; - - /** Handle edge case where stream is cancelled before any response, which creates a blank page */ - if (!conversation.conversationId && hasNoResponse) { - const currentConvoId = - (submissionConvo.conversationId ?? conversation.conversationId) || Constants.NEW_CONVO; - if (isNewConvo && submissionConvo.conversationId) { - removeConvoFromAllQueries(queryClient, submissionConvo.conversationId); - } - - const isNewChat = - location.pathname === `/c/${Constants.NEW_CONVO}` && - currentConvoId === Constants.NEW_CONVO; - - setFinalMessages(currentConvoId, isNewChat ? [] : [...messages]); - setDraft({ - id: getConversationDraftId(runIndex, currentConvoId), - value: requestMessage?.text, - }); - restorePendingContext( - currentConvoId, - requestMessage?.quotes, - requestMessage?.reasoningOverride, - ); - if (isNewChat) { - requestChatFocus(); - navigate(`/c/${Constants.NEW_CONVO}`, { replace: true }); - } - return; - } - - /* Update messages; if assistants endpoint, client doesn't receive responseMessage */ - let finalMessages: TMessage[] = []; - if (runMessages) { - finalMessages = [...runMessages]; - } else if (isRegenerate && responseMessage) { - finalMessages = mergeRegenerateFinalMessages({ - messages: submission.regenerateMessages ?? currentMessages ?? messages, - responseMessage, - initialResponseId: submission.initialResponse.messageId, - }); - } else if (requestMessage != null && responseMessage != null) { - finalMessages = [...messages, requestMessage, responseMessage]; - } - - /* Preserve files and streamed content identity from current messages: - * files fill in when the server response lacks them, and the persisted - * (compacted) content is stamped with the indexes it streamed at so - * index-keyed renders don't remount the settled message. */ - if (finalMessages.length > 0) { - const currentMsgMap = new Map(currentMessages.map((m) => [m.messageId, m])); - for (let i = 0; i < finalMessages.length; i++) { - const msg = finalMessages[i]; - const currentMsg = currentMsgMap.get(msg.messageId); - if (!currentMsg) { - continue; - } - const preservedFiles = - msg.files == null && currentMsg.files?.length ? currentMsg.files : undefined; - const content = preserveStreamedContentIdentity(currentMsg.content, msg.content); - if (preservedFiles == null && content === msg.content) { - continue; - } - finalMessages[i] = { - ...msg, - ...(preservedFiles != null ? { files: preservedFiles } : {}), - ...(content !== msg.content ? { content } : {}), - }; - } - } - - if (finalMessages.length > 0) { - setFinalMessages(conversation.conversationId, finalMessages); - } else if ( - isAssistantsEndpoint(submissionConvo.endpoint) && - (!submissionConvo.conversationId || - submissionConvo.conversationId === Constants.NEW_CONVO) - ) { - queryClient.setQueryData( - [QueryKeys.messages, conversation.conversationId], - [...currentMessages], - ); - } - - if (isNewConvo && submissionConvo.conversationId) { - removeConvoFromAllQueries(queryClient, submissionConvo.conversationId); - } - - /** Stream snapshots cannot overwrite a rename or generated title already in the cache. */ - if (setConversation && isAddedRequest !== true) { - /* Pair the durable server stamp with the exact final messages cache object before - * point/list cache events can ask useConversationSeen to acknowledge it. */ - const serverLastResponseAt = serverConversation.lastResponseAt; - if (conversation.conversationId && !_isTemporary && serverLastResponseAt) { - markLocallyCommittedReply( - queryClient, - conversation.conversationId, - serverLastResponseAt, - ); - applyServerReplyStamp(queryClient, conversation.conversationId, { - lastResponseAt: serverLastResponseAt, - lastResponseMessageId: serverConversation.lastResponseMessageId, - updatedAt: serverConversation.updatedAt, - }); - } - setConversation((prevState) => { - const update = keepLocalCodeApprovalMode( - { ...prevState, ...(conversation as TConversation) }, - prevState, - conversation.conversationId, - ); - if (prevState?.model != null && prevState.model !== submissionConvo.model) { - update.model = prevState.model; - } - const titleState = getConvoTitle({ - queryClient, - conversationId: conversation.conversationId, - titleSetByUser: conversation.titleSetByUser, - titleRevision: conversation.titleRevision, - authoritative: true, - currentTitle: - !conversation.titleSetByUser && - !hasRealTitle(conversation.title) && - prevState?.conversationId === conversation.conversationId && - hasRealTitle(prevState.title) - ? prevState.title - : conversation.title, - }); - Object.assign(update, titleState); - if (conversation.conversationId) { - updateConvoInAllQueries(queryClient, conversation.conversationId, (convo) => ({ - ...convo, - ...titleState, - })); - queryClient.setQueryData( - [QueryKeys.conversation, conversation.conversationId], - (cachedConvo) => { - const merged = keepLocalCodeApprovalMode( - { ...cachedConvo, ...serverConversation } as TConversation, - prevState?.conversationId === conversation.conversationId - ? prevState - : cachedConvo, - conversation.conversationId, - ); - Object.assign(merged, titleState); - return merged; - }, - ); - } - return update; - }); - - if (conversation.conversationId && submission.ephemeralAgent) { - applyAgentTemplate({ - targetId: conversation.conversationId, - sourceId: submissionConvo.conversationId, - ephemeralAgent: submission.ephemeralAgent, - specName: submission.conversation?.spec, - startupConfig: queryClient.getQueryData(startupConfigKey(true)), - }); - } - - if (conversation.chatProjectId) { - queryClient.invalidateQueries([QueryKeys.projects]); - queryClient.invalidateQueries([QueryKeys.project, conversation.chatProjectId]); - } - - if (location.pathname === `/c/${Constants.NEW_CONVO}`) { - preserveSubagentAtomsForNewConvoIdRef.current = conversation.conversationId; - navigate(`/c/${conversation.conversationId}`, { replace: true }); - } - } - } finally { - setShowStopButton(false); - setIsSubmitting(false); - } - }, - [ - navigate, - getMessages, - setMessages, - queryClient, - setCompleted, - runIndex, - isAddedRequest, - announcePolite, - setConversation, - setIsSubmitting, - setShowStopButton, - location.pathname, - applyAgentTemplate, - attachmentHandler, - setSubmissionStart, - restorePendingContext, - reconcileFailedCodeDecision, - ], - ); - - const errorHandler = useCallback( - ({ data, submission }: { data?: TResData; submission: EventSubmission }) => { - setCompleted((prev) => new Set(prev.add(submission.initialResponse.messageId))); - setSubmissionStart(null); - - const { conversationId, errorResponse, recover } = resolveErrorTurn({ - data, - submission, - getMessages, - isNewConversationRoute: paramId === Constants.NEW_CONVO, - }); - const finalMessages = mergeErrorMessages({ ...submission, errorMessage: errorResponse }); - setMessages(finalMessages); - queryClient.setQueryData([QueryKeys.messages, conversationId], finalMessages); - const settled = resolveSettledErrorReadState({ data, submission, conversationId }); - if (settled) { - markLocallyCommittedReply(queryClient, conversationId, settled.lastResponseAt); - applyServerReplyStamp(queryClient, conversationId, settled); - } - if (recover) { - recoverConversation(conversationId, submission); - } - reconcileFailedCodeDecision(submission, conversationId); - setIsSubmitting(false); - }, - [ - setCompleted, - setMessages, - paramId, - setIsSubmitting, - setSubmissionStart, - getMessages, - queryClient, - recoverConversation, - reconcileFailedCodeDecision, - ], - ); - - const abortConversation = useCallback( - async (conversationId = '', submission: EventSubmission, messages?: TMessage[]) => { - const runAbortKey = `${conversationId}:${messages?.[messages.length - 1]?.messageId ?? ''}`; - const { endpoint: _endpoint, endpointType } = - (submission.conversation as TConversation | null) ?? {}; - const endpoint = endpointType ?? _endpoint; - if ( - !isAssistantsEndpoint(endpoint) && - messages?.[messages.length - 1] != null && - messages[messages.length - 2] != null - ) { - let requestMessage = messages[messages.length - 2]; - const _responseMessage = messages[messages.length - 1]; - if (requestMessage.messageId !== _responseMessage.parentMessageId) { - // the request message is the parent of response, which we search for backwards - for (let i = messages.length - 3; i >= 0; i--) { - if (messages[i].messageId === _responseMessage.parentMessageId) { - requestMessage = messages[i]; - break; - } - } - } - /** Sanitize content array to remove undefined parts from interrupted streaming */ - const responseMessage = { - ..._responseMessage, - content: _responseMessage.content?.filter((part) => part != null), - }; - try { - finalHandler( - { - conversation: { - conversationId, - }, - requestMessage, - responseMessage, - }, - submission, - ); - } catch (error) { - console.error('Error in finalHandler during abort:', error); - setShowStopButton(false); - setIsSubmitting(false); - setSubmissionStart(null); - } - return; - } else if (!isAssistantsEndpoint(endpoint)) { - const convoId = conversationId || `_${v4()}`; - logger.log('conversation', 'Aborted conversation with minimal messages, ID: ' + convoId); - recoverConversation(convoId, submission); - setIsSubmitting(false); - return; - } - - try { - const response = await transport.abortRun( - { endpoint: endpoint ?? '', abortKey: runAbortKey }, - { token }, - ); - - // Check if the response is JSON - const contentType = response.headers.get('content-type'); - if (contentType != null && contentType.includes('application/json')) { - const data = await response.json(); - if (response.status === 404) { - setIsSubmitting(false); - return; - } - if (data.final === true) { - finalHandler(data, submission); - } else { - cancelHandler(data, submission); - } - } else if (response.status === 204 || response.status === 200) { - setIsSubmitting(false); - } else { - throw new Error( - 'Unexpected response from server; Status: ' + - response.status + - ' ' + - response.statusText, - ); - } - } catch (error) { - const errorResponse = createErrorMessage({ - getMessages, - submission, - error, - }); - /** A compaction has no user row: its `userMessage` slot names the leaf - * the turn hangs off, which `messages` already holds. Writing it here - * would duplicate that id as an empty, self-parented user message, - * a phantom root the thread then folds into. */ - setMessages( - submission.compact === true - ? [...submission.messages, errorResponse] - : [...submission.messages, submission.userMessage, errorResponse], - ); - recoverConversation(conversationId || errorResponse.conversationId || v4(), submission); - setIsSubmitting(false); - } - }, - [ - token, - transport, - getMessages, - setMessages, - finalHandler, - cancelHandler, - setIsSubmitting, - setShowStopButton, - setSubmissionStart, - recoverConversation, - ], - ); + const ctx: EventHandlerContext = { + queryClient, + setMessages, + getMessages, + setCompleted, + isAddedRequest, + runIndex, + setConversation, + setIsSubmitting, + setShowStopButton, + setAbortScroll, + setSubmissionStart, + announcePolite, + lastAnnouncementTimeRef, + applyAgentTemplate, + focusRegeneratedResponse, + location, + navigate, + paramId, + restorePendingContext, + reconcileFailedCodeDecision, + recoverConversation, + attachmentHandler, + preserveSubagentAtomsForNewConvoIdRef, + }; + const { messageHandler, syncHandler, createdHandler, titleHandler } = useTurnHandlers(ctx); + const finalHandler = useFinalHandler(ctx); + const { errorHandler, abortConversation } = useAbortHandlers(ctx, finalHandler); return { stepHandler, diff --git a/client/src/hooks/SSE/useFinalHandler.ts b/client/src/hooks/SSE/useFinalHandler.ts new file mode 100644 index 00000000000..59c52e8eb2b --- /dev/null +++ b/client/src/hooks/SSE/useFinalHandler.ts @@ -0,0 +1,364 @@ +import { useCallback } from 'react'; +import { QueryKeys, Constants, isAssistantsEndpoint } from 'librechat-data-provider'; +import { + keepLocalCodeApprovalMode, + mergeRegenerateFinalMessages, + isInitialNewConversationSubmission, + getExistingConversationAbortMessages, +} from '@librechat/chat'; +import type { + TMessage, + TConversation, + TStartupConfig, + EventSubmission, +} from 'librechat-data-provider'; +import type { EventHandlerContext } from './handlers'; +import type { TFinalResData } from '~/common'; +import { + setDraft, + getConversationDraftId, + hasRealTitle, + requestChatFocus, + getAllContentText, + updateConvoInAllQueries, + applyServerReplyStamp, + markLocallyCommittedReply, + removeConvoFromAllQueries, + preserveStreamedContentIdentity, +} from '~/utils'; +import { startupConfigKey, queueTitleGeneration } from '~/data-provider'; +import { getConvoTitle } from './turns'; + +/** Settles a finished turn: the final messages, the conversation it belongs to, and an early abort that restores the draft. */ +export default function useFinalHandler(ctx: EventHandlerContext) { + const { + navigate, + getMessages, + setMessages, + queryClient, + setCompleted, + runIndex, + isAddedRequest, + announcePolite, + setConversation, + setIsSubmitting, + setShowStopButton, + location, + applyAgentTemplate, + attachmentHandler, + setSubmissionStart, + restorePendingContext, + reconcileFailedCodeDecision, + preserveSubagentAtomsForNewConvoIdRef, + } = ctx; + + const finalHandler = useCallback( + (data: TFinalResData, submission: EventSubmission) => { + const { requestMessage, responseMessage, conversation, runMessages } = data; + const { + messages, + conversation: submissionConvo, + isRegenerate = false, + isTemporary: _isTemporary = false, + } = submission; + const serverConversation = conversation as TConversation; + setSubmissionStart(null); + + try { + // Handle early abort - aborted before any response message was saved. + if ((data as Record).earlyAbort) { + console.log('[finalHandler] Early abort detected - no response message saved'); + setShowStopButton(false); + setIsSubmitting(false); + + const currentConvoId = submissionConvo.conversationId; + const isInitialNewConvo = isInitialNewConversationSubmission(submission); + const isExistingConvo = + currentConvoId && currentConvoId !== Constants.NEW_CONVO && !isInitialNewConvo; + if (isExistingConvo) { + reconcileFailedCodeDecision(submission, currentConvoId); + const abortMessages = getExistingConversationAbortMessages({ + messages, + isRegenerate, + currentMessages: getMessages(), + regenerateMessages: submission.regenerateMessages, + }); + setMessages(abortMessages); + queryClient.setQueryData( + [QueryKeys.messages, currentConvoId], + abortMessages, + ); + setDraft({ id: currentConvoId, value: requestMessage?.text }); + restorePendingContext( + currentConvoId, + requestMessage?.quotes, + requestMessage?.reasoningOverride, + ); + return; + } + + if (currentConvoId && currentConvoId !== Constants.NEW_CONVO) { + removeConvoFromAllQueries(queryClient, currentConvoId); + queryClient.removeQueries({ queryKey: [QueryKeys.conversation, currentConvoId] }); + queryClient.removeQueries({ queryKey: [QueryKeys.messages, currentConvoId] }); + } + setMessages([]); + queryClient.setQueryData([QueryKeys.messages, Constants.NEW_CONVO], []); + setDraft({ + id: getConversationDraftId(runIndex, Constants.NEW_CONVO), + value: requestMessage?.text, + }); + restorePendingContext( + String(Constants.NEW_CONVO), + requestMessage?.quotes, + requestMessage?.reasoningOverride, + ); + if (location.pathname !== `/c/${Constants.NEW_CONVO}`) { + navigate(`/c/${Constants.NEW_CONVO}`, { replace: true }); + } + return; + } + + if (responseMessage?.attachments && responseMessage.attachments.length > 0) { + // Process each attachment through the attachmentHandler + responseMessage.attachments.forEach((attachment) => { + const attachmentData = { + ...attachment, + messageId: responseMessage.messageId, + }; + + attachmentHandler({ + data: attachmentData, + submission: submission as EventSubmission, + }); + }); + } + + setCompleted((prev) => new Set(prev.add(submission.initialResponse.messageId))); + + const currentMessages = getMessages(); + /* Early return if messages are empty; i.e., the user navigated away */ + if (!currentMessages || currentMessages.length === 0) { + return; + } + + /* a11y announcements */ + announcePolite({ message: 'end', isStatus: true }); + announcePolite({ message: getAllContentText(responseMessage) }); + + const isNewConvo = conversation.conversationId !== submissionConvo.conversationId; + + // Skip temporary conversations; the server never generates titles for them. + if (isNewConvo && conversation.conversationId && !_isTemporary) { + queueTitleGeneration(conversation.conversationId); + } + + const setFinalMessages = (id: string | null, _messages: TMessage[]) => { + setMessages(_messages); + queryClient.setQueryData([QueryKeys.messages, id], _messages); + }; + + const hasNoResponse = + responseMessage?.content?.[0]?.['text']?.value === + submission.initialResponse?.content?.[0]?.['text']?.value || + !!responseMessage?.content?.[0]?.['tool_call']?.auth; + + /** Handle edge case where stream is cancelled before any response, which creates a blank page */ + if (!conversation.conversationId && hasNoResponse) { + const currentConvoId = + (submissionConvo.conversationId ?? conversation.conversationId) || Constants.NEW_CONVO; + if (isNewConvo && submissionConvo.conversationId) { + removeConvoFromAllQueries(queryClient, submissionConvo.conversationId); + } + + const isNewChat = + location.pathname === `/c/${Constants.NEW_CONVO}` && + currentConvoId === Constants.NEW_CONVO; + + setFinalMessages(currentConvoId, isNewChat ? [] : [...messages]); + setDraft({ + id: getConversationDraftId(runIndex, currentConvoId), + value: requestMessage?.text, + }); + restorePendingContext( + currentConvoId, + requestMessage?.quotes, + requestMessage?.reasoningOverride, + ); + if (isNewChat) { + requestChatFocus(); + navigate(`/c/${Constants.NEW_CONVO}`, { replace: true }); + } + return; + } + + /* Update messages; if assistants endpoint, client doesn't receive responseMessage */ + let finalMessages: TMessage[] = []; + if (runMessages) { + finalMessages = [...runMessages]; + } else if (isRegenerate && responseMessage) { + finalMessages = mergeRegenerateFinalMessages({ + messages: submission.regenerateMessages ?? currentMessages ?? messages, + responseMessage, + initialResponseId: submission.initialResponse.messageId, + }); + } else if (requestMessage != null && responseMessage != null) { + finalMessages = [...messages, requestMessage, responseMessage]; + } + + /* Preserve files and streamed content identity from current messages: + * files fill in when the server response lacks them, and the persisted + * (compacted) content is stamped with the indexes it streamed at so + * index-keyed renders don't remount the settled message. */ + if (finalMessages.length > 0) { + const currentMsgMap = new Map(currentMessages.map((m) => [m.messageId, m])); + for (let i = 0; i < finalMessages.length; i++) { + const msg = finalMessages[i]; + const currentMsg = currentMsgMap.get(msg.messageId); + if (!currentMsg) { + continue; + } + const preservedFiles = + msg.files == null && currentMsg.files?.length ? currentMsg.files : undefined; + const content = preserveStreamedContentIdentity(currentMsg.content, msg.content); + if (preservedFiles == null && content === msg.content) { + continue; + } + finalMessages[i] = { + ...msg, + ...(preservedFiles != null ? { files: preservedFiles } : {}), + ...(content !== msg.content ? { content } : {}), + }; + } + } + + if (finalMessages.length > 0) { + setFinalMessages(conversation.conversationId, finalMessages); + } else if ( + isAssistantsEndpoint(submissionConvo.endpoint) && + (!submissionConvo.conversationId || + submissionConvo.conversationId === Constants.NEW_CONVO) + ) { + queryClient.setQueryData( + [QueryKeys.messages, conversation.conversationId], + [...currentMessages], + ); + } + + if (isNewConvo && submissionConvo.conversationId) { + removeConvoFromAllQueries(queryClient, submissionConvo.conversationId); + } + + /** Stream snapshots cannot overwrite a rename or generated title already in the cache. */ + if (setConversation && isAddedRequest !== true) { + /* Pair the durable server stamp with the exact final messages cache object before + * point/list cache events can ask useConversationSeen to acknowledge it. */ + const serverLastResponseAt = serverConversation.lastResponseAt; + if (conversation.conversationId && !_isTemporary && serverLastResponseAt) { + markLocallyCommittedReply( + queryClient, + conversation.conversationId, + serverLastResponseAt, + ); + applyServerReplyStamp(queryClient, conversation.conversationId, { + lastResponseAt: serverLastResponseAt, + lastResponseMessageId: serverConversation.lastResponseMessageId, + updatedAt: serverConversation.updatedAt, + }); + } + setConversation((prevState) => { + const update = keepLocalCodeApprovalMode( + { ...prevState, ...(conversation as TConversation) }, + prevState, + conversation.conversationId, + ); + if (prevState?.model != null && prevState.model !== submissionConvo.model) { + update.model = prevState.model; + } + const titleState = getConvoTitle({ + queryClient, + conversationId: conversation.conversationId, + titleSetByUser: conversation.titleSetByUser, + titleRevision: conversation.titleRevision, + authoritative: true, + currentTitle: + !conversation.titleSetByUser && + !hasRealTitle(conversation.title) && + prevState?.conversationId === conversation.conversationId && + hasRealTitle(prevState.title) + ? prevState.title + : conversation.title, + }); + Object.assign(update, titleState); + if (conversation.conversationId) { + updateConvoInAllQueries(queryClient, conversation.conversationId, (convo) => ({ + ...convo, + ...titleState, + })); + queryClient.setQueryData( + [QueryKeys.conversation, conversation.conversationId], + (cachedConvo) => { + const merged = keepLocalCodeApprovalMode( + { ...cachedConvo, ...serverConversation } as TConversation, + prevState?.conversationId === conversation.conversationId + ? prevState + : cachedConvo, + conversation.conversationId, + ); + Object.assign(merged, titleState); + return merged; + }, + ); + } + return update; + }); + + if (conversation.conversationId && submission.ephemeralAgent) { + applyAgentTemplate({ + targetId: conversation.conversationId, + sourceId: submissionConvo.conversationId, + ephemeralAgent: submission.ephemeralAgent, + specName: submission.conversation?.spec, + startupConfig: queryClient.getQueryData(startupConfigKey(true)), + }); + } + + if (conversation.chatProjectId) { + queryClient.invalidateQueries([QueryKeys.projects]); + queryClient.invalidateQueries([QueryKeys.project, conversation.chatProjectId]); + } + + if (location.pathname === `/c/${Constants.NEW_CONVO}`) { + preserveSubagentAtomsForNewConvoIdRef.current = conversation.conversationId; + navigate(`/c/${conversation.conversationId}`, { replace: true }); + } + } + } finally { + setShowStopButton(false); + setIsSubmitting(false); + } + }, + [ + navigate, + getMessages, + setMessages, + queryClient, + setCompleted, + runIndex, + isAddedRequest, + announcePolite, + setConversation, + setIsSubmitting, + setShowStopButton, + location.pathname, + applyAgentTemplate, + attachmentHandler, + setSubmissionStart, + restorePendingContext, + reconcileFailedCodeDecision, + preserveSubagentAtomsForNewConvoIdRef, + ], + ); + + return finalHandler; +} diff --git a/client/src/hooks/SSE/useTurnHandlers.ts b/client/src/hooks/SSE/useTurnHandlers.ts new file mode 100644 index 00000000000..63b49595efd --- /dev/null +++ b/client/src/hooks/SSE/useTurnHandlers.ts @@ -0,0 +1,335 @@ +import { useCallback } from 'react'; +import { buildCreatedInitialResponse } from '@librechat/chat'; +import { QueryKeys, Constants, tConvoUpdateSchema } from 'librechat-data-provider'; +import type { + TMessage, + TConversation, + TStartupConfig, + EventSubmission, + ChatCreatedFrame, +} from 'librechat-data-provider'; +import type { EventHandlerContext } from './handlers'; +import { + scrollToEnd, + hasRealTitle, + withoutListFlags, + setDocumentTitle, + upsertConvoInAllQueries, + updateConvoInAllQueries, +} from '~/utils'; +import { startupConfigKey, markTitleGenerationProcessed } from '~/data-provider'; +import { MESSAGE_UPDATE_INTERVAL } from '~/common'; +import { getConvoTitle } from './turns'; +export type TSyncData = { + sync: boolean; + thread_id: string; + messages?: TMessage[]; + requestMessage: TMessage; + responseMessage: TMessage; + conversationId: string; +}; + +export type TTitleEvent = { + event: 'title'; + data?: { + conversationId?: string; + title?: string; + }; +}; + +/** Apply the resolved request choice only after acknowledgement. A saved decision may have + * moved since this event was emitted, so replay must never replace it. */ +export function acknowledgedCodeEnvironment( + current: TConversation | null, + submission: EventSubmission, +): Pick { + const { codeEnvironmentMode, codeWorkspaces } = submission; + if (codeEnvironmentMode == null && !codeWorkspaces?.length) return {}; + const saved = + current?.conversationId != null && + current.conversationId !== Constants.NEW_CONVO && + current.conversationId !== Constants.PENDING_CONVO; + if (saved && (current.codeEnvironmentMode != null || current.codeWorkspaces?.length)) return {}; + return { + codeEnvironmentMode, + codeWorkspaces: codeEnvironmentMode === 'without_attached' ? undefined : codeWorkspaces, + }; +} + +/** The handlers that open a turn and keep its conversation current: legacy text, sync, created and title events. */ +export default function useTurnHandlers(ctx: EventHandlerContext) { + const { + queryClient, + setMessages, + setIsSubmitting, + setShowStopButton, + setConversation, + isAddedRequest, + announcePolite, + lastAnnouncementTimeRef, + setAbortScroll, + applyAgentTemplate, + focusRegeneratedResponse, + location, + } = ctx; + + const messageHandler = useCallback( + (data: string | undefined, submission: EventSubmission) => { + const { messages, userMessage, initialResponse, isRegenerate = false } = submission; + const text = data ?? ''; + setIsSubmitting(true); + + const currentTime = Date.now(); + if (currentTime - lastAnnouncementTimeRef.current > MESSAGE_UPDATE_INTERVAL) { + announcePolite({ message: 'composing', isStatus: true }); + lastAnnouncementTimeRef.current = currentTime; + } + + if (isRegenerate) { + setMessages([ + ...messages, + { + ...initialResponse, + text, + }, + ]); + } else { + setMessages([ + ...messages, + userMessage, + { + ...initialResponse, + text, + }, + ]); + } + }, + [setMessages, announcePolite, setIsSubmitting, lastAnnouncementTimeRef], + ); + + const syncHandler = useCallback( + (data: TSyncData, submission: EventSubmission) => { + const { conversationId, thread_id, responseMessage, requestMessage } = data; + const { initialResponse, messages: _messages, userMessage, isTemporary = false } = submission; + /** Swap the optimistic user row for the server-stamped one IN PLACE. + * Filtering it out and re-appending at the tail would order any of its + * already-present children (abandoned responses from preempted + * attempts) before their parent, and the message tree hoists such rows + * into phantom root branches, a folded thread. */ + const userIndex = _messages.findIndex((msg) => msg.messageId === userMessage.messageId); + const messages = + userIndex >= 0 + ? _messages.map((msg, i) => (i === userIndex ? requestMessage : msg)) + : [..._messages, requestMessage]; + + const nextResponseMessage = { + ...initialResponse, + ...responseMessage, + }; + + setMessages([...messages, nextResponseMessage]); + + announcePolite({ + message: 'start', + isStatus: true, + }); + + let update = {} as TConversation; + if (setConversation && !isAddedRequest) { + setConversation((prevState) => { + const titleState = getConvoTitle({ + queryClient, + conversationId, + currentTitle: prevState?.title, + titleSetByUser: prevState?.titleSetByUser, + titleRevision: prevState?.titleRevision, + }); + update = tConvoUpdateSchema.parse({ + ...prevState, + ...acknowledgedCodeEnvironment(prevState, submission), + conversationId, + thread_id, + ...titleState, + messages: [requestMessage.messageId, responseMessage.messageId], + }) as TConversation; + return update; + }); + + if (!isTemporary) { + const sidebarUpdate = withoutListFlags(update); + if (requestMessage.parentMessageId === Constants.NO_PARENT) { + upsertConvoInAllQueries(queryClient, sidebarUpdate); + } else { + updateConvoInAllQueries(queryClient, update.conversationId!, () => sidebarUpdate, true); + } + if (update.chatProjectId) { + queryClient.invalidateQueries([QueryKeys.projects]); + queryClient.invalidateQueries([QueryKeys.project, update.chatProjectId]); + } + } + } else if (setConversation) { + setConversation((prevState) => { + update = tConvoUpdateSchema.parse({ + ...prevState, + ...acknowledgedCodeEnvironment(prevState, submission), + conversationId, + thread_id, + messages: [requestMessage.messageId, responseMessage.messageId], + }) as TConversation; + return update; + }); + } + + setShowStopButton(true); + }, + [queryClient, setMessages, isAddedRequest, announcePolite, setConversation, setShowStopButton], + ); + + const createdHandler = useCallback( + (data: ChatCreatedFrame, submission: EventSubmission) => { + const { messages, userMessage, isRegenerate = false, isTemporary = false } = submission; + /** + * The spread carries `manualSkills` through from + * `submission.initialResponse`: `useChatFunctions` seeds the field + * there at construction so the assistant placeholder already has it + * by the time this handler fires. Subsequent `useStepHandler` + * spreads and `updateContent` spreads preserve it, and + * `finalHandler`'s server-backed `responseMessage` replacement + * drops it, which is the right behavior: by finalize the real + * `skill` tool_call is in `content` and takes over rendering. + */ + const initialResponse = buildCreatedInitialResponse({ + initialResponse: submission.initialResponse, + userMessage, + isRegenerate, + }); + if (isRegenerate) { + setMessages([...messages, initialResponse]); + focusRegeneratedResponse(initialResponse.parentMessageId); + } else { + setMessages([...messages, userMessage, initialResponse]); + } + + const { conversationId, parentMessageId } = userMessage; + lastAnnouncementTimeRef.current = Date.now(); + announcePolite({ + message: 'start', + isStatus: true, + }); + + let update = {} as TConversation; + if (setConversation && !isAddedRequest) { + setConversation((prevState) => { + const titleState = getConvoTitle({ + queryClient, + conversationId, + currentTitle: prevState?.title, + titleSetByUser: prevState?.titleSetByUser, + titleRevision: prevState?.titleRevision, + }); + update = tConvoUpdateSchema.parse({ + ...prevState, + ...acknowledgedCodeEnvironment(prevState, submission), + conversationId, + ...titleState, + }) as TConversation; + return update; + }); + + if (!isTemporary) { + const sidebarUpdate = withoutListFlags(update); + if (parentMessageId === Constants.NO_PARENT) { + upsertConvoInAllQueries(queryClient, sidebarUpdate); + } else { + updateConvoInAllQueries(queryClient, update.conversationId!, () => sidebarUpdate, true); + } + if (update.chatProjectId) { + queryClient.invalidateQueries([QueryKeys.projects]); + queryClient.invalidateQueries([QueryKeys.project, update.chatProjectId]); + } + } + } else if (setConversation) { + setConversation((prevState) => { + update = tConvoUpdateSchema.parse({ + ...prevState, + ...acknowledgedCodeEnvironment(prevState, submission), + conversationId, + }) as TConversation; + return update; + }); + } + + if (conversationId) { + applyAgentTemplate({ + targetId: conversationId, + sourceId: submission.conversation?.conversationId, + ephemeralAgent: submission.ephemeralAgent, + specName: submission.conversation?.spec, + startupConfig: queryClient.getQueryData(startupConfigKey(true)), + }); + } + + scrollToEnd(() => setAbortScroll(false)); + }, + [ + setMessages, + queryClient, + setAbortScroll, + isAddedRequest, + announcePolite, + setConversation, + applyAgentTemplate, + focusRegeneratedResponse, + lastAnnouncementTimeRef, + ], + ); + + const titleHandler = useCallback( + (event: TTitleEvent) => { + const { conversationId, title: generatedTitle } = event.data ?? {}; + if (!conversationId || !hasRealTitle(generatedTitle)) { + return; + } + const titleState = getConvoTitle({ + queryClient, + conversationId, + currentTitle: generatedTitle, + }); + const title = titleState.title ?? generatedTitle; + + queryClient.setQueryData([QueryKeys.conversation, conversationId], (convo) => + convo ? { ...convo, ...titleState, title } : convo, + ); + updateConvoInAllQueries(queryClient, conversationId, (convo) => ({ + ...convo, + ...titleState, + title, + })); + markTitleGenerationProcessed(conversationId); + + if (location.pathname.includes(conversationId)) { + setDocumentTitle(title); + } + + if (setConversation && !isAddedRequest) { + setConversation((prevState) => { + if (!prevState) { + return prevState; + } + if (prevState.conversationId && prevState.conversationId !== conversationId) { + return prevState; + } + return { + ...prevState, + conversationId, + ...titleState, + title, + }; + }); + } + }, + [queryClient, location.pathname, setConversation, isAddedRequest], + ); + + return { messageHandler, syncHandler, createdHandler, titleHandler }; +}