mirror of
https://github.com/danny-avila/LibreChat.git
synced 2025-12-21 19:00:13 +01:00
feat: Add active job management for user and show progress in conversation list
- Implemented a new endpoint to retrieve active generation job IDs for the current user, enhancing user experience by allowing visibility of ongoing tasks. - Integrated active job tracking in the Conversations component, displaying generation indicators based on active jobs. - Optimized job management in the GenerationJobManager and InMemoryJobStore to support user-specific job queries, ensuring efficient resource handling and cleanup. - Updated relevant components and hooks to utilize the new active jobs feature, improving overall application responsiveness and user feedback.
This commit is contained in:
parent
c69bb0b14d
commit
5a2aafb7d0
11 changed files with 279 additions and 20 deletions
|
|
@ -113,6 +113,17 @@ router.get('/chat/stream/:streamId', async (req, res) => {
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @route GET /chat/active
|
||||||
|
* @desc Get all active generation job IDs for the current user
|
||||||
|
* @access Private
|
||||||
|
* @returns { activeJobIds: string[] }
|
||||||
|
*/
|
||||||
|
router.get('/chat/active', async (req, res) => {
|
||||||
|
const activeJobIds = await GenerationJobManager.getActiveJobIdsForUser(req.user.id);
|
||||||
|
res.json({ activeJobIds });
|
||||||
|
});
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @route GET /chat/status/:conversationId
|
* @route GET /chat/status/:conversationId
|
||||||
* @desc Check if there's an active generation job for a conversation
|
* @desc Check if there's an active generation job for a conversation
|
||||||
|
|
|
||||||
|
|
@ -7,6 +7,7 @@ import { List, AutoSizer, CellMeasurer, CellMeasurerCache } from 'react-virtuali
|
||||||
import type { TConversation } from 'librechat-data-provider';
|
import type { TConversation } from 'librechat-data-provider';
|
||||||
import { useLocalize, TranslationKeys, useFavorites, useShowMarketplace } from '~/hooks';
|
import { useLocalize, TranslationKeys, useFavorites, useShowMarketplace } from '~/hooks';
|
||||||
import FavoritesList from '~/components/Nav/Favorites/FavoritesList';
|
import FavoritesList from '~/components/Nav/Favorites/FavoritesList';
|
||||||
|
import { useActiveJobs } from '~/data-provider';
|
||||||
import { groupConversationsByDate, cn } from '~/utils';
|
import { groupConversationsByDate, cn } from '~/utils';
|
||||||
import Convo from './Convo';
|
import Convo from './Convo';
|
||||||
import store from '~/store';
|
import store from '~/store';
|
||||||
|
|
@ -120,18 +121,28 @@ const MemoizedConvo = memo(
|
||||||
conversation,
|
conversation,
|
||||||
retainView,
|
retainView,
|
||||||
toggleNav,
|
toggleNav,
|
||||||
|
isGenerating,
|
||||||
}: {
|
}: {
|
||||||
conversation: TConversation;
|
conversation: TConversation;
|
||||||
retainView: () => void;
|
retainView: () => void;
|
||||||
toggleNav: () => void;
|
toggleNav: () => void;
|
||||||
|
isGenerating: boolean;
|
||||||
}) => {
|
}) => {
|
||||||
return <Convo conversation={conversation} retainView={retainView} toggleNav={toggleNav} />;
|
return (
|
||||||
|
<Convo
|
||||||
|
conversation={conversation}
|
||||||
|
retainView={retainView}
|
||||||
|
toggleNav={toggleNav}
|
||||||
|
isGenerating={isGenerating}
|
||||||
|
/>
|
||||||
|
);
|
||||||
},
|
},
|
||||||
(prevProps, nextProps) => {
|
(prevProps, nextProps) => {
|
||||||
return (
|
return (
|
||||||
prevProps.conversation.conversationId === nextProps.conversation.conversationId &&
|
prevProps.conversation.conversationId === nextProps.conversation.conversationId &&
|
||||||
prevProps.conversation.title === nextProps.conversation.title &&
|
prevProps.conversation.title === nextProps.conversation.title &&
|
||||||
prevProps.conversation.endpoint === nextProps.conversation.endpoint
|
prevProps.conversation.endpoint === nextProps.conversation.endpoint &&
|
||||||
|
prevProps.isGenerating === nextProps.isGenerating
|
||||||
);
|
);
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
|
|
@ -149,11 +160,19 @@ const Conversations: FC<ConversationsProps> = ({
|
||||||
}) => {
|
}) => {
|
||||||
const localize = useLocalize();
|
const localize = useLocalize();
|
||||||
const search = useRecoilValue(store.search);
|
const search = useRecoilValue(store.search);
|
||||||
|
const resumableEnabled = useRecoilValue(store.resumableStreams);
|
||||||
const { favorites, isLoading: isFavoritesLoading } = useFavorites();
|
const { favorites, isLoading: isFavoritesLoading } = useFavorites();
|
||||||
const isSmallScreen = useMediaQuery('(max-width: 768px)');
|
const isSmallScreen = useMediaQuery('(max-width: 768px)');
|
||||||
const convoHeight = isSmallScreen ? 44 : 34;
|
const convoHeight = isSmallScreen ? 44 : 34;
|
||||||
const showAgentMarketplace = useShowMarketplace();
|
const showAgentMarketplace = useShowMarketplace();
|
||||||
|
|
||||||
|
// Fetch active job IDs for showing generation indicators
|
||||||
|
const { data: activeJobsData } = useActiveJobs(resumableEnabled);
|
||||||
|
const activeJobIds = useMemo(
|
||||||
|
() => new Set(activeJobsData?.activeJobIds ?? []),
|
||||||
|
[activeJobsData?.activeJobIds],
|
||||||
|
);
|
||||||
|
|
||||||
// Determine if FavoritesList will render content
|
// Determine if FavoritesList will render content
|
||||||
const shouldShowFavorites =
|
const shouldShowFavorites =
|
||||||
!search.query && (isFavoritesLoading || favorites.length > 0 || showAgentMarketplace);
|
!search.query && (isFavoritesLoading || favorites.length > 0 || showAgentMarketplace);
|
||||||
|
|
@ -292,9 +311,15 @@ const Conversations: FC<ConversationsProps> = ({
|
||||||
}
|
}
|
||||||
|
|
||||||
if (item.type === 'convo') {
|
if (item.type === 'convo') {
|
||||||
|
const isGenerating = activeJobIds.has(item.convo.conversationId ?? '');
|
||||||
return (
|
return (
|
||||||
<MeasuredRow key={key} {...rowProps}>
|
<MeasuredRow key={key} {...rowProps}>
|
||||||
<MemoizedConvo conversation={item.convo} retainView={moveToTop} toggleNav={toggleNav} />
|
<MemoizedConvo
|
||||||
|
conversation={item.convo}
|
||||||
|
retainView={moveToTop}
|
||||||
|
toggleNav={toggleNav}
|
||||||
|
isGenerating={isGenerating}
|
||||||
|
/>
|
||||||
</MeasuredRow>
|
</MeasuredRow>
|
||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
@ -311,6 +336,7 @@ const Conversations: FC<ConversationsProps> = ({
|
||||||
isChatsExpanded,
|
isChatsExpanded,
|
||||||
setIsChatsExpanded,
|
setIsChatsExpanded,
|
||||||
shouldShowFavorites,
|
shouldShowFavorites,
|
||||||
|
activeJobIds,
|
||||||
],
|
],
|
||||||
);
|
);
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -19,9 +19,15 @@ interface ConversationProps {
|
||||||
conversation: TConversation;
|
conversation: TConversation;
|
||||||
retainView: () => void;
|
retainView: () => void;
|
||||||
toggleNav: () => void;
|
toggleNav: () => void;
|
||||||
|
isGenerating?: boolean;
|
||||||
}
|
}
|
||||||
|
|
||||||
export default function Conversation({ conversation, retainView, toggleNav }: ConversationProps) {
|
export default function Conversation({
|
||||||
|
conversation,
|
||||||
|
retainView,
|
||||||
|
toggleNav,
|
||||||
|
isGenerating = false,
|
||||||
|
}: ConversationProps) {
|
||||||
const params = useParams();
|
const params = useParams();
|
||||||
const localize = useLocalize();
|
const localize = useLocalize();
|
||||||
const { showToast } = useToastContext();
|
const { showToast } = useToastContext();
|
||||||
|
|
@ -182,12 +188,35 @@ export default function Conversation({ conversation, retainView, toggleNav }: Co
|
||||||
isSmallScreen={isSmallScreen}
|
isSmallScreen={isSmallScreen}
|
||||||
localize={localize}
|
localize={localize}
|
||||||
>
|
>
|
||||||
<EndpointIcon
|
{isGenerating ? (
|
||||||
conversation={conversation}
|
<svg
|
||||||
endpointsConfig={endpointsConfig}
|
className="h-5 w-5 flex-shrink-0 animate-spin text-text-primary"
|
||||||
size={20}
|
viewBox="0 0 24 24"
|
||||||
context="menu-item"
|
fill="none"
|
||||||
/>
|
aria-label={localize('com_ui_generating')}
|
||||||
|
>
|
||||||
|
<circle
|
||||||
|
className="opacity-25"
|
||||||
|
cx="12"
|
||||||
|
cy="12"
|
||||||
|
r="10"
|
||||||
|
stroke="currentColor"
|
||||||
|
strokeWidth="3"
|
||||||
|
/>
|
||||||
|
<path
|
||||||
|
className="opacity-75"
|
||||||
|
fill="currentColor"
|
||||||
|
d="M4 12a8 8 0 018-8V0C5.373 0 0 5.373 0 12h4zm2 5.291A7.962 7.962 0 014 12H0c0 3.042 1.135 5.824 3 7.938l3-2.647z"
|
||||||
|
/>
|
||||||
|
</svg>
|
||||||
|
) : (
|
||||||
|
<EndpointIcon
|
||||||
|
conversation={conversation}
|
||||||
|
endpointsConfig={endpointsConfig}
|
||||||
|
size={20}
|
||||||
|
context="menu-item"
|
||||||
|
/>
|
||||||
|
)}
|
||||||
</ConvoLink>
|
</ConvoLink>
|
||||||
)}
|
)}
|
||||||
<div
|
<div
|
||||||
|
|
|
||||||
|
|
@ -1,5 +1,5 @@
|
||||||
import { useQuery } from '@tanstack/react-query';
|
import { useQuery } from '@tanstack/react-query';
|
||||||
import { request } from 'librechat-data-provider';
|
import { QueryKeys, request, dataService } from 'librechat-data-provider';
|
||||||
import type { Agents } from 'librechat-data-provider';
|
import type { Agents } from 'librechat-data-provider';
|
||||||
|
|
||||||
export interface StreamStatusResponse {
|
export interface StreamStatusResponse {
|
||||||
|
|
@ -43,3 +43,35 @@ export function useStreamStatus(conversationId: string | undefined, enabled = tr
|
||||||
retry: false,
|
retry: false,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Query key for active jobs
|
||||||
|
*/
|
||||||
|
export const activeJobsQueryKey = [QueryKeys.activeJobs] as const;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* React Query hook for getting all active job IDs for the current user.
|
||||||
|
* Used to show generation indicators in the conversation list.
|
||||||
|
*
|
||||||
|
* Key behaviors:
|
||||||
|
* - Fetches on mount to get initial state (handles page refresh)
|
||||||
|
* - Refetches on window focus (handles multi-tab scenarios)
|
||||||
|
* - Optimistic updates from useResumableSSE when jobs start/complete
|
||||||
|
* - Polls every 5s while there are active jobs (catches completions when navigated away)
|
||||||
|
*/
|
||||||
|
export function useActiveJobs(enabled = true) {
|
||||||
|
return useQuery({
|
||||||
|
queryKey: activeJobsQueryKey,
|
||||||
|
queryFn: () => dataService.getActiveJobs(),
|
||||||
|
enabled,
|
||||||
|
staleTime: 5_000, // 5s - short to catch completions quickly
|
||||||
|
refetchOnMount: true,
|
||||||
|
refetchOnWindowFocus: true, // Catch up on tab switch (multi-tab scenario)
|
||||||
|
// Poll every 5s while there are active jobs to catch completions when navigated away
|
||||||
|
refetchInterval: (data) => {
|
||||||
|
const hasActiveJobs = (data?.activeJobIds?.length ?? 0) > 0;
|
||||||
|
return hasActiveJobs ? 5_000 : false;
|
||||||
|
},
|
||||||
|
retry: false,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
|
||||||
|
|
@ -2,6 +2,7 @@ import { useEffect, useState, useRef, useCallback } from 'react';
|
||||||
import { v4 } from 'uuid';
|
import { v4 } from 'uuid';
|
||||||
import { SSE } from 'sse.js';
|
import { SSE } from 'sse.js';
|
||||||
import { useSetRecoilState } from 'recoil';
|
import { useSetRecoilState } from 'recoil';
|
||||||
|
import { useQueryClient } from '@tanstack/react-query';
|
||||||
import {
|
import {
|
||||||
request,
|
request,
|
||||||
Constants,
|
Constants,
|
||||||
|
|
@ -12,10 +13,16 @@ import {
|
||||||
import type { TMessage, TPayload, TSubmission, EventSubmission } from 'librechat-data-provider';
|
import type { TMessage, TPayload, TSubmission, EventSubmission } from 'librechat-data-provider';
|
||||||
import type { EventHandlerParams } from './useEventHandlers';
|
import type { EventHandlerParams } from './useEventHandlers';
|
||||||
import { useGenTitleMutation, useGetStartupConfig, useGetUserBalance } from '~/data-provider';
|
import { useGenTitleMutation, useGetStartupConfig, useGetUserBalance } from '~/data-provider';
|
||||||
|
import { activeJobsQueryKey } from '~/data-provider/SSE/queries';
|
||||||
import { useAuthContext } from '~/hooks/AuthContext';
|
import { useAuthContext } from '~/hooks/AuthContext';
|
||||||
import useEventHandlers from './useEventHandlers';
|
import useEventHandlers from './useEventHandlers';
|
||||||
import store from '~/store';
|
import store from '~/store';
|
||||||
|
|
||||||
|
/** Response type for active jobs query */
|
||||||
|
interface ActiveJobsResponse {
|
||||||
|
activeJobIds: string[];
|
||||||
|
}
|
||||||
|
|
||||||
const clearDraft = (conversationId?: string | null) => {
|
const clearDraft = (conversationId?: string | null) => {
|
||||||
if (conversationId) {
|
if (conversationId) {
|
||||||
localStorage.removeItem(`${LocalStorageKeys.TEXT_DRAFT}${conversationId}`);
|
localStorage.removeItem(`${LocalStorageKeys.TEXT_DRAFT}${conversationId}`);
|
||||||
|
|
@ -55,9 +62,36 @@ export default function useResumableSSE(
|
||||||
runIndex = 0,
|
runIndex = 0,
|
||||||
) {
|
) {
|
||||||
const genTitle = useGenTitleMutation();
|
const genTitle = useGenTitleMutation();
|
||||||
|
const queryClient = useQueryClient();
|
||||||
const setActiveRunId = useSetRecoilState(store.activeRunFamily(runIndex));
|
const setActiveRunId = useSetRecoilState(store.activeRunFamily(runIndex));
|
||||||
|
|
||||||
const { token, isAuthenticated } = useAuthContext();
|
const { token, isAuthenticated } = useAuthContext();
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Optimistically add a job ID to the active jobs cache.
|
||||||
|
* Called when generation starts.
|
||||||
|
*/
|
||||||
|
const addActiveJob = useCallback(
|
||||||
|
(jobId: string) => {
|
||||||
|
queryClient.setQueryData<ActiveJobsResponse>(activeJobsQueryKey, (old) => ({
|
||||||
|
activeJobIds: [...new Set([...(old?.activeJobIds ?? []), jobId])],
|
||||||
|
}));
|
||||||
|
},
|
||||||
|
[queryClient],
|
||||||
|
);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Optimistically remove a job ID from the active jobs cache.
|
||||||
|
* Called when generation completes, aborts, or errors.
|
||||||
|
*/
|
||||||
|
const removeActiveJob = useCallback(
|
||||||
|
(jobId: string) => {
|
||||||
|
queryClient.setQueryData<ActiveJobsResponse>(activeJobsQueryKey, (old) => ({
|
||||||
|
activeJobIds: (old?.activeJobIds ?? []).filter((id) => id !== jobId),
|
||||||
|
}));
|
||||||
|
},
|
||||||
|
[queryClient],
|
||||||
|
);
|
||||||
const [_completed, setCompleted] = useState(new Set());
|
const [_completed, setCompleted] = useState(new Set());
|
||||||
const [streamId, setStreamId] = useState<string | null>(null);
|
const [streamId, setStreamId] = useState<string | null>(null);
|
||||||
const setAbortScroll = useSetRecoilState(store.abortScrollFamily(runIndex));
|
const setAbortScroll = useSetRecoilState(store.abortScrollFamily(runIndex));
|
||||||
|
|
@ -155,6 +189,8 @@ export default function useResumableSSE(
|
||||||
}
|
}
|
||||||
// Clear handler maps on stream completion to prevent memory leaks
|
// Clear handler maps on stream completion to prevent memory leaks
|
||||||
clearStepMaps();
|
clearStepMaps();
|
||||||
|
// Optimistically remove from active jobs
|
||||||
|
removeActiveJob(currentStreamId);
|
||||||
(startupConfig?.balance?.enabled ?? false) && balanceQuery.refetch();
|
(startupConfig?.balance?.enabled ?? false) && balanceQuery.refetch();
|
||||||
sse.close();
|
sse.close();
|
||||||
setStreamId(null);
|
setStreamId(null);
|
||||||
|
|
@ -303,15 +339,31 @@ export default function useResumableSSE(
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Error event - fired on actual network failures (non-200, connection lost, etc.)
|
* Error event - fired on actual network failures (non-200, connection lost, etc.)
|
||||||
* This should trigger reconnection with exponential backoff.
|
* This should trigger reconnection with exponential backoff, except for 404 errors.
|
||||||
*/
|
*/
|
||||||
sse.addEventListener('error', async (e: MessageEvent) => {
|
sse.addEventListener('error', async (e: MessageEvent) => {
|
||||||
console.log('[ResumableSSE] Stream error (network failure) - will attempt reconnect');
|
|
||||||
(startupConfig?.balance?.enabled ?? false) && balanceQuery.refetch();
|
(startupConfig?.balance?.enabled ?? false) && balanceQuery.refetch();
|
||||||
|
|
||||||
|
/* @ts-ignore - sse.js types don't expose responseCode */
|
||||||
|
const responseCode = e.responseCode;
|
||||||
|
|
||||||
|
// 404 means job doesn't exist (completed/deleted) - don't retry
|
||||||
|
if (responseCode === 404) {
|
||||||
|
console.log('[ResumableSSE] Stream not found (404) - job completed or expired');
|
||||||
|
sse.close();
|
||||||
|
// Optimistically remove from active jobs since job is gone
|
||||||
|
removeActiveJob(currentStreamId);
|
||||||
|
setIsSubmitting(false);
|
||||||
|
setShowStopButton(false);
|
||||||
|
setStreamId(null);
|
||||||
|
reconnectAttemptRef.current = 0;
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
console.log('[ResumableSSE] Stream error (network failure) - will attempt reconnect');
|
||||||
|
|
||||||
// Check for 401 and try to refresh token (same pattern as useSSE)
|
// Check for 401 and try to refresh token (same pattern as useSSE)
|
||||||
/* @ts-ignore */
|
if (responseCode === 401) {
|
||||||
if (e.responseCode === 401) {
|
|
||||||
try {
|
try {
|
||||||
const refreshResponse = await request.refreshToken();
|
const refreshResponse = await request.refreshToken();
|
||||||
const newToken = refreshResponse?.token ?? '';
|
const newToken = refreshResponse?.token ?? '';
|
||||||
|
|
@ -330,9 +382,8 @@ export default function useResumableSSE(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
sse.close();
|
|
||||||
|
|
||||||
if (reconnectAttemptRef.current < MAX_RETRIES) {
|
if (reconnectAttemptRef.current < MAX_RETRIES) {
|
||||||
|
// Increment counter BEFORE close() so abort handler knows we're reconnecting
|
||||||
reconnectAttemptRef.current++;
|
reconnectAttemptRef.current++;
|
||||||
const delay = Math.min(1000 * Math.pow(2, reconnectAttemptRef.current - 1), 30000);
|
const delay = Math.min(1000 * Math.pow(2, reconnectAttemptRef.current - 1), 30000);
|
||||||
|
|
||||||
|
|
@ -340,6 +391,8 @@ export default function useResumableSSE(
|
||||||
`[ResumableSSE] Reconnecting in ${delay}ms (attempt ${reconnectAttemptRef.current}/${MAX_RETRIES})`,
|
`[ResumableSSE] Reconnecting in ${delay}ms (attempt ${reconnectAttemptRef.current}/${MAX_RETRIES})`,
|
||||||
);
|
);
|
||||||
|
|
||||||
|
sse.close();
|
||||||
|
|
||||||
reconnectTimeoutRef.current = setTimeout(() => {
|
reconnectTimeoutRef.current = setTimeout(() => {
|
||||||
if (submissionRef.current) {
|
if (submissionRef.current) {
|
||||||
// Reconnect with isResume=true to get sync event with any missed content
|
// Reconnect with isResume=true to get sync event with any missed content
|
||||||
|
|
@ -353,7 +406,10 @@ export default function useResumableSSE(
|
||||||
setShowStopButton(true);
|
setShowStopButton(true);
|
||||||
} else {
|
} else {
|
||||||
console.error('[ResumableSSE] Max reconnect attempts reached');
|
console.error('[ResumableSSE] Max reconnect attempts reached');
|
||||||
|
sse.close();
|
||||||
errorHandler({ data: undefined, submission: currentSubmission as EventSubmission });
|
errorHandler({ data: undefined, submission: currentSubmission as EventSubmission });
|
||||||
|
// Optimistically remove from active jobs on max retries
|
||||||
|
removeActiveJob(currentStreamId);
|
||||||
setIsSubmitting(false);
|
setIsSubmitting(false);
|
||||||
setShowStopButton(false);
|
setShowStopButton(false);
|
||||||
setStreamId(null);
|
setStreamId(null);
|
||||||
|
|
@ -362,17 +418,23 @@ export default function useResumableSSE(
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Abort event - fired when sse.close() is called (intentional close).
|
* Abort event - fired when sse.close() is called (intentional close).
|
||||||
* This happens on cleanup/navigation. Do NOT reconnect, just reset UI.
|
* This happens on cleanup/navigation OR when error handler closes to reconnect.
|
||||||
* The backend stream continues running - useResumeOnLoad will restore if user returns.
|
* Only reset state if we're NOT in a reconnection cycle.
|
||||||
*/
|
*/
|
||||||
sse.addEventListener('abort', () => {
|
sse.addEventListener('abort', () => {
|
||||||
|
// If we're in a reconnection cycle, don't reset state
|
||||||
|
// (error handler will set up the reconnect timeout)
|
||||||
|
if (reconnectAttemptRef.current > 0) {
|
||||||
|
console.log('[ResumableSSE] Stream closed for reconnect - preserving state');
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
console.log('[ResumableSSE] Stream aborted (intentional close) - no reconnect');
|
console.log('[ResumableSSE] Stream aborted (intentional close) - no reconnect');
|
||||||
// Clear any pending reconnect attempts
|
// Clear any pending reconnect attempts
|
||||||
if (reconnectTimeoutRef.current) {
|
if (reconnectTimeoutRef.current) {
|
||||||
clearTimeout(reconnectTimeoutRef.current);
|
clearTimeout(reconnectTimeoutRef.current);
|
||||||
reconnectTimeoutRef.current = null;
|
reconnectTimeoutRef.current = null;
|
||||||
}
|
}
|
||||||
reconnectAttemptRef.current = 0;
|
|
||||||
// Reset UI state - useResumeOnLoad will restore if user returns to this conversation
|
// Reset UI state - useResumeOnLoad will restore if user returns to this conversation
|
||||||
setIsSubmitting(false);
|
setIsSubmitting(false);
|
||||||
setShowStopButton(false);
|
setShowStopButton(false);
|
||||||
|
|
@ -425,6 +487,7 @@ export default function useResumableSSE(
|
||||||
setMessages,
|
setMessages,
|
||||||
startupConfig?.balance?.enabled,
|
startupConfig?.balance?.enabled,
|
||||||
balanceQuery,
|
balanceQuery,
|
||||||
|
removeActiveJob,
|
||||||
],
|
],
|
||||||
);
|
);
|
||||||
|
|
||||||
|
|
@ -522,6 +585,8 @@ export default function useResumableSSE(
|
||||||
// Resume: just subscribe to existing stream, don't start new generation
|
// Resume: just subscribe to existing stream, don't start new generation
|
||||||
console.log('[ResumableSSE] Resuming existing stream:', resumeStreamId);
|
console.log('[ResumableSSE] Resuming existing stream:', resumeStreamId);
|
||||||
setStreamId(resumeStreamId);
|
setStreamId(resumeStreamId);
|
||||||
|
// Optimistically add to active jobs (in case it's not already there)
|
||||||
|
addActiveJob(resumeStreamId);
|
||||||
subscribeToStream(resumeStreamId, submission, true); // isResume=true
|
subscribeToStream(resumeStreamId, submission, true); // isResume=true
|
||||||
} else {
|
} else {
|
||||||
// New generation: start and then subscribe
|
// New generation: start and then subscribe
|
||||||
|
|
@ -529,6 +594,8 @@ export default function useResumableSSE(
|
||||||
const newStreamId = await startGeneration(submission);
|
const newStreamId = await startGeneration(submission);
|
||||||
if (newStreamId) {
|
if (newStreamId) {
|
||||||
setStreamId(newStreamId);
|
setStreamId(newStreamId);
|
||||||
|
// Optimistically add to active jobs
|
||||||
|
addActiveJob(newStreamId);
|
||||||
subscribeToStream(newStreamId, submission);
|
subscribeToStream(newStreamId, submission);
|
||||||
} else {
|
} else {
|
||||||
console.error('[ResumableSSE] Failed to get streamId from startGeneration');
|
console.error('[ResumableSSE] Failed to get streamId from startGeneration');
|
||||||
|
|
@ -547,6 +614,8 @@ export default function useResumableSSE(
|
||||||
clearTimeout(reconnectTimeoutRef.current);
|
clearTimeout(reconnectTimeoutRef.current);
|
||||||
reconnectTimeoutRef.current = null;
|
reconnectTimeoutRef.current = null;
|
||||||
}
|
}
|
||||||
|
// Reset reconnect counter before closing (so abort handler doesn't think we're reconnecting)
|
||||||
|
reconnectAttemptRef.current = 0;
|
||||||
if (sseRef.current) {
|
if (sseRef.current) {
|
||||||
sseRef.current.close();
|
sseRef.current.close();
|
||||||
sseRef.current = null;
|
sseRef.current = null;
|
||||||
|
|
|
||||||
|
|
@ -903,6 +903,18 @@ class GenerationJobManagerClass {
|
||||||
return { running, complete, error, aborted };
|
return { running, complete, error, aborted };
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Get active job IDs for a user.
|
||||||
|
* Returns conversation IDs of running jobs belonging to the user.
|
||||||
|
* Performs self-healing cleanup of stale entries.
|
||||||
|
*
|
||||||
|
* @param userId - The user ID to query
|
||||||
|
* @returns Array of conversation IDs with active jobs
|
||||||
|
*/
|
||||||
|
async getActiveJobIdsForUser(userId: string): Promise<string[]> {
|
||||||
|
return this.jobStore.getActiveJobIdsByUser(userId);
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Destroy the manager.
|
* Destroy the manager.
|
||||||
* Cleans up all resources including runtime state, buffers, and stores.
|
* Cleans up all resources including runtime state, buffers, and stores.
|
||||||
|
|
|
||||||
|
|
@ -26,6 +26,9 @@ export class InMemoryJobStore implements IJobStore {
|
||||||
private contentState = new Map<string, ContentState>();
|
private contentState = new Map<string, ContentState>();
|
||||||
private cleanupInterval: NodeJS.Timeout | null = null;
|
private cleanupInterval: NodeJS.Timeout | null = null;
|
||||||
|
|
||||||
|
/** Maps userId -> Set of streamIds (conversationIds) for active jobs */
|
||||||
|
private userJobMap = new Map<string, Set<string>>();
|
||||||
|
|
||||||
/** Time to keep completed jobs before cleanup (0 = immediate) */
|
/** Time to keep completed jobs before cleanup (0 = immediate) */
|
||||||
private ttlAfterComplete = 0;
|
private ttlAfterComplete = 0;
|
||||||
|
|
||||||
|
|
@ -76,6 +79,15 @@ export class InMemoryJobStore implements IJobStore {
|
||||||
};
|
};
|
||||||
|
|
||||||
this.jobs.set(streamId, job);
|
this.jobs.set(streamId, job);
|
||||||
|
|
||||||
|
// Track job by userId for efficient user-scoped queries
|
||||||
|
let userJobs = this.userJobMap.get(userId);
|
||||||
|
if (!userJobs) {
|
||||||
|
userJobs = new Set();
|
||||||
|
this.userJobMap.set(userId, userJobs);
|
||||||
|
}
|
||||||
|
userJobs.add(streamId);
|
||||||
|
|
||||||
logger.debug(`[InMemoryJobStore] Created job: ${streamId}`);
|
logger.debug(`[InMemoryJobStore] Created job: ${streamId}`);
|
||||||
|
|
||||||
return job;
|
return job;
|
||||||
|
|
@ -94,6 +106,18 @@ export class InMemoryJobStore implements IJobStore {
|
||||||
}
|
}
|
||||||
|
|
||||||
async deleteJob(streamId: string): Promise<void> {
|
async deleteJob(streamId: string): Promise<void> {
|
||||||
|
// Remove from user's job set before deleting
|
||||||
|
const job = this.jobs.get(streamId);
|
||||||
|
if (job) {
|
||||||
|
const userJobs = this.userJobMap.get(job.userId);
|
||||||
|
if (userJobs) {
|
||||||
|
userJobs.delete(streamId);
|
||||||
|
if (userJobs.size === 0) {
|
||||||
|
this.userJobMap.delete(job.userId);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
this.jobs.delete(streamId);
|
this.jobs.delete(streamId);
|
||||||
this.contentState.delete(streamId);
|
this.contentState.delete(streamId);
|
||||||
logger.debug(`[InMemoryJobStore] Deleted job: ${streamId}`);
|
logger.debug(`[InMemoryJobStore] Deleted job: ${streamId}`);
|
||||||
|
|
@ -178,9 +202,42 @@ export class InMemoryJobStore implements IJobStore {
|
||||||
}
|
}
|
||||||
this.jobs.clear();
|
this.jobs.clear();
|
||||||
this.contentState.clear();
|
this.contentState.clear();
|
||||||
|
this.userJobMap.clear();
|
||||||
logger.debug('[InMemoryJobStore] Destroyed');
|
logger.debug('[InMemoryJobStore] Destroyed');
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Get active job IDs for a user.
|
||||||
|
* Returns conversation IDs of running jobs belonging to the user.
|
||||||
|
* Also performs self-healing cleanup: removes stale entries for jobs that no longer exist.
|
||||||
|
*/
|
||||||
|
async getActiveJobIdsByUser(userId: string): Promise<string[]> {
|
||||||
|
const trackedIds = this.userJobMap.get(userId);
|
||||||
|
if (!trackedIds || trackedIds.size === 0) {
|
||||||
|
return [];
|
||||||
|
}
|
||||||
|
|
||||||
|
const activeIds: string[] = [];
|
||||||
|
|
||||||
|
for (const streamId of trackedIds) {
|
||||||
|
const job = this.jobs.get(streamId);
|
||||||
|
// Only include if job exists AND is still running
|
||||||
|
if (job && job.status === 'running') {
|
||||||
|
activeIds.push(streamId);
|
||||||
|
} else {
|
||||||
|
// Self-healing: job completed/deleted but mapping wasn't cleaned - fix it now
|
||||||
|
trackedIds.delete(streamId);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Clean up empty set
|
||||||
|
if (trackedIds.size === 0) {
|
||||||
|
this.userJobMap.delete(userId);
|
||||||
|
}
|
||||||
|
|
||||||
|
return activeIds;
|
||||||
|
}
|
||||||
|
|
||||||
// ===== Content State Methods =====
|
// ===== Content State Methods =====
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
||||||
|
|
@ -122,6 +122,16 @@ export interface IJobStore {
|
||||||
/** Destroy the store and release resources */
|
/** Destroy the store and release resources */
|
||||||
destroy(): Promise<void>;
|
destroy(): Promise<void>;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Get active job IDs for a user.
|
||||||
|
* Returns conversation IDs of running jobs belonging to the user.
|
||||||
|
* Also performs self-healing cleanup of stale entries.
|
||||||
|
*
|
||||||
|
* @param userId - The user ID to query
|
||||||
|
* @returns Array of conversation IDs with active jobs
|
||||||
|
*/
|
||||||
|
getActiveJobIdsByUser(userId: string): Promise<string[]>;
|
||||||
|
|
||||||
// ===== Content State Methods =====
|
// ===== Content State Methods =====
|
||||||
// These methods manage volatile content state tied to each job.
|
// These methods manage volatile content state tied to each job.
|
||||||
// In-memory: Uses WeakRef to graph for live access
|
// In-memory: Uses WeakRef to graph for live access
|
||||||
|
|
|
||||||
|
|
@ -226,6 +226,8 @@ export const agents = ({ path = '', options }: { path?: string; options?: object
|
||||||
return url;
|
return url;
|
||||||
};
|
};
|
||||||
|
|
||||||
|
export const activeJobs = () => `${BASE_URL}/api/agents/chat/active`;
|
||||||
|
|
||||||
export const mcp = {
|
export const mcp = {
|
||||||
tools: `${BASE_URL}/api/mcp/tools`,
|
tools: `${BASE_URL}/api/mcp/tools`,
|
||||||
servers: `${BASE_URL}/api/mcp/servers`,
|
servers: `${BASE_URL}/api/mcp/servers`,
|
||||||
|
|
|
||||||
|
|
@ -1037,3 +1037,12 @@ export function getGraphApiToken(params: q.GraphTokenParams): Promise<q.GraphTok
|
||||||
export function getDomainServerBaseUrl(): string {
|
export function getDomainServerBaseUrl(): string {
|
||||||
return `${endpoints.apiBaseUrl()}/api`;
|
return `${endpoints.apiBaseUrl()}/api`;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/* Active Jobs */
|
||||||
|
export interface ActiveJobsResponse {
|
||||||
|
activeJobIds: string[];
|
||||||
|
}
|
||||||
|
|
||||||
|
export const getActiveJobs = (): Promise<ActiveJobsResponse> => {
|
||||||
|
return request.get(endpoints.activeJobs());
|
||||||
|
};
|
||||||
|
|
|
||||||
|
|
@ -60,6 +60,8 @@ export enum QueryKeys {
|
||||||
/* MCP Servers */
|
/* MCP Servers */
|
||||||
mcpServers = 'mcpServers',
|
mcpServers = 'mcpServers',
|
||||||
mcpServer = 'mcpServer',
|
mcpServer = 'mcpServer',
|
||||||
|
/* Active Jobs */
|
||||||
|
activeJobs = 'activeJobs',
|
||||||
}
|
}
|
||||||
|
|
||||||
// Dynamic query keys that require parameters
|
// Dynamic query keys that require parameters
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue