diff --git a/client/src/hooks/Chat/__tests__/useQueueDrain.spec.tsx b/client/src/hooks/Chat/__tests__/useQueueDrain.spec.tsx index d638f89f99..9757121ed0 100644 --- a/client/src/hooks/Chat/__tests__/useQueueDrain.spec.tsx +++ b/client/src/hooks/Chat/__tests__/useQueueDrain.spec.tsx @@ -175,6 +175,94 @@ describe('useQueueDrain', () => { await waitFor(() => expect(mockMarkFilesUsage).not.toHaveBeenCalled()); }); + /** The server caps ids per request, so a queue holding more than one batch + * must renew in several; truncating would leave later messages on their + * enqueue-time hold. */ + it('renews every queued attachment across multiple capped batches', async () => { + const manyFiles = (prefix: string, n: number) => + Array.from({ length: n }, (_, i) => ({ file_id: `${prefix}-${i}`, type: 'image/png' })); + const { setters } = setup(({ set }) => { + set(store.queuedMessagesByConvoId(CONVO_ID), [ + { ...queuedMessage('q1', 'first'), files: manyFiles('sent', 2) }, + { ...queuedMessage('q2', 'second'), files: manyFiles('a', 10) }, + { ...queuedMessage('q3', 'third'), files: manyFiles('b', 4) }, + ]); + }); + + act(() => { + setters.setRunEnd!(runEnd()); + }); + + await waitFor(() => expect(mockMarkFilesUsage).toHaveBeenCalledTimes(2)); + const sent = mockMarkFilesUsage.mock.calls.flatMap((call) => call[0].file_ids); + expect(sent).toHaveLength(14); + expect(sent).toContain('b-3'); + expect(sent).not.toContain('sent-0'); + for (const call of mockMarkFilesUsage.mock.calls) { + expect(call[0].file_ids.length).toBeLessThanOrEqual(10); + } + }); + + /** A refused send puts the item back with its run-end signal already + * consumed, so nothing else would touch it before the next drain. */ + it('renews a restored item when the send is refused', async () => { + const { ask, setters } = setup(({ set }) => { + set(store.queuedMessagesByConvoId(CONVO_ID), [ + { ...queuedMessage('q1', 'refused'), files: [{ file_id: 'restored', type: 'image/png' }] }, + ]); + }); + ask.mockReturnValue(false); + + act(() => { + setters.setRunEnd!(runEnd()); + }); + + await waitFor(() => expect(mockMarkFilesUsage).toHaveBeenCalledTimes(1)); + expect(mockMarkFilesUsage).toHaveBeenCalledWith({ file_ids: ['restored'] }); + }); + + /** A single run can pause for approval more than once, so renewal cannot + * depend on catching drain transitions alone. */ + it('renews on a heartbeat while items stay queued', async () => { + jest.useFakeTimers(); + try { + setup(({ set }) => { + set(store.queuedMessagesByConvoId(CONVO_ID), [ + { ...queuedMessage('q1', 'waiting'), files: [{ file_id: 'held', type: 'image/png' }] }, + ]); + }); + + expect(mockMarkFilesUsage).not.toHaveBeenCalled(); + act(() => { + jest.advanceTimersByTime(30 * 60 * 1000); + }); + expect(mockMarkFilesUsage).toHaveBeenCalledWith({ file_ids: ['held'] }); + + act(() => { + jest.advanceTimersByTime(30 * 60 * 1000); + }); + expect(mockMarkFilesUsage).toHaveBeenCalledTimes(2); + } finally { + jest.useRealTimers(); + } + }); + + it('emits no heartbeat when the queue holds no attachments', async () => { + jest.useFakeTimers(); + try { + setup(({ set }) => { + set(store.queuedMessagesByConvoId(CONVO_ID), [queuedMessage('q1', 'no files')]); + }); + + act(() => { + jest.advanceTimersByTime(2 * 60 * 60 * 1000); + }); + expect(mockMarkFilesUsage).not.toHaveBeenCalled(); + } finally { + jest.useRealTimers(); + } + }); + it('passes carried quotes + manual skills through as overrides', async () => { const { ask, setters } = setup(({ set }) => { set(store.queuedMessagesByConvoId(CONVO_ID), [ diff --git a/client/src/hooks/Chat/useQueueDrain.ts b/client/src/hooks/Chat/useQueueDrain.ts index 8498d5e7da..e54a5bfe04 100644 --- a/client/src/hooks/Chat/useQueueDrain.ts +++ b/client/src/hooks/Chat/useQueueDrain.ts @@ -9,6 +9,10 @@ import store from '~/store'; /** Mirrors the server's per-request cap on a usage touch. */ const QUEUE_USAGE_MAX_FILES = 10; +/** Well under the server's smallest hold (24h), so no gap between renewals + * can outlive one, however long a run pauses. */ +const QUEUE_USAGE_RENEW_INTERVAL_MS = 30 * 60 * 1000; + const collectQueuedFileIds = (items: QueuedMessage[]): string[] => { const fileIds: string[] = []; for (const item of items) { @@ -16,14 +20,22 @@ const collectQueuedFileIds = (items: QueuedMessage[]): string[] => { if (typeof file.file_id === 'string' && file.file_id.length > 0) { fileIds.push(file.file_id); } - if (fileIds.length === QUEUE_USAGE_MAX_FILES) { - return fileIds; - } } } return fileIds; }; +/** The server caps ids per request, so a queue holding more than one batch + * has to renew in several. Truncating instead would leave everything after + * the first batch on its enqueue-time hold. */ +const batchFileIds = (fileIds: string[]): string[][] => { + const batches: string[][] = []; + for (let i = 0; i < fileIds.length; i += QUEUE_USAGE_MAX_FILES) { + batches.push(fileIds.slice(i, i + QUEUE_USAGE_MAX_FILES)); + } + return batches; +}; + /** * Auto-sends queued follow-up messages when a run finishes. * @@ -50,6 +62,35 @@ export default function useQueueDrain( ); const isSubmitting = useRecoilValue(store.isSubmittingFamily(index)); const { mutate: markFilesUsage } = useMarkFilesUsageMutation(); + const ownQueue = useRecoilValue( + store.queuedMessagesByConvoId(activeConversationId ?? Constants.NEW_CONVO), + ); + + /** + * Heartbeat renewal while anything is queued. + * + * Renewing only at drain transitions ties an attachment's survival to + * catching every state change, and a single run can stretch well past one + * hold: it may interrupt for approval more than once, and each pause can + * run to the configured window. Rather than hook every transition, renew on + * a cadence far shorter than the hold itself, so no single gap can outlive + * it. Bounded regardless: the server clamps each renewal against the file's + * upload time. A queue nobody has open stops emitting these and lapses + * normally. + */ + useEffect(() => { + const fileIds = collectQueuedFileIds(ownQueue); + if (fileIds.length === 0) { + return; + } + const renew = () => { + for (const file_ids of batchFileIds(fileIds)) { + markFilesUsage({ file_ids }); + } + }; + const timer = setInterval(renew, QUEUE_USAGE_RENEW_INTERVAL_MS); + return () => clearInterval(timer); + }, [ownQueue, markFilesUsage]); // Fully synchronous reads (getLoadable): a useRecoilCallback snapshot is // only guaranteed valid for the callback's synchronous execution, so no @@ -165,15 +206,6 @@ export default function useQueueDrain( return; } const { next, conversationId, remainderFileIds } = drained; - /** Renew the TTL hold on what stays queued. Items behind this one wait - * another full run (which may itself pause for approval), so a hold - * taken once at enqueue would lapse before a deep queue drains. The - * server caps each renewal against the upload time, so this cannot - * extend a file indefinitely. Fire-and-forget: send-time marking is - * the backstop. */ - if (remainderFileIds.length > 0) { - markFilesUsage({ file_ids: remainderFileIds }); - } // The queued item is the FULL submission context: explicit (possibly // empty) overrides stop `ask` from vacuuming up files, quotes, or skill // picks the user has staged in the composer for their NEXT message. @@ -192,6 +224,21 @@ export default function useQueueDrain( // available for manual send. restoreQueued(conversationId, next); } + /** Renew the TTL hold on everything still queued. Items behind this one + * wait another full run (which may itself pause for approval), so a hold + * taken once at enqueue would lapse before a deep queue drains. Runs + * after `ask` so a refused send, whose item goes back on the queue with + * its run-end signal already consumed, is renewed too rather than left + * on its original hold. The server clamps every renewal against the + * upload time, so this cannot extend a file indefinitely. + * Fire-and-forget: send-time marking is the backstop. */ + const toRenew = + accepted === false + ? [...collectQueuedFileIds([next]), ...remainderFileIds] + : remainderFileIds; + for (const file_ids of batchFileIds(toRenew)) { + markFilesUsage({ file_ids }); + } }, [ runEnd, parkedRunEnd,