diff --git a/client/src/hooks/SSE/__tests__/useResumableSSE.spec.ts b/client/src/hooks/SSE/__tests__/useResumableSSE.spec.ts index cc198a27a0..97a397cab1 100644 --- a/client/src/hooks/SSE/__tests__/useResumableSSE.spec.ts +++ b/client/src/hooks/SSE/__tests__/useResumableSSE.spec.ts @@ -8,6 +8,7 @@ import { request, } from 'librechat-data-provider'; import type { TMessage, TSubmission } from 'librechat-data-provider'; +import type { PendingSteer } from '~/store/families'; type SSEEventListener = (e: Partial & { responseCode?: number }) => void; @@ -258,7 +259,7 @@ jest.mock('librechat-data-provider', () => { }; }); -import useResumableSSE from '~/hooks/SSE/useResumableSSE'; +import useResumableSSE, { selectLocalSteersForQueue } from '~/hooks/SSE/useResumableSSE'; const CONV_ID = 'conv-abc-123'; @@ -3853,3 +3854,67 @@ describe('useResumableSSE', () => { unmount(); }); }); + +/** + * `convertLocalSteersToQueued` (the `final` handler and the intentional-close + * `abort` listener both call it, alongside the two failure-terminal paths + * covered above) is a thin `useRecoilCallback` wrapper: read the conversation's + * chips, run them through this selection, hand the result to `useSteerConvert`. + * This file's `useRecoilCallback: () => jest.fn()` mock (see above — "the + * hook's steer-chip/queue callbacks need a RecoilRoot; these tests render + * bare") makes that wrapper, and every other recoil-callback in this hook + * (`resolveSteerChip`, `seedSteerChips` included — neither has a direct test + * in this file either), inert: calling it from the `final`/`abort` source + * lines is invisible to `mockConvertSteersToQueued` assertions here, since the + * mock discards the real closure before it can ever call through. Covering + * the selection logic directly (this is the exact piece of logic finding 3 + * was about — which statuses survive a run end) rather than faking a + * RecoilRoot-backed integration around the rest of this large hook's mocks. + */ +describe('selectLocalSteersForQueue', () => { + const chip = (over: Partial = {}): PendingSteer => ({ + steerId: 's1', + text: 'default text', + status: 'pending', + createdAt: 1, + ...over, + }); + + it('includes pending and failed chips, excluding sending', () => { + const chips = [ + chip({ steerId: 'p1', status: 'pending' }), + chip({ steerId: 'f1', status: 'failed' }), + chip({ steerId: 'sending-1', status: 'sending' }), + ]; + expect(selectLocalSteersForQueue(chips).map((steer) => steer.steerId)).toEqual(['p1', 'f1']); + }); + + it('converts a failed local chip present at a run-end path into a queueable item', () => { + // The leak finding 3 was about: a `failed` chip carries a local-* id the + // server never reports, so `data.pendingSteers`/the abort response can + // never carry it — this selection is the ONLY place left that can catch + // it before the `final`/`abort` paths hand off to `useSteerConvert`. + const failed = chip({ steerId: 'local-failed', status: 'failed', text: 'redo this' }); + expect(selectLocalSteersForQueue([failed])).toEqual([ + expect.objectContaining({ steerId: 'local-failed', text: 'redo this', createdAt: 1 }), + ]); + }); + + it('does not sweep a sending chip: its own POST callback owns it', () => { + // Converting it here too would race that callback — a late ACK's re-add + // in `resolveAcknowledgedSteer` could then double-queue the same words. + const sending = chip({ steerId: 'in-flight', status: 'sending' }); + expect(selectLocalSteersForQueue([sending])).toEqual([]); + }); + + it('carries files only when present', () => { + const withFiles = chip({ + steerId: 'p2', + files: [{ file_id: 'f1', filename: 'a.png' }], + }); + const withoutFiles = chip({ steerId: 'p3' }); + const [withResult, withoutResult] = selectLocalSteersForQueue([withFiles, withoutFiles]); + expect(withResult.files).toEqual(withFiles.files); + expect(withoutResult.files).toBeUndefined(); + }); +}); diff --git a/client/src/hooks/SSE/useResumableSSE.ts b/client/src/hooks/SSE/useResumableSSE.ts index 1c46de44ae..712d11dd56 100644 --- a/client/src/hooks/SSE/useResumableSSE.ts +++ b/client/src/hooks/SSE/useResumableSSE.ts @@ -32,7 +32,7 @@ import type { TActivityLabelEvent, } from 'librechat-data-provider'; import type { ActiveJobsResponse, StreamStatusResponse } from '~/data-provider'; -import type { DrainAfterAbort, QueuedMessageOrigin } from '~/store/families'; +import type { DrainAfterAbort, QueuedMessageOrigin, PendingSteer } from '~/store/families'; import type { GenerationProtocolVersion } from '~/data-provider'; import type { EventHandlerParams } from './useEventHandlers'; import type { TResData } from '~/common'; @@ -607,6 +607,42 @@ const mergeResumeMessages = ( return [...nextMessages, userMessage, responseMessage]; }; +/** + * Local chips with no injection-boundary event left to resolve them: `pending` + * behind a server id that no `on_steer_applied` will ever confirm (the run + * ended), or `failed` and never having reached the server at all. Either one + * left behind survives past this run end and renders under whatever comes + * next — `PendingSteers` reads the same atom keyed only by conversation, not + * by run. + * + * `sending` chips are deliberately excluded: they have their own in-flight + * POST whose `onSuccess`/`onError` will settle them, and sweeping them here + * too would race that callback — a late ACK's re-add in + * `resolveAcknowledgedSteer` could then double-queue the same words. + */ +export function selectLocalSteersForQueue( + chips: PendingSteer[], + /** Ids the caller has already routed elsewhere this run end; matched against + * both ids a chip can be known by, since either may be the one excluded. */ + excluded: ReadonlySet = new Set(), +): TPendingSteer[] { + return chips + .filter( + (steer) => + (steer.status === 'pending' || steer.status === 'failed') && + !excluded.has(steer.steerId) && + (steer.clientSteerId == null || !excluded.has(steer.clientSteerId)), + ) + .map((steer) => ({ + steerId: steer.steerId, + ...(steer.clientSteerId && { clientSteerId: steer.clientSteerId }), + text: steer.text, + createdAt: steer.createdAt, + ...(steer.files && steer.files.length > 0 && { files: steer.files }), + ...(steer.queuedOrigin && { queuedOrigin: steer.queuedOrigin }), + })); +} + /** * Hook for resumable SSE streams. * Separates generation start (POST) from stream subscription (GET EventSource). @@ -938,14 +974,12 @@ export default function useResumableSSE( * HTTP response consumes the same data as a fallback in useChatHelpers). */ const convertSteersToQueued = useSteerConvert(); - /** Error events carry no `pendingSteers` payload (the server drops its copy - * on failure), but every acknowledged OR failed chip's text is local — - * convert both to queued follow-ups so the user's words survive a failed - * run. `sending` chips settle through their own POST callbacks (404 falls - * back to queue/send); `pending` and `failed` chips have no such callback - * waiting on them, so leaving either behind would strand it — and worse, - * `PendingSteers` reads this same atom on the NEXT run's reply, so a - * stranded chip would leak into a message it was never part of. */ + /** Sweeps this conversation's local chips (see `selectLocalSteersForQueue`) + * into queued follow-ups. Error events carry no `pendingSteers` payload of + * their own (the server drops its copy on failure) — this is the only + * source of truth for those runs; the `final` and intentional-`abort` + * paths call it alongside their own server-reported list as a backstop for + * `failed` chips, which never rode that list at all. */ const convertLocalSteersToQueued = useRecoilCallback( ({ snapshot }) => ( @@ -957,22 +991,10 @@ export default function useResumableSSE( }, ) => { const chips = snapshot.getLoadable(store.pendingSteersByConvoId(conversationId)).getValue(); - const excluded = new Set(options?.excludeSteerIds ?? []); - const settled = chips - .filter( - (steer) => - (steer.status === 'pending' || steer.status === 'failed') && - !excluded.has(steer.steerId) && - (steer.clientSteerId == null || !excluded.has(steer.clientSteerId)), - ) - .map((steer) => ({ - steerId: steer.steerId, - ...(steer.clientSteerId && { clientSteerId: steer.clientSteerId }), - text: steer.text, - createdAt: steer.createdAt, - ...(steer.files && steer.files.length > 0 && { files: steer.files }), - ...(steer.queuedOrigin && { queuedOrigin: steer.queuedOrigin }), - })); + const settled = selectLocalSteersForQueue( + chips, + new Set(options?.excludeSteerIds ?? []), + ); if (settled.length > 0) { convertSteersToQueued(conversationId, settled, { claimParked: options?.claimParked, @@ -1491,6 +1513,11 @@ export default function useResumableSSE( generationProtocolVersion, }, ); + // A `failed` chip never reached the server, so it never rides + // `data.pendingSteers` above — without this it survives a NORMAL + // completion and renders (with a live Retry) under the NEXT run's + // reply. Idempotent alongside the call above: both dedupe by id. + convertLocalSteersToQueued(finalConvoId); let finalHandled = false; try { finalHandler(data, currentSubmission as EventSubmission); @@ -3077,6 +3104,12 @@ export default function useResumableSSE( * merge into the next response in this conversation. On a resume the * collected usage is re-folded via backfillUsage, so nothing is lost. */ resetLive({ ...currentSubmission, userMessage }); + // No final/error event fires on this path, so it's the only place left + // to sweep a local `failed` chip — otherwise it survives this close and + // renders (with a live Retry) under whatever run starts next. + convertLocalSteersToQueued( + currentSubmission.conversation?.conversationId ?? currentStreamId, + ); }); // Start the SSE connection