diff --git a/client/src/hooks/SSE/__tests__/transport.spec.ts b/client/src/hooks/SSE/__tests__/transport.spec.ts new file mode 100644 index 0000000000..28ea0cffea --- /dev/null +++ b/client/src/hooks/SSE/__tests__/transport.spec.ts @@ -0,0 +1,83 @@ +import { SSE } from 'sse.js'; + +/** + * `useResumableSSE` decides that a hidden tab lost its stream by reading the + * transport's own `readyState`. That only holds because sse.js marks the + * connection closed on the ordinary XHR `load` path — a response body that + * simply ends dispatches no error and no abort, so `readyState` is the single + * observable left. The assumption belongs to the library rather than to our + * hook, so it is pinned here against the real module instead of a mock. + */ +type XHRListener = (event: { currentTarget: FakeXHR }) => void; + +class FakeXHR { + status = 200; + responseText = ''; + withCredentials = false; + aborted = false; + private readonly listeners: Record = {}; + + addEventListener(type: string, listener: XHRListener) { + (this.listeners[type] ??= []).push(listener); + } + + open() {} + setRequestHeader() {} + send() {} + + abort() { + this.aborted = true; + this.emit('abort'); + } + + emit(type: string) { + for (const listener of this.listeners[type] ?? []) { + listener({ currentTarget: this }); + } + } +} + +describe('sse.js transport contract', () => { + const OriginalXHR = global.XMLHttpRequest; + let xhr: FakeXHR; + + beforeEach(() => { + xhr = new FakeXHR(); + global.XMLHttpRequest = jest.fn(() => xhr) as unknown as typeof XMLHttpRequest; + }); + + afterEach(() => { + global.XMLHttpRequest = OriginalXHR; + }); + + it('marks the connection closed when the response body ends without a terminal event', () => { + const sse = new SSE('/api/agents/chat/stream/convo-1', { method: 'GET' }); + const onError = jest.fn(); + const onAbort = jest.fn(); + sse.addEventListener('error', onError); + sse.addEventListener('abort', onAbort); + + xhr.responseText = 'event: message\ndata: {"created":true}\n\n'; + xhr.emit('progress'); + expect(sse.readyState).not.toBe(SSE.CLOSED); + + /** The intermediary ended the body under a frozen tab: XHR reports an + * ordinary load, and sse.js dispatches nothing for it. */ + xhr.emit('load'); + + expect(onError).not.toHaveBeenCalled(); + expect(onAbort).not.toHaveBeenCalled(); + expect(sse.readyState).toBe(SSE.CLOSED); + }); + + it('dispatches abort when the user agent cancels an in-flight request', () => { + const sse = new SSE('/api/agents/chat/stream/convo-1', { method: 'GET' }); + const onAbort = jest.fn(); + sse.addEventListener('abort', onAbort); + + xhr.emit('abort'); + + expect(onAbort).toHaveBeenCalledTimes(1); + expect(sse.readyState).toBe(SSE.CLOSED); + }); +}); diff --git a/client/src/hooks/SSE/__tests__/useResumableSSE.spec.ts b/client/src/hooks/SSE/__tests__/useResumableSSE.spec.ts index c7643a1382..4f09207441 100644 --- a/client/src/hooks/SSE/__tests__/useResumableSSE.spec.ts +++ b/client/src/hooks/SSE/__tests__/useResumableSSE.spec.ts @@ -17,14 +17,17 @@ interface MockSSEInstance { stream: jest.Mock; close: jest.Mock; headers: Record; + readyState: number; _listeners: Record; _emit: (event: string, data?: Partial & { responseCode?: number }) => void; } const mockSSEInstances: MockSSEInstance[] = []; +const MOCK_SSE_OPEN = 1; +const MOCK_SSE_CLOSED = 2; -jest.mock('sse.js', () => ({ - SSE: jest +jest.mock('sse.js', () => { + const SSE = jest .fn() .mockImplementation((url: string, options?: { headers?: Record }) => { const listeners: Record = {}; @@ -34,15 +37,24 @@ jest.mock('sse.js', () => ({ listeners[event] = cb; }), stream: jest.fn(), - close: jest.fn(), + close: jest.fn(() => { + if (instance.readyState === 2) { + return; + } + instance.readyState = 2; + instance._emit('abort'); + }), headers: { ...options?.headers }, + readyState: 1, _listeners: listeners, _emit: (event, data = {}) => listeners[event]?.(data as MessageEvent), }; mockSSEInstances.push(instance); return instance; - }), -})); + }) as jest.Mock & { CLOSED: number }; + SSE.CLOSED = 2; + return { SSE }; +}); const mockSetQueryData = jest.fn(); const mockGetQueryData = jest.fn(); @@ -3755,6 +3767,233 @@ describe('useResumableSSE', () => { unmount(); }); + it('reconnects when the user agent aborts a live stream instead of going idle', async () => { + jest.useFakeTimers(); + (request.post as jest.Mock).mockResolvedValue({ + streamId: 'stream-epoch', + status: 'started', + generationCreatedAt: 1000, + }); + const submission = buildSubmission(); + const chatHelpers = buildChatHelpers(); + + const { unmount } = renderHook(() => useResumableSSE(submission, chatHelpers)); + await flushMicrotasks(); + + const initialSSE = getLastSSE(); + const sseCount = mockSSEInstances.length; + mockSetIsSubmitting.mockClear(); + + /** Backgrounding a mobile browser cancels the in-flight XHR, which surfaces + * as an abort with no terminal event behind it. */ + await act(async () => { + initialSSE._emit('abort'); + }); + + expect(mockSetIsSubmitting).not.toHaveBeenCalledWith(false); + expect(mockSetIsSubmitting).toHaveBeenCalledWith(true); + + await advanceRetryTimer(1000); + + expect(mockSSEInstances).toHaveLength(sseCount + 1); + expect(getLastSSE()._url).toBe( + '/api/agents/chat/stream/stream-epoch?resume=true&generationCreatedAt=1000&generationProtocolVersion=2', + ); + unmount(); + }); + + it('keeps climbing the ladder when the user agent aborts the replacement before it opens', async () => { + jest.useFakeTimers(); + (request.post as jest.Mock).mockResolvedValue({ + streamId: 'stream-epoch', + status: 'started', + generationCreatedAt: 1000, + }); + const submission = buildSubmission(); + const chatHelpers = buildChatHelpers(); + + const { unmount } = renderHook(() => useResumableSSE(submission, chatHelpers)); + await flushMicrotasks(); + + const sseCount = mockSSEInstances.length; + + await act(async () => { + getLastSSE()._emit('abort'); + }); + await advanceRetryTimer(1000); + expect(mockSSEInstances).toHaveLength(sseCount + 1); + + /** The retry fired while the tab was still backgrounded, so the user agent + * cancels the replacement too — before it ever emits `open`, which is what + * would have cleared the shared reconnect counter. Recovery has to read + * this as the replacement's own failure, not the previous connection's + * deliberate close. */ + await act(async () => { + getLastSSE()._emit('abort'); + }); + await advanceRetryTimer(2000); + + expect(mockSSEInstances).toHaveLength(sseCount + 2); + expect(getLastSSE()._url).toBe( + '/api/agents/chat/stream/stream-epoch?resume=true&generationCreatedAt=1000&generationProtocolVersion=2', + ); + unmount(); + }); + + it('does not reconnect on the abort that follows a FINAL event', async () => { + jest.useFakeTimers(); + const submission = buildSubmission(); + const chatHelpers = buildChatHelpers(); + + const { unmount } = renderHook(() => useResumableSSE(submission, chatHelpers)); + await flushMicrotasks(); + + const sse = getLastSSE(); + const sseCount = mockSSEInstances.length; + + await act(async () => { + sse._emit('message', { + data: JSON.stringify({ + final: true, + conversation: { conversationId: CONV_ID }, + requestMessage: { + messageId: 'msg-1', + conversationId: CONV_ID, + text: 'Hello', + isCreatedByUser: true, + }, + responseMessage: { + messageId: 'resp-1', + parentMessageId: 'msg-1', + conversationId: CONV_ID, + text: 'Done', + isCreatedByUser: false, + }, + }), + }); + }); + + mockSetIsSubmitting.mockClear(); + mockSetShowStopButton.mockClear(); + + await act(async () => { + sse._emit('abort'); + }); + await advanceRetryTimer(1000); + + expect(mockSSEInstances).toHaveLength(sseCount); + expect(mockSetIsSubmitting).not.toHaveBeenCalledWith(true); + expect(mockSetShowStopButton).not.toHaveBeenCalledWith(true); + unmount(); + }); + + it('re-attaches on foreground when the stream closed while the page was hidden', async () => { + (request.post as jest.Mock).mockResolvedValue({ + streamId: 'stream-epoch', + status: 'started', + generationCreatedAt: 1000, + }); + const submission = buildSubmission(); + const chatHelpers = buildChatHelpers(); + + const { unmount } = renderHook(() => useResumableSSE(submission, chatHelpers)); + await flushMicrotasks(); + + const initialSSE = getLastSSE(); + const sseCount = mockSSEInstances.length; + /** The response body ended under a frozen tab, so sse.js dispatched + * neither an error nor an abort — only the closed transport is left. */ + initialSSE.readyState = MOCK_SSE_CLOSED; + + await act(async () => { + document.dispatchEvent(new Event('visibilitychange')); + }); + + expect(mockSSEInstances).toHaveLength(sseCount + 1); + expect(getLastSSE()._url).toBe( + '/api/agents/chat/stream/stream-epoch?resume=true&generationCreatedAt=1000&generationProtocolVersion=2', + ); + unmount(); + }); + + it('leaves a still-open stream alone when the page returns to the foreground', async () => { + const submission = buildSubmission(); + const chatHelpers = buildChatHelpers(); + + const { unmount } = renderHook(() => useResumableSSE(submission, chatHelpers)); + await flushMicrotasks(); + + const initialSSE = getLastSSE(); + const sseCount = mockSSEInstances.length; + initialSSE.readyState = MOCK_SSE_OPEN; + + await act(async () => { + document.dispatchEvent(new Event('visibilitychange')); + }); + + expect(mockSSEInstances).toHaveLength(sseCount); + unmount(); + }); + + it('does not re-attach on foreground once FINAL has closed the stream', async () => { + const submission = buildSubmission(); + const chatHelpers = buildChatHelpers(); + + const { unmount } = renderHook(() => useResumableSSE(submission, chatHelpers)); + await flushMicrotasks(); + + const sse = getLastSSE(); + const sseCount = mockSSEInstances.length; + + await act(async () => { + sse._emit('message', { + data: JSON.stringify({ + final: true, + conversation: { conversationId: CONV_ID }, + requestMessage: { + messageId: 'msg-1', + conversationId: CONV_ID, + text: 'Hello', + isCreatedByUser: true, + }, + responseMessage: { + messageId: 'resp-1', + parentMessageId: 'msg-1', + conversationId: CONV_ID, + text: 'Done', + isCreatedByUser: false, + }, + }), + }); + }); + + expect(sse.readyState).toBe(MOCK_SSE_CLOSED); + + await act(async () => { + document.dispatchEvent(new Event('visibilitychange')); + }); + + expect(mockSSEInstances).toHaveLength(sseCount); + unmount(); + }); + + it('does not re-attach on foreground after a 404 already reconciled the run', async () => { + const { sse, unmount } = await render404Scenario(); + const sseCount = mockSSEInstances.length; + + /** The 404 reconcile leaves the submission installed and the attachment + * pointing at a stream the server no longer has, so only the retirement + * flag keeps the foreground path from resurrecting it. */ + expect(sse.readyState).toBe(MOCK_SSE_CLOSED); + + await act(async () => { + document.dispatchEvent(new Event('visibilitychange')); + }); + + expect(mockSSEInstances).toHaveLength(sseCount); + unmount(); + }); + it('parses and surfaces server-sent error events (no responseCode, JSON data)', async () => { const submission = buildSubmission(); const chatHelpers = buildChatHelpers(); diff --git a/client/src/hooks/SSE/useResumableSSE.ts b/client/src/hooks/SSE/useResumableSSE.ts index 9f418e8076..88d2ed6b00 100644 --- a/client/src/hooks/SSE/useResumableSSE.ts +++ b/client/src/hooks/SSE/useResumableSSE.ts @@ -103,6 +103,10 @@ const clearMatchingDrainAfterAbort = ( : armed; const MAX_RETRIES = 5; +const RECONNECT_BASE_DELAY_MS = 1000; +const RECONNECT_MAX_DELAY_MS = 30000; +const getReconnectDelay = (attempt: number) => + Math.min(RECONNECT_BASE_DELAY_MS * Math.pow(2, attempt - 1), RECONNECT_MAX_DELAY_MS); const START_GENERATION_NETWORK_RETRIES = 3; const START_GENERATION_READINESS_TIMEOUT_MS = 120000; const SERVER_NOT_READY_CODE = 'SERVER_NOT_READY'; @@ -700,6 +704,9 @@ export default function useResumableSSE( const setShowStopButton = useSetRecoilState(store.showStopButtonByIndex(runIndex)); const sseRef = useRef(null); + /** Removes the foreground re-attach listener owned by the newest + * subscription; exactly one is registered at a time. */ + const stopForegroundReattachRef = useRef<(() => void) | null>(null); const reconnectAttemptRef = useRef(0); const reconnectTimeoutRef = useRef(null); const submissionRef = useRef(null); @@ -1180,6 +1187,12 @@ export default function useResumableSSE( let { userMessage } = currentSubmission; let textIndex: number | null = null; let finalReceived = false; + /** This subscription must never be revived. Set by the terminal + * recoveries that do not ride a FINAL or served-error frame — the 404 + * and retry-ceiling reconciles both leave the attachment pointing at a + * stream the server no longer has — and by the dev-only navigation + * simulator below. `finalReceived` covers the frame-carried terminals. */ + let subscriptionRetired = false; const preCreatedStepEvents: Array[0]> = []; const replayPreCreatedStepEvents = () => { if (!isCurrentSubscription() || preCreatedStepEvents.length === 0) { @@ -1502,6 +1515,71 @@ export default function useResumableSSE( sseRef.current === sse && submissionRef.current === currentSubmission; + /** + * Whether THIS connection was closed by this hook. The abort listener + * used to infer that from `reconnectAttemptRef`, but that ref is shared + * across the reconnect ladder and stays raised from the moment a retry is + * scheduled until the replacement connection opens — so a user agent that + * cancelled the replacement before it opened (the ordinary case when the + * retry timer fires while the tab is still backgrounded) read as the + * previous connection's deliberate close, and recovery stopped there. + * Ownership is per connection, so the flag must be too. + */ + let closedByUs = false; + const closeStream = () => { + closedByUs = true; + sse.close(); + }; + + /** + * A suspended page can lose its stream without the transport ever + * reporting it. When a mobile browser freezes the tab, an intermediary + * closes the response body underneath it, and XHR surfaces that as an + * ordinary load — sse.js dispatches nothing for a body that simply ends, + * so neither the error nor the abort recovery above ever runs. Nothing + * else re-reads the conversation while the pane stays mounted, which is + * why the response the run finished writing only appears after a reload. + * + * Re-attaching on the way back costs one request and only when this + * subscription's transport is already gone with no terminal event and no + * recovery of its own in flight. A live job replays what was missed; a + * finished one 404s into the durable refetch. + */ + const handleForegroundReattach = () => { + if ( + document.visibilityState !== 'visible' || + sse.readyState !== SSE.CLOSED || + finalReceived || + subscriptionRetired || + replacementHandoffRef.current || + !isCurrentSubscription() || + !submissionRef.current + ) { + return; + } + logger.log('ResumableSSE', 'Stream was closed while hidden - re-attaching on foreground'); + /** Any backoff still pending targets this same attachment, so returning + * to the app supersedes it rather than waiting the delay out; its + * callback finds a superseded subscription and does nothing. */ + if (reconnectTimeoutRef.current) { + clearTimeout(reconnectTimeoutRef.current); + reconnectTimeoutRef.current = null; + } + reconnectAttemptRef.current = Math.max(reconnectAttemptRef.current, 1); + subscribeToStream( + currentStreamId, + submissionRef.current, + true, + generationCreatedAt, + generationProtocolVersion, + lifecycleSignal, + ); + }; + stopForegroundReattachRef.current?.(); + document.addEventListener('visibilitychange', handleForegroundReattach); + stopForegroundReattachRef.current = () => + document.removeEventListener('visibilitychange', handleForegroundReattach); + sse.addEventListener('open', () => { if (!isCurrentSubscription()) { return; @@ -1645,7 +1723,7 @@ export default function useResumableSSE( removeActiveJob(currentStreamId); clearAttachedGenerationCreatedAt(); (startupConfig?.balance?.enabled ?? false) && balanceQuery.refetch(); - sse.close(); + closeStream(); setStreamId(null); optimisticStreamIdsRef.current.delete(currentStreamId); createdStreamIdsRef.current.delete(currentStreamId); @@ -2102,7 +2180,7 @@ export default function useResumableSSE( }); replacementHandoffRef.current = true; cancelSteerRetryFrames(); - sse.close(); + closeStream(); if (reconnectTimeoutRef.current) { clearTimeout(reconnectTimeoutRef.current); reconnectTimeoutRef.current = null; @@ -2168,7 +2246,7 @@ export default function useResumableSSE( reason, }); reconnectAttemptRef.current = Math.max(reconnectAttemptRef.current, 1); - sse.close(); + closeStream(); if (reconnectTimeoutRef.current) { clearTimeout(reconnectTimeoutRef.current); } @@ -2282,7 +2360,7 @@ export default function useResumableSSE( // Retry the stale epoch; the fenced HTTP endpoint will return the // dedicated replacement response as soon as the new job is visible. reconnectAttemptRef.current = Math.max(reconnectAttemptRef.current, 1); - sse.close(); + closeStream(); reconnectTimeoutRef.current = setTimeout(() => { if (isCurrentSubscription() && submissionRef.current) { subscribeToStream( @@ -2299,7 +2377,7 @@ export default function useResumableSSE( } reconnectAttemptRef.current = Math.max(reconnectAttemptRef.current, 1); - sse.close(); + closeStream(); clearStepMaps(); let persistedMessages: TMessage[] | undefined; const messageQueryKey = [QueryKeys.messages, reconciliationConvoId] as const; @@ -2503,7 +2581,7 @@ export default function useResumableSSE( * * Order matters: check responseCode first since HTTP errors may also include data */ - sse.addEventListener('error', async (e: MessageEvent) => { + const handleTransportFailure = async (e: MessageEvent) => { if (!isCurrentSubscription()) { return; } @@ -2541,7 +2619,7 @@ export default function useResumableSSE( // terminal cleanup or a handoff to a newer generation epoch. reconnectAttemptRef.current = Math.max(reconnectAttemptRef.current, 1); cancelSteerRetryFrames(); - sse.close(); + closeStream(); /** Terminal: drop any in-flight live estimate so the gauge doesn't * keep counting stale streamed output after the stream ends */ resetLive({ ...currentSubmission, userMessage }); @@ -2746,6 +2824,7 @@ export default function useResumableSSE( removeConvoFromAllQueries(queryClient, currentStreamId); } } + subscriptionRetired = true; setIsSubmitting(false); setShowStopButton(false); @@ -2820,7 +2899,7 @@ export default function useResumableSSE( } logger.log('ResumableSSE', 'Server-sent error event received:', e.data); cancelSteerRetryFrames(); - sse.close(); + closeStream(); /** FLUSH (not cancel): the error card below is built from the cache * tail, so queued tokens must land first — and a stale trailing * frame must never overwrite the error write. */ @@ -2942,14 +3021,14 @@ export default function useResumableSSE( if (reconnectAttemptRef.current < MAX_RETRIES) { // Increment counter BEFORE close() so abort handler knows we're reconnecting reconnectAttemptRef.current++; - const delay = Math.min(1000 * Math.pow(2, reconnectAttemptRef.current - 1), 30000); + const delay = getReconnectDelay(reconnectAttemptRef.current); logger.log( 'ResumableSSE', `Reconnecting in ${delay}ms (attempt ${reconnectAttemptRef.current}/${MAX_RETRIES})`, ); - sse.close(); + closeStream(); reconnectTimeoutRef.current = setTimeout(() => { if (isCurrentSubscription() && submissionRef.current) { @@ -2971,7 +3050,7 @@ export default function useResumableSSE( setShowStopButton(generationCreatedAt != null); } else { logger.error('ResumableSSE', 'Max reconnect attempts reached'); - sse.close(); + closeStream(); flushPendingDeltas(); const recoveryConvoId = currentSubmission.conversation?.conversationId ?? currentStreamId; let status: Awaited> | undefined; @@ -3162,6 +3241,7 @@ export default function useResumableSSE( ) { removeConvoFromAllQueries(queryClient, currentStreamId); } + subscriptionRetired = true; setIsSubmitting(false); setShowStopButton(false); let recoveryOutcome: 'completed' | 'aborted' | 'error' = 'error'; @@ -3182,17 +3262,46 @@ export default function useResumableSSE( createdStreamIdsRef.current.delete(currentStreamId); reconnectAttemptRef.current = 0; } - }); + }; + + sse.addEventListener('error', handleTransportFailure); /** - * Abort event - fired when sse.close() is called (intentional close). - * This happens on cleanup/navigation OR when error handler closes to reconnect. - * Only reset state if we're NOT in a reconnection cycle. + * Abort event - fired when the underlying XHR is cancelled, either by one + * of this hook's own closes or by the user agent. */ sse.addEventListener('abort', () => { if (!isCurrentSubscription()) { return; } + + /** + * A cancellation this hook did not issue came from the user agent, + * which cancels in-flight requests when a mobile browser is + * backgrounded or the page is frozen — and the generation it was + * carrying is still running server-side. + * + * Treating that as a deliberate close is what strands the response: + * the pane goes idle holding whatever partial content arrived before + * the switch, looking finished, and nothing re-reads the conversation + * until a reload or a navigation remounts the messages query. It is a + * dropped connection by every meaningful measure, so hand it to the + * transport-failure path verbatim rather than re-deriving a ladder + * beside it: that one already climbs its backoff, adjudicates the + * retry ceiling against durable status, and terminalizes into the + * refetch when the job turns out to have finished meanwhile. + */ + if (!closedByUs) { + logger.log( + 'ResumableSSE', + 'Stream aborted by the user agent - recovering as transport failure', + ); + void handleTransportFailure({ + responseCode: 0, + } as MessageEvent & { responseCode?: number }); + return; + } + if (replacementHandoffRef.current) { logger.log('ResumableSSE', 'Stream closed for generation handoff - preserving state'); return; @@ -3244,7 +3353,8 @@ export default function useResumableSSE( /** Simulate clean close (navigation away) - triggers abort event → no reconnection */ debugWindow.__closeClean = () => { logger.log('Debug', 'Simulating clean close (navigation away)...'); - sse.close(); + subscriptionRetired = true; + closeStream(); }; } }, @@ -3520,6 +3630,8 @@ export default function useResumableSSE( clearTimeout(reconnectTimeoutRef.current); reconnectTimeoutRef.current = null; } + stopForegroundReattachRef.current?.(); + stopForegroundReattachRef.current = null; // Close SSE but do NOT dispatch cancel - navigation should not abort if (sseRef.current) { sseRef.current.close(); @@ -4033,6 +4145,8 @@ export default function useResumableSSE( cancelAnimationFrame(frameId); } steerRetryFrames.clear(); + stopForegroundReattachRef.current?.(); + stopForegroundReattachRef.current = null; // Reset reconnect counter before closing (so abort handler doesn't think we're reconnecting) reconnectAttemptRef.current = 0; if (sseRef.current) {