diff --git a/api/server/controllers/agents/request.js b/api/server/controllers/agents/request.js index 80ff52fb3a..7f588f3472 100644 --- a/api/server/controllers/agents/request.js +++ b/api/server/controllers/agents/request.js @@ -140,6 +140,10 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit client = result.client; + if (client?.sender) { + GenerationJobManager.updateMetadata(streamId, { sender: client.sender }); + } + res.json({ streamId, status: 'started' }); let conversationId = reqConversationId; diff --git a/api/server/routes/agents/index.js b/api/server/routes/agents/index.js index 5e727eb90c..44a04e8db9 100644 --- a/api/server/routes/agents/index.js +++ b/api/server/routes/agents/index.js @@ -156,12 +156,27 @@ router.post('/chat/abort', (req, res) => { logger.debug(`[AgentStream] Method: ${req.method}, Path: ${req.path}`); logger.debug(`[AgentStream] Body:`, req.body); - const { streamId, abortKey } = req.body; + const { streamId, conversationId, abortKey } = req.body; + + // Try to find job by streamId first, then by conversationId, then by abortKey + let jobStreamId = streamId; + let job = jobStreamId ? GenerationJobManager.getJob(jobStreamId) : null; + + if (!job && conversationId) { + job = GenerationJobManager.getJobByConversation(conversationId); + if (job) { + jobStreamId = job.streamId; + } + } + + if (!job && abortKey) { + jobStreamId = abortKey.split(':')[0]; + job = GenerationJobManager.getJob(jobStreamId); + } - const jobStreamId = streamId || abortKey?.split(':')?.[0]; logger.debug(`[AgentStream] Computed jobStreamId: ${jobStreamId}`); - if (jobStreamId && GenerationJobManager.hasJob(jobStreamId)) { + if (job && jobStreamId) { logger.debug(`[AgentStream] Job found, aborting: ${jobStreamId}`); GenerationJobManager.abortJob(jobStreamId); logger.debug(`[AgentStream] Job aborted successfully: ${jobStreamId}`); diff --git a/client/src/data-provider/SSE/index.ts b/client/src/data-provider/SSE/index.ts new file mode 100644 index 0000000000..d0720956a0 --- /dev/null +++ b/client/src/data-provider/SSE/index.ts @@ -0,0 +1,2 @@ +export * from './queries'; +export * from './mutations'; diff --git a/client/src/data-provider/SSE/mutations.ts b/client/src/data-provider/SSE/mutations.ts new file mode 100644 index 0000000000..f24fed1b07 --- /dev/null +++ b/client/src/data-provider/SSE/mutations.ts @@ -0,0 +1,39 @@ +import { useMutation } from '@tanstack/react-query'; +import { request } from 'librechat-data-provider'; + +export interface AbortStreamParams { + /** The stream ID to abort (if known) */ + streamId?: string; + /** The conversation ID to abort (backend will look up the job) */ + conversationId?: string; +} + +export interface AbortStreamResponse { + success: boolean; + aborted?: string; + error?: string; +} + +/** + * Abort an ongoing generation stream. + * The backend will emit a `done` event with `aborted: true` to the SSE stream, + * allowing the client to handle cleanup via the normal event flow. + * + * Can pass either streamId or conversationId - backend will find the job. + */ +export const abortStream = async (params: AbortStreamParams): Promise => { + console.log('[abortStream] Calling abort endpoint with params:', params); + const result = (await request.post('/api/agents/chat/abort', params)) as AbortStreamResponse; + console.log('[abortStream] Abort response:', result); + return result; +}; + +/** + * React Query mutation hook for aborting a generation stream. + * Use this when the user explicitly clicks the stop button. + */ +export function useAbortStreamMutation() { + return useMutation({ + mutationFn: abortStream, + }); +} diff --git a/client/src/data-provider/queries/streamStatus.ts b/client/src/data-provider/SSE/queries.ts similarity index 77% rename from client/src/data-provider/queries/streamStatus.ts rename to client/src/data-provider/SSE/queries.ts index 4b34290ba6..45bc6cacae 100644 --- a/client/src/data-provider/queries/streamStatus.ts +++ b/client/src/data-provider/SSE/queries.ts @@ -1,5 +1,6 @@ import { useQuery } from '@tanstack/react-query'; import { request } from 'librechat-data-provider'; +import type { Agents } from 'librechat-data-provider'; export interface StreamStatusResponse { active: boolean; @@ -8,6 +9,7 @@ export interface StreamStatusResponse { chunkCount?: number; aggregatedContent?: Array<{ type: string; text?: string }>; createdAt?: number; + resumeState?: Agents.ResumeState; } /** @@ -19,8 +21,12 @@ export const streamStatusQueryKey = (conversationId: string) => ['streamStatus', * Fetch stream status for a conversation */ export const fetchStreamStatus = async (conversationId: string): Promise => { - const response = await request.get(`/api/agents/chat/status/${conversationId}`); - return response.data; + console.log('[fetchStreamStatus] Fetching status for:', conversationId); + const result = await request.get( + `/api/agents/chat/status/${conversationId}`, + ); + console.log('[fetchStreamStatus] Result:', result); + return result; }; /** diff --git a/client/src/data-provider/index.ts b/client/src/data-provider/index.ts index d32fb46d0b..bfc87bb232 100644 --- a/client/src/data-provider/index.ts +++ b/client/src/data-provider/index.ts @@ -15,3 +15,4 @@ export * from './queries'; export * from './roles'; export * from './tags'; export * from './MCP'; +export * from './SSE'; diff --git a/client/src/hooks/Chat/useChatHelpers.ts b/client/src/hooks/Chat/useChatHelpers.ts index b5ab9aee27..f52ff1f17e 100644 --- a/client/src/hooks/Chat/useChatHelpers.ts +++ b/client/src/hooks/Chat/useChatHelpers.ts @@ -1,10 +1,10 @@ import { useCallback, useState } from 'react'; -import { QueryKeys } from 'librechat-data-provider'; +import { QueryKeys, isAssistantsEndpoint } from 'librechat-data-provider'; import { useQueryClient } from '@tanstack/react-query'; import { useRecoilState, useResetRecoilState, useSetRecoilState } from 'recoil'; import type { TMessage } from 'librechat-data-provider'; +import { useAbortStreamMutation, useGetMessagesByConvoId } from '~/data-provider'; import useChatFunctions from '~/hooks/Chat/useChatFunctions'; -import { useGetMessagesByConvoId } from '~/data-provider'; import { useAuthContext } from '~/hooks/AuthContext'; import useNewConvo from '~/hooks/useNewConvo'; import store from '~/store'; @@ -17,11 +17,12 @@ export default function useChatHelpers(index = 0, paramId?: string) { const queryClient = useQueryClient(); const { isAuthenticated } = useAuthContext(); + const abortMutation = useAbortStreamMutation(); const { newConversation } = useNewConvo(index); const { useCreateConversationAtom } = store; const { conversation, setConversation } = useCreateConversationAtom(index); - const { conversationId } = conversation ?? {}; + const { conversationId, endpoint, endpointType } = conversation ?? {}; const queryParam = paramId === 'new' ? paramId : (conversationId ?? paramId ?? ''); @@ -107,7 +108,43 @@ export default function useChatHelpers(index = 0, paramId?: string) { } }; - const stopGenerating = () => clearAllSubmissions(); + /** + * Stop generation - for non-assistants endpoints, calls abort endpoint first. + * The abort endpoint will cause the backend to emit a `done` event with `aborted: true`, + * which will be handled by the SSE event handler to clean up UI. + * Assistants endpoint has its own abort mechanism via useEventHandlers.abortConversation. + */ + const stopGenerating = useCallback(async () => { + const actualEndpoint = endpointType ?? endpoint; + const isAssistants = isAssistantsEndpoint(actualEndpoint); + console.log('[useChatHelpers] stopGenerating called', { + conversationId, + endpoint, + endpointType, + actualEndpoint, + isAssistants, + }); + + // For non-assistants endpoints (using resumable streams), call abort endpoint first + if (conversationId && !isAssistants) { + try { + console.log('[useChatHelpers] Calling abort mutation for:', conversationId); + await abortMutation.mutateAsync({ conversationId }); + console.log('[useChatHelpers] Abort mutation succeeded'); + // The SSE will receive a `done` event with `aborted: true` and clean up + // We still clear submissions as a fallback + clearAllSubmissions(); + } catch (error) { + console.error('[useChatHelpers] Abort failed:', error); + // Fall back to clearing submissions + clearAllSubmissions(); + } + } else { + // For assistants endpoints, just clear submissions (existing behavior) + console.log('[useChatHelpers] Assistants endpoint, just clearing submissions'); + clearAllSubmissions(); + } + }, [conversationId, endpoint, endpointType, abortMutation, clearAllSubmissions]); const handleStopGenerating = (e: React.MouseEvent) => { e.preventDefault(); diff --git a/client/src/hooks/SSE/useResumableSSE.ts b/client/src/hooks/SSE/useResumableSSE.ts index 7bbc35e06a..1495a81b98 100644 --- a/client/src/hooks/SSE/useResumableSSE.ts +++ b/client/src/hooks/SSE/useResumableSSE.ts @@ -11,7 +11,6 @@ import { } from 'librechat-data-provider'; import type { TMessage, TPayload, TSubmission, EventSubmission } from 'librechat-data-provider'; import type { EventHandlerParams } from './useEventHandlers'; -import type { TResData } from '~/common'; import { useGenTitleMutation, useGetStartupConfig, useGetUserBalance } from '~/data-provider'; import { useAuthContext } from '~/hooks/AuthContext'; import useEventHandlers from './useEventHandlers'; @@ -43,6 +42,11 @@ const MAX_RETRIES = 5; * Hook for resumable SSE streams. * Separates generation start (POST) from stream subscription (GET EventSource). * Supports auto-reconnection with exponential backoff. + * + * Key behavior: + * - Navigation away does NOT abort the generation (just closes SSE) + * - Only explicit abort (via stop button → backend abort endpoint) stops generation + * - Backend emits `done` event with `aborted: true` on abort, handled via finalHandler */ export default function useResumableSSE( submission: TSubmission | null, @@ -83,7 +87,6 @@ export default function useResumableSSE( contentHandler, createdHandler, attachmentHandler, - abortConversation, } = useEventHandlers({ genTitle, setMessages, @@ -104,6 +107,7 @@ export default function useResumableSSE( /** * Subscribe to stream via SSE library (supports custom headers) + * Follows same auth pattern as useSSE */ const subscribeToStream = useCallback( (currentStreamId: string, currentSubmission: TSubmission) => { @@ -131,6 +135,11 @@ export default function useResumableSSE( const data = JSON.parse(e.data); if (data.final != null) { + console.log('[ResumableSSE] Received FINAL event', { + aborted: data.aborted, + conversationId: data.conversation?.conversationId, + hasResponseMessage: !!data.responseMessage, + }); clearDraft(currentSubmission.conversation?.conversationId); try { finalHandler(data, currentSubmission as EventSubmission); @@ -146,6 +155,10 @@ export default function useResumableSSE( } if (data.created != null) { + console.log('[ResumableSSE] Received CREATED event', { + messageId: data.message?.messageId, + conversationId: data.message?.conversationId, + }); const runId = v4(); setActiveRunId(runId); userMessage = { @@ -171,6 +184,10 @@ export default function useResumableSSE( } if (data.sync != null) { + console.log('[ResumableSSE] Received SYNC event', { + conversationId: data.conversationId, + hasResumeState: !!data.resumeState, + }); const runId = v4(); setActiveRunId(runId); syncHandler(data, { ...currentSubmission, userMessage } as EventSubmission); @@ -200,68 +217,33 @@ export default function useResumableSSE( } }); - // Handle cancel event (triggered when stop button is clicked) - sse.addEventListener('cancel', async () => { - console.log('[ResumableSSE] Cancel requested, aborting job'); - sse.close(); - - // Call abort endpoint to stop backend generation - try { - await fetch('/api/agents/chat/abort', { - method: 'POST', - headers: { - 'Content-Type': 'application/json', - Authorization: `Bearer ${token}`, - }, - body: JSON.stringify({ streamId: currentStreamId }), - }); - } catch (error) { - console.error('[ResumableSSE] Error aborting job:', error); - } - - // Handle UI cleanup via abortConversation - const latestMessages = getMessages(); - const conversationId = latestMessages?.[latestMessages.length - 1]?.conversationId; - try { - await abortConversation( - conversationId ?? - userMessage.conversationId ?? - currentSubmission.conversation?.conversationId ?? - '', - currentSubmission as EventSubmission, - latestMessages, - ); - } catch (error) { - console.error('[ResumableSSE] Error during abort:', error); - setIsSubmitting(false); - setShowStopButton(false); - } - setStreamId(null); - }); - sse.addEventListener('error', async (e: MessageEvent) => { - console.log('[ResumableSSE] Stream error, connection closed'); - sse.close(); + console.log('[ResumableSSE] Stream error'); + (startupConfig?.balance?.enabled ?? false) && balanceQuery.refetch(); - // Check for 401 and try to refresh token + // Check for 401 and try to refresh token (same pattern as useSSE) /* @ts-ignore */ if (e.responseCode === 401) { try { const refreshResponse = await request.refreshToken(); const newToken = refreshResponse?.token ?? ''; - if (newToken) { - request.dispatchTokenUpdatedEvent(newToken); - // Retry with new token - if (submissionRef.current) { - subscribeToStream(currentStreamId, submissionRef.current); - } - return; + if (!newToken) { + throw new Error('Token refresh failed.'); } + // Update headers on same SSE instance and retry (like useSSE) + sse.headers = { + Authorization: `Bearer ${newToken}`, + }; + request.dispatchTokenUpdatedEvent(newToken); + sse.stream(); + return; } catch (error) { console.log('[ResumableSSE] Token refresh failed:', error); } } + sse.close(); + if (reconnectAttemptRef.current < MAX_RETRIES) { reconnectAttemptRef.current++; const delay = Math.min(1000 * Math.pow(2, reconnectAttemptRef.current - 1), 30000); @@ -303,13 +285,12 @@ export default function useResumableSSE( setIsSubmitting, startupConfig?.balance?.enabled, balanceQuery, - abortConversation, - getMessages, ], ); /** * Start generation (POST request that returns streamId) + * Uses request.post which has axios interceptors for automatic token refresh */ const startGeneration = useCallback( async (currentSubmission: TSubmission): Promise => { @@ -324,24 +305,10 @@ export default function useResumableSSE( : `${payloadData.server}?resumable=true`; try { - const response = await fetch(url, { - method: 'POST', - headers: { - 'Content-Type': 'application/json', - Authorization: `Bearer ${token}`, - }, - body: JSON.stringify(payload), - }); - - if (!response.ok) { - const errorData = await response.json().catch(() => ({})); - throw new Error(errorData.error || `Failed to start generation: ${response.statusText}`); - } - - const { streamId: newStreamId } = await response.json(); - console.log('[ResumableSSE] Generation started:', { streamId: newStreamId }); - - return newStreamId; + // Use request.post which handles auth token refresh via axios interceptors + const data = (await request.post(url, payload)) as { streamId: string }; + console.log('[ResumableSSE] Generation started:', { streamId: data.streamId }); + return data.streamId; } catch (error) { console.error('[ResumableSSE] Error starting generation:', error); errorHandler({ data: undefined, submission: currentSubmission as EventSubmission }); @@ -349,15 +316,18 @@ export default function useResumableSSE( return null; } }, - [token, clearStepMaps, errorHandler, setIsSubmitting], + [clearStepMaps, errorHandler, setIsSubmitting], ); useEffect(() => { if (!submission || Object.keys(submission).length === 0) { + console.log('[ResumableSSE] No submission, cleaning up'); + // Clear reconnect timeout if submission is cleared if (reconnectTimeoutRef.current) { clearTimeout(reconnectTimeoutRef.current); reconnectTimeoutRef.current = null; } + // Close SSE but do NOT dispatch cancel - navigation should not abort if (sseRef.current) { sseRef.current.close(); sseRef.current = null; @@ -368,36 +338,56 @@ export default function useResumableSSE( return; } + const resumeStreamId = (submission as TSubmission & { resumeStreamId?: string }).resumeStreamId; + console.log('[ResumableSSE] Effect triggered', { + conversationId: submission.conversation?.conversationId, + hasResumeStreamId: !!resumeStreamId, + resumeStreamId, + userMessageId: submission.userMessage?.messageId, + }); + submissionRef.current = submission; const initStream = async () => { setIsSubmitting(true); + setShowStopButton(true); - const newStreamId = await startGeneration(submission); - if (newStreamId) { - setStreamId(newStreamId); - subscribeToStream(newStreamId, submission); + if (resumeStreamId) { + // Resume: just subscribe to existing stream, don't start new generation + console.log('[ResumableSSE] Resuming existing stream:', resumeStreamId); + setStreamId(resumeStreamId); + subscribeToStream(resumeStreamId, submission); + } else { + // New generation: start and then subscribe + console.log('[ResumableSSE] Starting NEW generation'); + const newStreamId = await startGeneration(submission); + if (newStreamId) { + setStreamId(newStreamId); + subscribeToStream(newStreamId, submission); + } else { + console.error('[ResumableSSE] Failed to get streamId from startGeneration'); + } } }; initStream(); return () => { + console.log('[ResumableSSE] Cleanup - closing SSE, resetting UI state'); + // Cleanup on unmount/navigation - close connection but DO NOT abort backend + // Reset UI state so it doesn't leak to other conversations + // If user returns to this conversation, useResumeOnLoad will restore the state if (reconnectTimeoutRef.current) { clearTimeout(reconnectTimeoutRef.current); reconnectTimeoutRef.current = null; } if (sseRef.current) { - const isCancelled = sseRef.current.readyState <= 1; sseRef.current.close(); - if (isCancelled) { - // Dispatch cancel event to trigger abort - const e = new Event('cancel'); - /* @ts-ignore */ - sseRef.current.dispatchEvent(e); - } sseRef.current = null; } + // Reset UI state on cleanup - useResumeOnLoad will restore if needed + setIsSubmitting(false); + setShowStopButton(false); }; // eslint-disable-next-line react-hooks/exhaustive-deps }, [submission]); diff --git a/client/src/hooks/SSE/useResumeOnLoad.ts b/client/src/hooks/SSE/useResumeOnLoad.ts index 370d8c4ca2..4349c219d7 100644 --- a/client/src/hooks/SSE/useResumeOnLoad.ts +++ b/client/src/hooks/SSE/useResumeOnLoad.ts @@ -2,6 +2,7 @@ import { useEffect, useRef } from 'react'; import { useSetRecoilState, useRecoilValue } from 'recoil'; import { Constants, tMessageSchema } from 'librechat-data-provider'; import type { TMessage, TConversation, TSubmission, Agents } from 'librechat-data-provider'; +import { useStreamStatus } from '~/data-provider'; import store from '~/store'; /** @@ -77,7 +78,9 @@ function buildSubmissionFromResumeState( isRegenerate: false, isTemporary: false, endpointOption: {}, - } as TSubmission; + // Signal to useResumableSSE to subscribe to existing stream instead of starting new + resumeStreamId: streamId, + } as TSubmission & { resumeStreamId: string }; } /** @@ -97,67 +100,121 @@ export default function useResumeOnLoad( const resumableEnabled = useRecoilValue(store.resumableStreams); const setSubmission = useSetRecoilState(store.submissionByIndex(runIndex)); const currentSubmission = useRecoilValue(store.submissionByIndex(runIndex)); - const hasResumedRef = useRef(null); + // Track conversations we've already processed (either resumed or skipped) + const processedConvoRef = useRef(null); // Check for active stream when conversation changes - // const { data: streamStatus, isSuccess } = useStreamStatus( - // conversationId, - // resumableEnabled && !currentSubmission, // Only check if no active submission - // ); + // Only check if resumable is enabled and no active submission + const shouldCheck = + resumableEnabled && + !currentSubmission && + !!conversationId && + conversationId !== Constants.NEW_CONVO && + processedConvoRef.current !== conversationId; // Don't re-check processed convos + + const { data: streamStatus, isSuccess } = useStreamStatus(conversationId, shouldCheck); useEffect(() => { - // if (!resumableEnabled || !conversationId || !isSuccess || !streamStatus) { - if (!resumableEnabled || !conversationId) { + console.log('[ResumeOnLoad] Effect check', { + resumableEnabled, + conversationId, + hasCurrentSubmission: !!currentSubmission, + currentSubmissionConvoId: currentSubmission?.conversation?.conversationId, + isSuccess, + streamStatusActive: streamStatus?.active, + streamStatusStreamId: streamStatus?.streamId, + processedConvoRef: processedConvoRef.current, + }); + + if (!resumableEnabled || !conversationId || conversationId === Constants.NEW_CONVO) { + console.log('[ResumeOnLoad] Skipping - not enabled or new convo'); return; } - // Don't resume if we already have an active submission + // Don't resume if we already have an active submission (we started it ourselves) if (currentSubmission) { + console.log('[ResumeOnLoad] Skipping - already have active submission, marking as processed'); + // Mark as processed so we don't try again + processedConvoRef.current = conversationId; return; } - // Don't resume the same conversation twice - if (hasResumedRef.current === conversationId) { + // Wait for stream status query to complete + if (!isSuccess || !streamStatus) { + console.log('[ResumeOnLoad] Waiting for stream status query'); return; } + // Don't process the same conversation twice + if (processedConvoRef.current === conversationId) { + console.log('[ResumeOnLoad] Skipping - already processed this conversation'); + return; + } + + // Mark as processed immediately to prevent race conditions + processedConvoRef.current = conversationId; + // Check if there's an active job to resume - // if (!streamStatus.active || !streamStatus.streamId) { - // return; - // } + if (!streamStatus.active || !streamStatus.streamId) { + console.log('[ResumeOnLoad] No active job to resume for:', conversationId); + return; + } - // console.log('[ResumeOnLoad] Found active job, creating submission...', { - // streamId: streamStatus.streamId, - // status: streamStatus.status, - // }); - - hasResumedRef.current = conversationId; + console.log('[ResumeOnLoad] Found active job, creating submission...', { + streamId: streamStatus.streamId, + status: streamStatus.status, + resumeState: streamStatus.resumeState, + }); const messages = getMessages() || []; - // Minimal submission without resume state - const lastMessage = messages[messages.length - 1]; - const submission: TSubmission = { - messages, - userMessage: lastMessage ?? ({ messageId: 'resume', conversationId, text: '' } as TMessage), - initialResponse: { - messageId: 'resume_', + // Build submission from resume state if available + if (streamStatus.resumeState) { + const submission = buildSubmissionFromResumeState( + streamStatus.resumeState, + streamStatus.streamId, + messages, conversationId, - text: '', - content: [{ type: 'text', text: '' }], - } as TMessage, - conversation: { conversationId, title: 'Resumed Chat' } as TConversation, - isRegenerate: false, - isTemporary: false, - endpointOption: {}, - } as TSubmission; - setSubmission(submission); - }, [conversationId, resumableEnabled, currentSubmission, getMessages, setSubmission]); + ); + 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_', + conversationId, + text: '', + content: streamStatus.aggregatedContent ?? [{ type: 'text', text: '' }], + } as TMessage, + conversation: { conversationId, title: 'Resumed Chat' } as TConversation, + isRegenerate: false, + isTemporary: false, + endpointOption: {}, + // Signal to useResumableSSE to subscribe to existing stream instead of starting new + resumeStreamId: streamStatus.streamId, + } as TSubmission & { resumeStreamId: string }; + setSubmission(submission); + } + }, [ + conversationId, + resumableEnabled, + currentSubmission, + isSuccess, + streamStatus, + getMessages, + setSubmission, + ]); - // Reset hasResumedRef when conversation changes + // Reset processedConvoRef when conversation changes to a different one useEffect(() => { - if (conversationId !== hasResumedRef.current) { - hasResumedRef.current = null; + if (conversationId && conversationId !== processedConvoRef.current) { + // Only reset if we're navigating to a DIFFERENT conversation + // This allows re-checking when navigating back + processedConvoRef.current = null; } }, [conversationId]); } diff --git a/packages/api/src/stream/GenerationJobManager.ts b/packages/api/src/stream/GenerationJobManager.ts index 9df9c4b1ad..e6af5f8161 100644 --- a/packages/api/src/stream/GenerationJobManager.ts +++ b/packages/api/src/stream/GenerationJobManager.ts @@ -177,6 +177,7 @@ class GenerationJobManagerClass { ); // Create a final event for abort so clients can properly handle UI cleanup + const userMessageId = job.metadata.userMessage?.messageId; const abortFinalEvent = { final: true, conversation: { @@ -185,18 +186,23 @@ class GenerationJobManagerClass { title: 'New Chat', requestMessage: job.metadata.userMessage ? { - messageId: job.metadata.userMessage.messageId, + messageId: userMessageId, + parentMessageId: job.metadata.userMessage.parentMessageId, conversationId: job.metadata.conversationId, text: job.metadata.userMessage.text ?? '', + isCreatedByUser: true, } : null, responseMessage: { - messageId: - job.metadata.responseMessageId ?? `${job.metadata.userMessage?.messageId ?? 'aborted'}_`, + messageId: job.metadata.responseMessageId ?? `${userMessageId ?? 'aborted'}_`, + parentMessageId: userMessageId, // Link response to user message conversationId: job.metadata.conversationId, content: job.aggregatedContent ?? [], + sender: job.metadata.sender ?? 'AI', unfinished: true, - error: true, + /** Not an error - the job was intentionally aborted */ + error: false, + isCreatedByUser: false, }, aborted: true, } as unknown as ServerSentEvent; diff --git a/packages/api/src/stream/types.ts b/packages/api/src/stream/types.ts index 337ebcc17c..ac5e49087f 100644 --- a/packages/api/src/stream/types.ts +++ b/packages/api/src/stream/types.ts @@ -9,6 +9,8 @@ export interface GenerationJobMetadata { userMessage?: Agents.UserMessageMeta; /** Response message ID for tracking */ responseMessageId?: string; + /** Sender label for the response (e.g., "GPT-4.1", "Claude") */ + sender?: string; } export type GenerationJobStatus = 'running' | 'complete' | 'error' | 'aborted';