diff --git a/client/src/data-provider/SSE/__tests__/queries.test.ts b/client/src/data-provider/SSE/__tests__/queries.test.ts index 2afb0e1256..b408a9d571 100644 --- a/client/src/data-provider/SSE/__tests__/queries.test.ts +++ b/client/src/data-provider/SSE/__tests__/queries.test.ts @@ -43,6 +43,7 @@ import { queueTitleGeneration, useActiveJobs, resetActiveJobsGrace, + extendActiveJobsGrace, getActiveJobsRefetchInterval, ACTIVE_JOBS_POLL_MS, ACTIVE_JOBS_SUCCESSOR_GRACE_MS, @@ -130,6 +131,36 @@ describe('useActiveJobs successor grace', () => { expect(getActiveJobsRefetchInterval({ activeJobIds: [] })).toBe(ACTIVE_JOBS_POLL_MS); }); + it('keeps listening for as long as a successor is known to be owed', () => { + /** The deterministic path: a queued turn the backend has taken ownership + * of needs no window, because the client already knows a run is coming. */ + expect(getActiveJobsRefetchInterval({ activeJobIds: [] }, true)).toBe(ACTIVE_JOBS_POLL_MS); + + now += ACTIVE_JOBS_SUCCESSOR_GRACE_MS * 10; + expect(getActiveJobsRefetchInterval({ activeJobIds: [] }, true)).toBe(ACTIVE_JOBS_POLL_MS); + }); + + it('restarts the window from a live observation the list never saw', () => { + /** A pane attached to the run a queued turn produced, and that run + * finished before the list polled. The list's clock is stale; the + * observation is what keeps an unpredicted continuation discoverable. */ + now += ACTIVE_JOBS_SUCCESSOR_GRACE_MS * 3; + expect(getActiveJobsRefetchInterval({ activeJobIds: [] })).toBe(false); + + extendActiveJobsGrace(); + expect(getActiveJobsRefetchInterval({ activeJobIds: [] })).toBe(ACTIVE_JOBS_POLL_MS); + + now += ACTIVE_JOBS_SUCCESSOR_GRACE_MS + 1; + expect(getActiveJobsRefetchInterval({ activeJobIds: [] })).toBe(false); + }); + + it('stops once a successor is no longer owed', () => { + expect(getActiveJobsRefetchInterval({ activeJobIds: [] }, true)).toBe(ACTIVE_JOBS_POLL_MS); + + now += ACTIVE_JOBS_SUCCESSOR_GRACE_MS + 1; + expect(getActiveJobsRefetchInterval({ activeJobIds: [] }, false)).toBe(false); + }); + it('stops once the handover window has passed without a successor', () => { expect(getActiveJobsRefetchInterval({ activeJobIds: ['convo-1'] })).toBe(ACTIVE_JOBS_POLL_MS); diff --git a/client/src/data-provider/SSE/__tests__/queuedTurns.test.ts b/client/src/data-provider/SSE/__tests__/queuedTurns.test.ts index be7931d9be..3e1db9a5b9 100644 --- a/client/src/data-provider/SSE/__tests__/queuedTurns.test.ts +++ b/client/src/data-provider/SSE/__tests__/queuedTurns.test.ts @@ -34,6 +34,7 @@ import { isDefiniteQueuedTurnRejection, isDefiniteQueuedTurnsUnsupported, shouldPollAgentQueuedTurns, + isQueuedTurnSuccessorOwed, useAgentQueuedTurns, } from '../queuedTurns'; @@ -144,6 +145,35 @@ describe('Agent queued-turn data adapter', () => { expect(shouldPollAgentQueuedTurns([{ status: 'queued' }])).toBe(true); }); + it('counts an admitted turn as an owed run, unlike the receipt poll', () => { + /** + * The receipt poll stops at `admitted` because nothing is left to wait + * for. A pane that is not attached is in the opposite position: `admitted` + * is the strongest evidence it will get that a run exists, and the receipt + * is dropped from the projection right after — so whoever decides whether + * to keep listening for that run has to count it. + */ + expect(shouldPollAgentQueuedTurns([{ status: 'admitted' }])).toBe(false); + expect(isQueuedTurnSuccessorOwed([{ status: 'admitted' }])).toBe(true); + }); + + it('owes a run for the same live statuses the receipt poll waits on', () => { + expect(isQueuedTurnSuccessorOwed([{ status: 'queued' }])).toBe(true); + expect(isQueuedTurnSuccessorOwed([{ status: 'claimed' }])).toBe(true); + expect( + isQueuedTurnSuccessorOwed([ + { status: 'claimed', failure: { code: 'ADMISSION_INDETERMINATE' } }, + ]), + ).toBe(false); + }); + + it('owes nothing once a turn can no longer produce a run', () => { + expect(isQueuedTurnSuccessorOwed([{ status: 'cancelled' }])).toBe(false); + expect(isQueuedTurnSuccessorOwed([{ status: 'dead' }])).toBe(false); + expect(isQueuedTurnSuccessorOwed([])).toBe(false); + expect(isQueuedTurnSuccessorOwed(undefined)).toBe(false); + }); + it('refreshes stopped reconciliation work on every mount and focus', () => { renderHook(() => useAgentQueuedTurns('conversation/one', true)); diff --git a/client/src/data-provider/SSE/queries.ts b/client/src/data-provider/SSE/queries.ts index d533155949..cf0d8b8889 100644 --- a/client/src/data-provider/SSE/queries.ts +++ b/client/src/data-provider/SSE/queries.ts @@ -223,14 +223,16 @@ export const ACTIVE_JOBS_POLL_MS = 5_000; /** * How long to keep asking after the last job this client knew about ended. * - * A successor the server admits for itself — a queued turn handed to the - * backend, a background-tool continuation — starts as part of terminalizing - * the run before it. But finishing a run also empties this list, partly - * because the client removes its own job optimistically the moment `FINAL` - * lands. Stopping the poll on an empty list therefore goes quiet at exactly - * the moment a successor appears, and with the tab already focused there is no - * focus refetch to recover: the pane waits for a reload. Keep asking across - * that handover instead. + * Fallback only, for a successor nothing local predicts — a background-tool + * continuation, say. A queued turn is known in advance and reports itself + * through `expectsSuccessor`, which needs no window at all. + * + * A successor the server admits for itself starts as part of terminalizing the + * run before it. But finishing a run also empties this list, partly because + * the client removes its own job optimistically the moment `FINAL` lands. + * Stopping the poll on an empty list therefore goes quiet at exactly the + * moment a successor appears, and with the tab already focused there is no + * focus refetch to recover: the pane waits for a reload. */ export const ACTIVE_JOBS_SUCCESSOR_GRACE_MS = 30_000; @@ -243,17 +245,44 @@ export function resetActiveJobsGrace(): void { lastListedActiveJobsAt = 0; } -export function getActiveJobsRefetchInterval(data?: ActiveJobsResponse): number | false { +/** + * A generation was just observed live by some other means — a pane attached to + * the run a queued turn produced — without the list ever reporting it, because + * the run started and finished between two polls. The list's own notion of + * "recently active" is therefore stale, and an unpredicted successor after + * that run (a background-tool continuation) would find no poll left to notice + * it. Restart the handover window from this observation instead. + */ +export function extendActiveJobsGrace(): void { + lastListedActiveJobsAt = Date.now(); +} + +export function getActiveJobsRefetchInterval( + data?: ActiveJobsResponse, + expectsSuccessor = false, +): number | false { if ((data?.activeJobIds?.length ?? 0) > 0) { lastListedActiveJobsAt = Date.now(); return ACTIVE_JOBS_POLL_MS; } + /** A caller that knows the backend owes it a run says so outright, and keeps + * listening for as long as that stays true — no guess at how long admission + * takes. The window below is only for successors nothing local predicts. */ + if (expectsSuccessor) { + return ACTIVE_JOBS_POLL_MS; + } return Date.now() - lastListedActiveJobsAt < ACTIVE_JOBS_SUCCESSOR_GRACE_MS ? ACTIVE_JOBS_POLL_MS : false; } -export function useActiveJobs(enabled = true) { +/** + * @param expectsSuccessor Set by a caller holding positive evidence that the + * backend owes this client a run it will start itself — a queued turn admitted + * server-side, say. Observers each keep their own interval, so declaring it + * affects only the caller that knows. + */ +export function useActiveJobs(enabled = true, expectsSuccessor = false) { return useQuery({ queryKey: [QueryKeys.activeJobs], queryFn: () => dataService.getActiveJobs(), @@ -265,7 +294,8 @@ export function useActiveJobs(enabled = true) { * invisible until some unrelated refetch, and returning to the tab is * exactly when a pane needs to know whether its history has moved. */ refetchOnWindowFocus: 'always', - refetchInterval: getActiveJobsRefetchInterval, + refetchInterval: (data?: ActiveJobsResponse) => + getActiveJobsRefetchInterval(data, expectsSuccessor), retry: false, }); } diff --git a/client/src/data-provider/SSE/queuedTurns.ts b/client/src/data-provider/SSE/queuedTurns.ts index df991619aa..ea7097adc6 100644 --- a/client/src/data-provider/SSE/queuedTurns.ts +++ b/client/src/data-provider/SSE/queuedTurns.ts @@ -95,6 +95,28 @@ export function shouldPollAgentQueuedTurns( ); } +/** + * Whether the backend owes this conversation a run it will start itself. + * + * Wider than {@link shouldPollAgentQueuedTurns} in exactly one way: `admitted`. + * The receipt poll stops there because there is nothing left to wait for — but + * for a pane that is not attached, `admitted` is the strongest evidence it will + * get that a run exists, and the receipt is dropped from the projection almost + * immediately after (the chip goes, so its id leaves the known set). Anyone + * deciding whether to keep listening for that run must count it. + */ +export function isQueuedTurnSuccessorOwed(receipts: unknown): boolean { + return ( + Array.isArray(receipts) && + receipts.some( + (item: AgentQueuedTurnReceipt) => + item.status === 'queued' || + item.status === 'admitted' || + (item.status === 'claimed' && item.failure?.code !== 'ADMISSION_INDETERMINATE'), + ) + ); +} + export function useAgentQueuedTurns( conversationId: string, enabled: boolean, diff --git a/client/src/hooks/SSE/__tests__/useResumeOnLoad.spec.tsx b/client/src/hooks/SSE/__tests__/useResumeOnLoad.spec.tsx index 939ca44878..e3a5d039e8 100644 --- a/client/src/hooks/SSE/__tests__/useResumeOnLoad.spec.tsx +++ b/client/src/hooks/SSE/__tests__/useResumeOnLoad.spec.tsx @@ -11,11 +11,24 @@ import store from '~/store'; const mockUseStreamStatus = jest.fn(); const mockUseActiveJobs = jest.fn(); +const mockUseAgentQueuedTurns = jest.fn(); +const mockExtendActiveJobsGrace = jest.fn(); jest.mock('~/data-provider', () => ({ useStreamStatus: (conversationId: string | undefined, enabled: boolean) => mockUseStreamStatus(conversationId, enabled), - useActiveJobs: (enabled?: boolean) => mockUseActiveJobs(enabled), + useActiveJobs: (enabled?: boolean, expectsSuccessor?: boolean) => + mockUseActiveJobs(enabled, expectsSuccessor), + useAgentQueuedTurns: (conversationId: string, enabled: boolean) => + mockUseAgentQueuedTurns(conversationId, enabled), + /** Real predicate and real window: admission semantics are the thing under + * test here, and a restated copy in a mock is exactly the drift sharing a + * function is meant to avoid. */ + isQueuedTurnSuccessorOwed: jest.requireActual('~/data-provider/SSE/queuedTurns') + .isQueuedTurnSuccessorOwed, + ACTIVE_JOBS_SUCCESSOR_GRACE_MS: jest.requireActual('~/data-provider/SSE/queries') + .ACTIVE_JOBS_SUCCESSOR_GRACE_MS, + extendActiveJobsGrace: () => mockExtendActiveJobsGrace(), streamStatusQueryKey: (conversationId: string) => ['streamStatus', conversationId], })); @@ -101,6 +114,8 @@ function renderUseResumeOnLoad({ defaultOptions: { queries: { retry: false } }, }); let setSubmissionState: ((submission: TSubmission | null) => void) | undefined; + let setIsSubmittingState: ((value: boolean) => void) | undefined; + let setAttachedEpochState: ((value: number | null) => void) | undefined; const initializeState = (snapshot: MutableSnapshot) => { snapshot.set(store.conversationByIndex(0), buildConversation(conversationId)); snapshot.set(store.submissionByIndex(0), submission); @@ -120,6 +135,10 @@ function renderUseResumeOnLoad({ const SubmissionProbe = () => { const currentSubmission = useRecoilValue(store.submissionByIndex(0)); setSubmissionState = useSetRecoilState(store.submissionByIndex(0)); + setIsSubmittingState = useSetRecoilState(store.isSubmittingFamily(0)); + setAttachedEpochState = useSetRecoilState( + store.activeGenerationCreatedAtByConvoId(conversationId), + ); onSubmission?.(currentSubmission); return null; }; @@ -163,9 +182,13 @@ function renderUseResumeOnLoad({ queryClient, getMessages, setSubmission: (nextSubmission: TSubmission | null) => setSubmissionState?.(nextSubmission), - ...renderHook(() => useResumeOnLoad(conversationId, getMessages, 0, messagesLoaded), { - wrapper, - }), + setIsSubmitting: (value: boolean) => setIsSubmittingState?.(value), + setAttachedGenerationCreatedAt: (value: number | null) => setAttachedEpochState?.(value), + ...renderHook( + (props?: { conversationId?: string }) => + useResumeOnLoad(props?.conversationId ?? conversationId, getMessages, 0, messagesLoaded), + { wrapper, initialProps: { conversationId } }, + ), }; } @@ -180,6 +203,9 @@ describe('useResumeOnLoad', () => { }); mockUseActiveJobs.mockReset(); mockUseActiveJobs.mockReturnValue({ data: { activeJobIds: [] }, dataUpdatedAt: 1 }); + mockUseAgentQueuedTurns.mockReset(); + mockUseAgentQueuedTurns.mockReturnValue({ data: [], dataUpdatedAt: 1 }); + mockExtendActiveJobsGrace.mockReset(); }); afterEach(() => { @@ -752,6 +778,789 @@ describe('useResumeOnLoad', () => { expect(mockUseStreamStatus).not.toHaveBeenCalledWith(CONVERSATION_ID, true); }); + it('declares an owed successor so the active list keeps being asked', async () => { + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + /** + * A queued turn the backend has taken ownership of: `useQueueDrain` + * declines the boundary, so no submission is built here and the run must + * be recognised from the active list — which finishing the previous run + * empties. Saying the successor is owed is what keeps that list live, + * rather than betting on how long admission takes. + */ + mockUseAgentQueuedTurns.mockReturnValue({ data: [{ status: 'queued' }], dataUpdatedAt: 2 }); + + renderUseResumeOnLoad({ messages: [buildUserMessage(CONVERSATION_ID)] }); + await act(async () => { + await Promise.resolve(); + }); + + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + }); + + it('declares nothing owed while its own run is the one in flight', async () => { + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + mockUseAgentQueuedTurns.mockReturnValue({ data: [{ status: 'queued' }], dataUpdatedAt: 2 }); + + /** The queued turn is waiting behind a run this pane is already + * streaming; that attachment is what will report the successor. */ + renderUseResumeOnLoad({ + submission: buildSubmission(CONVERSATION_ID), + isSubmitting: true, + messages: [buildUserMessage(CONVERSATION_ID)], + }); + await act(async () => { + await Promise.resolve(); + }); + + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, false); + }); + + it('declares nothing owed for a turn the server cannot admit', async () => { + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ status: 'claimed', failure: { code: 'ADMISSION_INDETERMINATE' } }], + dataUpdatedAt: 2, + }); + + renderUseResumeOnLoad({ messages: [buildUserMessage(CONVERSATION_ID)] }); + await act(async () => { + await Promise.resolve(); + }); + + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, false); + }); + + it('keeps listening after the admitted receipt is dropped, until the list reports the run', async () => { + jest.useFakeTimers(); + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + /** + * Admission outlasted the handover window, so the wall-clock fallback is + * already spent by the time the turn is admitted. The receipt reports + * `admitted` for one fetch and is then dropped from the projection — + * before the active list has observed the run it started. Keying on the + * receipt alone stops listening in exactly that gap. + */ + mockUseAgentQueuedTurns.mockReturnValue({ data: [{ status: 'queued' }], dataUpdatedAt: 2 }); + const { rerender } = renderUseResumeOnLoad({ messages: [buildUserMessage(CONVERSATION_ID)] }); + await act(async () => { + await Promise.resolve(); + }); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ status: 'admitted' }], + dataUpdatedAt: 3, + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + + mockUseAgentQueuedTurns.mockReturnValue({ data: [], dataUpdatedAt: 4 }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + /** Receipt gone, list still empty: the owed state has to carry this. */ + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + + /** The list naming the conversation is an announcement, not delivery — + * it carries no generation identity, so it cannot prove the successor + * exists. It arms a status read; attaching to a generation other than + * the predecessor's is what finally releases the owed state. */ + mockUseActiveJobs.mockReturnValue({ + data: { activeJobIds: [CONVERSATION_ID] }, + dataUpdatedAt: 5, + }); + mockUseStreamStatus.mockReturnValue(ACTIVE_STATUS); + await act(async () => { + jest.advanceTimersByTime(ACTIVE_JOB_REARM_INTERVAL_MS + 1); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, false); + jest.useRealTimers(); + }); + + it('arms off an owed receipt alone, without waiting for the list', async () => { + const observedSubmissions: Array = []; + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + const { rerender } = renderUseResumeOnLoad({ + messages: [buildUserMessage(CONVERSATION_ID)], + onSubmission: (currentSubmission) => observedSubmissions.push(currentSubmission), + }); + await act(async () => { + await Promise.resolve(); + }); + + mockUseStreamStatus.mockClear(); + /** The run started and will be over before the next list poll. The + * status read is the authoritative answer and must not wait on it. */ + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ status: 'admitted' }], + dataUpdatedAt: 2, + }); + mockUseStreamStatus.mockReturnValue(ACTIVE_STATUS); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + + expect(mockUseStreamStatus).toHaveBeenCalledWith(CONVERSATION_ID, true); + const attached = observedSubmissions[observedSubmissions.length - 1] as + | (TSubmission & { resumeStreamId?: string }) + | null; + expect(attached?.resumeStreamId).toBe(CONVERSATION_ID); + }); + + it('lets the owed state lapse once the receipt has gone quiet for the window', async () => { + jest.useFakeTimers(); + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + mockUseAgentQueuedTurns.mockReturnValue({ data: [{ status: 'queued' }], dataUpdatedAt: 2 }); + const { rerender } = renderUseResumeOnLoad({ messages: [buildUserMessage(CONVERSATION_ID)] }); + await act(async () => { + await Promise.resolve(); + }); + + /** A long wait behind an unadmitted turn must not burn the window: the + * countdown starts only when the receipt stops reporting. */ + await act(async () => { + jest.advanceTimersByTime(ACTIVE_JOB_REARM_INTERVAL_MS * 20); + }); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + + mockUseAgentQueuedTurns.mockReturnValue({ data: [], dataUpdatedAt: 3 }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + + await act(async () => { + jest.advanceTimersByTime(30_000 + 1); + }); + rerender(); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, false); + jest.useRealTimers(); + }); + + it('does not count the predecessor attachment as delivery of the successor', async () => { + const observedSubmissions: Array = []; + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + /** + * The turn was queued behind this pane's own run, so its receipt is + * admitted while that predecessor is still attached. If that attachment + * counted as delivery the latch would be cleared — or never recorded — + * and once the predecessor closes and the receipt is gone, nothing would + * be left to notice the successor. + */ + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ status: 'admitted' }], + dataUpdatedAt: 2, + }); + const { rerender, setIsSubmitting, setAttachedGenerationCreatedAt } = renderUseResumeOnLoad({ + submission: buildSubmission(CONVERSATION_ID), + isSubmitting: true, + attachedGenerationCreatedAt: 1000, + messages: [buildUserMessage(CONVERSATION_ID)], + onSubmission: (currentSubmission) => observedSubmissions.push(currentSubmission), + }); + await act(async () => { + await Promise.resolve(); + }); + /** Attached to the predecessor: nothing to poll for yet, nothing to arm. */ + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, false); + + /** Predecessor ends: its FINAL clears the epoch and submitting state and + * leaves the finished submission installed; the list is already empty + * from the optimistic removal, and the receipt is dropped. */ + mockUseAgentQueuedTurns.mockReturnValue({ data: [], dataUpdatedAt: 3 }); + mockUseStreamStatus.mockClear(); + mockUseStreamStatus.mockReturnValue(ACTIVE_STATUS); + await act(async () => { + setIsSubmitting(false); + setAttachedGenerationCreatedAt(null); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + + /** The latch carried the handover: with the receipt gone and the list + * empty, the pane still declared the successor owed... */ + expect(mockUseActiveJobs).toHaveBeenCalledWith(true, true); + /** ...which armed the status read, which attached to the successor. */ + expect(mockUseStreamStatus).toHaveBeenCalledWith(CONVERSATION_ID, true); + const attached = observedSubmissions[observedSubmissions.length - 1] as + | (TSubmission & { resumeStreamId?: string; resumeGenerationCreatedAt?: number }) + | null; + expect(attached?.resumeStreamId).toBe(CONVERSATION_ID); + expect(attached?.resumeGenerationCreatedAt).toBe(4242); + /** And attaching to a generation other than the predecessor's is the + * delivery that releases it. */ + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, false); + }); + + it('treats attachment to a different generation as delivery', async () => { + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ status: 'admitted' }], + dataUpdatedAt: 2, + }); + const { rerender, setIsSubmitting, setAttachedGenerationCreatedAt } = renderUseResumeOnLoad({ + submission: buildSubmission(CONVERSATION_ID), + isSubmitting: true, + attachedGenerationCreatedAt: 1000, + messages: [buildUserMessage(CONVERSATION_ID)], + }); + await act(async () => { + await Promise.resolve(); + }); + + /** Receipt gone, and this pane is now attached to a newer epoch — that + * IS the successor, so nothing remains owed once it detaches again. */ + mockUseAgentQueuedTurns.mockReturnValue({ data: [], dataUpdatedAt: 3 }); + await act(async () => { + setAttachedGenerationCreatedAt(2000); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + await act(async () => { + setIsSubmitting(false); + setAttachedGenerationCreatedAt(null); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, false); + }); + + it('does not let a listed, unattached predecessor deliver the successor', async () => { + const observedSubmissions: Array = []; + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + /** + * The predecessor runs elsewhere: this pane is not attached, but the + * active list names the conversation. The list has no generation identity, + * so being listed cannot prove the successor exists — treating it as + * delivery means the latch is never recorded, and a successor that runs + * inside a poll gap after the predecessor ends is never seen. + */ + mockUseActiveJobs.mockReturnValue({ + data: { activeJobIds: [CONVERSATION_ID] }, + dataUpdatedAt: 2, + }); + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'q1', status: 'queued', expectedPredecessorCreatedAt: 1000 }], + dataUpdatedAt: 2, + }); + const { rerender } = renderUseResumeOnLoad({ + messages: [buildUserMessage(CONVERSATION_ID)], + onSubmission: (currentSubmission) => observedSubmissions.push(currentSubmission), + }); + await act(async () => { + await Promise.resolve(); + }); + + /** Predecessor ends and the successor is admitted, runs, and finishes + * between two polls: list empty again, receipt already dropped. The + * first arm happened at mount (the listed predecessor announced), so + * step past the throttle before the transition. */ + const nowSpy = jest.spyOn(Date, 'now'); + const later = Date.now() + ACTIVE_JOB_REARM_INTERVAL_MS + 1; + nowSpy.mockImplementation(() => later); + mockUseActiveJobs.mockReturnValue({ data: { activeJobIds: [] }, dataUpdatedAt: 3 }); + mockUseAgentQueuedTurns.mockReturnValue({ data: [], dataUpdatedAt: 3 }); + mockUseStreamStatus.mockClear(); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + + /** Still owed — the latch survived the listed predecessor — so the pane + * keeps asking and re-reads status/history instead of going idle. */ + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + expect(mockUseStreamStatus).toHaveBeenCalledWith(CONVERSATION_ID, true); + nowSpy.mockRestore(); + }); + + it('learns the predecessor from a generation seen while the turn still waits', async () => { + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + /** Enqueued elsewhere without a boundary, so the only way to learn which + * generation is the predecessor is to see it live while the receipt is + * still queued. Once the turn is admitted, that same generation must + * not read as the successor just because a boundary was never sent. */ + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'q1', status: 'queued' }], + dataUpdatedAt: 2, + }); + const { rerender, setIsSubmitting, setAttachedGenerationCreatedAt } = renderUseResumeOnLoad({ + submission: buildSubmission(CONVERSATION_ID), + isSubmitting: false, + attachedGenerationCreatedAt: null, + messages: [buildUserMessage(CONVERSATION_ID)], + }); + await act(async () => { + await Promise.resolve(); + }); + + /** Predecessor becomes live here while the turn is still queued. */ + await act(async () => { + setIsSubmitting(true); + setAttachedGenerationCreatedAt(1000); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + /** Turn admitted while that same generation is still attached. */ + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'q1', status: 'admitted' }], + dataUpdatedAt: 3, + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + /** Predecessor ends; receipt dropped. */ + await act(async () => { + setIsSubmitting(false); + setAttachedGenerationCreatedAt(null); + }); + mockUseAgentQueuedTurns.mockReturnValue({ data: [], dataUpdatedAt: 4 }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + }); + + it('re-arms history only on a receipt transition, not on every receipt poll', async () => { + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + const { rerender, queryClient } = renderUseResumeOnLoad({ + messages: [buildUserMessage(CONVERSATION_ID)], + }); + await act(async () => { + await Promise.resolve(); + }); + const invalidate = jest.spyOn(queryClient, 'invalidateQueries').mockResolvedValue(undefined); + const historyRefetches = () => + invalidate.mock.calls.filter(([options]) => { + const key = (options as { queryKey?: unknown } | undefined)?.queryKey; + return Array.isArray(key) && key[0] === QueryKeys.messages; + }).length; + + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'q1', status: 'queued' }], + dataUpdatedAt: 2, + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + expect(historyRefetches()).toBe(1); + + /** The same queued receipt, refetched on its two-second heartbeat while + * admission is delayed. Nothing has happened; nothing should download. */ + const nowSpy = jest.spyOn(Date, 'now'); + let clock = Date.now(); + nowSpy.mockImplementation(() => clock); + for (const stamp of [3, 4, 5]) { + clock += ACTIVE_JOB_REARM_INTERVAL_MS + 1; + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'q1', status: 'queued' }], + dataUpdatedAt: stamp, + }); + /** The active list polls on its own heartbeat while a successor is + * owed, so the arm DOES re-enter here — only the gate keeps it from + * downloading history again. */ + mockUseActiveJobs.mockReturnValue({ data: { activeJobIds: [] }, dataUpdatedAt: stamp }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + } + expect(historyRefetches()).toBe(1); + + /** Admission is a transition: now history matters again. */ + clock += ACTIVE_JOB_REARM_INTERVAL_MS + 1; + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'q1', status: 'admitted' }], + dataUpdatedAt: 6, + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + expect(historyRefetches()).toBe(2); + nowSpy.mockRestore(); + }); + + it('expires the owed state on an absolute window even while its conversation is off-screen', async () => { + jest.useFakeTimers(); + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'q1', status: 'queued' }], + dataUpdatedAt: 2, + }); + const { rerender, queryClient } = renderUseResumeOnLoad({ + messages: [buildUserMessage(CONVERSATION_ID)], + }); + await act(async () => { + await Promise.resolve(); + }); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + + /** Navigate away. No receipts are observed for an off-screen + * conversation, so the latch must start its window now rather than + * waiting for a return it may never get. */ + mockUseAgentQueuedTurns.mockReturnValue({ data: [], dataUpdatedAt: 3 }); + const invalidate = jest.spyOn(queryClient, 'invalidateQueries').mockResolvedValue(undefined); + rerender({ conversationId: STALE_CONVERSATION_ID }); + await act(async () => { + await Promise.resolve(); + }); + await act(async () => { + jest.advanceTimersByTime(30_000 + 1); + }); + + /** Window spent off-screen: one repair refetch for THAT conversation. */ + expect(invalidate).toHaveBeenCalledWith({ + queryKey: [QueryKeys.messages, CONVERSATION_ID], + }); + + /** Returning must not reopen it. */ + rerender({ conversationId: CONVERSATION_ID }); + await act(async () => { + await Promise.resolve(); + }); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, false); + jest.useRealTimers(); + }); + + it('settles when the successor attaches while its admitted receipt is still cached', async () => { + const depthErrors: string[] = []; + const consoleError = jest.spyOn(console, 'error').mockImplementation((...args) => { + const text = args.map(String).join(' '); + if (text.includes('Maximum update depth')) { + depthErrors.push(text); + } + }); + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + /** + * The receipt carries the predecessor epoch and stays `admitted` in the + * cache for a fetch after the successor attaches. Attachment to a + * different generation is delivery — but the same still-reporting + * receipt must not immediately relatch, or clear/relatch cycles + * synchronously until the cache moves on. + */ + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'q1', status: 'admitted', effectivePredecessorCreatedAt: 1000 }], + dataUpdatedAt: 2, + }); + const { rerender, setIsSubmitting, setAttachedGenerationCreatedAt } = renderUseResumeOnLoad({ + submission: buildSubmission(CONVERSATION_ID), + isSubmitting: false, + attachedGenerationCreatedAt: null, + messages: [buildUserMessage(CONVERSATION_ID)], + }); + await act(async () => { + await Promise.resolve(); + }); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + + /** The successor (a different generation than the boundary) attaches. */ + await act(async () => { + setIsSubmitting(true); + setAttachedGenerationCreatedAt(2000); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + + expect(depthErrors).toHaveLength(0); + /** Delivered: with the receipt still cached, nothing is owed and the + * fallback window was restarted from this live observation. */ + expect(mockExtendActiveJobsGrace).toHaveBeenCalledTimes(1); + await act(async () => { + setIsSubmitting(false); + setAttachedGenerationCreatedAt(null); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, false); + consoleError.mockRestore(); + }); + + it("does not let a remount's cached receipt reopen a window that already started", async () => { + jest.useFakeTimers(); + jest.setSystemTime(1_000_000); + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'q1', status: 'queued' }], + dataUpdatedAt: 900_000, + }); + const { rerender, queryClient } = renderUseResumeOnLoad({ + messages: [buildUserMessage(CONVERSATION_ID)], + }); + await act(async () => { + await Promise.resolve(); + }); + + /** Leave: the window starts now (quietSince = 1_000_000). Receipts are + * fetched per conversation, so the other conversation reports none. */ + mockUseAgentQueuedTurns.mockReturnValue({ data: [], dataUpdatedAt: 900_000 }); + rerender({ conversationId: STALE_CONVERSATION_ID }); + await act(async () => { + await Promise.resolve(); + }); + await act(async () => { + jest.advanceTimersByTime(10_000); + }); + + /** Return: React Query first exposes the receipt cached BEFORE the + * window started, while its refetch is still in flight. That stale + * observation must not reset the clock. */ + const invalidate = jest.spyOn(queryClient, 'invalidateQueries').mockResolvedValue(undefined); + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'q1', status: 'queued' }], + dataUpdatedAt: 900_000, + }); + rerender({ conversationId: CONVERSATION_ID }); + await act(async () => { + await Promise.resolve(); + }); + /** The authoritative refetch: the turn is gone. */ + mockUseAgentQueuedTurns.mockReturnValue({ data: [], dataUpdatedAt: 1_010_500 }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + + /** Absolute expiry: 30s from the ORIGINAL start, not from the return. */ + await act(async () => { + jest.advanceTimersByTime(20_000 + 1); + }); + expect(invalidate).toHaveBeenCalledWith({ + queryKey: [QueryKeys.messages, CONVERSATION_ID], + }); + rerender(); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, false); + jest.useRealTimers(); + }); + + it('tracks delivery per queued turn, not per receipt set', async () => { + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + /** + * Two turns queued elsewhere behind the same generation. Delivering the + * first must not mark the second delivered — and the first successor, + * now the live link the second turn waits behind, is the SECOND turn's + * predecessor, not its successor. + */ + mockUseAgentQueuedTurns.mockReturnValue({ + data: [ + { queuedTurnId: 'q1', status: 'queued', expectedPredecessorCreatedAt: 1000 }, + { queuedTurnId: 'q2', status: 'queued', expectedPredecessorCreatedAt: 1000 }, + ], + dataUpdatedAt: 2, + }); + const { rerender, setIsSubmitting, setAttachedGenerationCreatedAt } = renderUseResumeOnLoad({ + submission: buildSubmission(CONVERSATION_ID), + isSubmitting: true, + attachedGenerationCreatedAt: 1000, + messages: [buildUserMessage(CONVERSATION_ID)], + }); + await act(async () => { + await Promise.resolve(); + }); + + /** q1 admitted; predecessor ends; pane attaches to q1's successor (2000). */ + mockUseAgentQueuedTurns.mockReturnValue({ + data: [ + { queuedTurnId: 'q1', status: 'admitted', effectivePredecessorCreatedAt: 1000 }, + { queuedTurnId: 'q2', status: 'queued', expectedPredecessorCreatedAt: 1000 }, + ], + dataUpdatedAt: 3, + }); + await act(async () => { + setIsSubmitting(false); + setAttachedGenerationCreatedAt(null); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + await act(async () => { + setIsSubmitting(true); + setAttachedGenerationCreatedAt(2000); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + + /** q1's receipt dropped; q2 still queued; successor 2000 finishes. The + * pane must still owe q2 — the whole point of per-turn tracking. */ + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'q2', status: 'queued', expectedPredecessorCreatedAt: 1000 }], + dataUpdatedAt: 4, + }); + await act(async () => { + setIsSubmitting(false); + setAttachedGenerationCreatedAt(null); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + + /** Pane re-attaches to 2000 (still q1's successor, q2's predecessor); + * q2 is admitted while it is live, and its receipt still names only the + * chain's ROOT. The link learned while q2 waited must outrank that: the + * live 2000 is q2's predecessor, not its delivery. */ + await act(async () => { + setIsSubmitting(true); + setAttachedGenerationCreatedAt(2000); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'q2', status: 'admitted', expectedPredecessorCreatedAt: 1000 }], + dataUpdatedAt: 5, + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + await act(async () => { + setIsSubmitting(false); + setAttachedGenerationCreatedAt(null); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + + /** q2's own successor (3000) is the delivery. */ + await act(async () => { + setIsSubmitting(true); + setAttachedGenerationCreatedAt(3000); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + await act(async () => { + setIsSubmitting(false); + setAttachedGenerationCreatedAt(null); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, false); + }); + + it("keeps one conversation's owed state when another also owes a run", async () => { + jest.useFakeTimers(); + jest.setSystemTime(2_000_000); + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'a1', status: 'queued' }], + dataUpdatedAt: 1_999_000, + }); + const { rerender, queryClient } = renderUseResumeOnLoad({ + messages: [buildUserMessage(CONVERSATION_ID)], + }); + await act(async () => { + await Promise.resolve(); + }); + + /** Navigate to B, which reports its own queued turn. A's latch must + * survive, and A's expiry must still fire while B is on screen. */ + const invalidate = jest.spyOn(queryClient, 'invalidateQueries').mockResolvedValue(undefined); + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'b1', status: 'queued' }], + dataUpdatedAt: 2_000_500, + }); + rerender({ conversationId: STALE_CONVERSATION_ID }); + await act(async () => { + await Promise.resolve(); + }); + await act(async () => { + jest.advanceTimersByTime(30_000 + 1); + }); + + expect(invalidate).toHaveBeenCalledWith({ + queryKey: [QueryKeys.messages, CONVERSATION_ID], + }); + /** B's own owed state is untouched by A's expiry. */ + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + jest.useRealTimers(); + }); + + it('does not treat a live generation as delivery when no boundary is known', async () => { + mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); + /** + * Latched while unattached, from a receipt that carries no boundary, then + * a generation comes live. It could be the predecessor as easily as the + * successor; guessing "delivered" would silence the poll while the run is + * still owed. Stay owed and let the bounded expiry be the fallback. + */ + mockUseAgentQueuedTurns.mockReturnValue({ + data: [{ queuedTurnId: 'q1', status: 'admitted' }], + dataUpdatedAt: 2, + }); + const { rerender, setIsSubmitting, setAttachedGenerationCreatedAt } = renderUseResumeOnLoad({ + submission: buildSubmission(CONVERSATION_ID), + isSubmitting: false, + attachedGenerationCreatedAt: null, + messages: [buildUserMessage(CONVERSATION_ID)], + }); + await act(async () => { + await Promise.resolve(); + }); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + + await act(async () => { + setIsSubmitting(true); + setAttachedGenerationCreatedAt(1000); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + await act(async () => { + setIsSubmitting(false); + setAttachedGenerationCreatedAt(null); + }); + rerender(); + await act(async () => { + await Promise.resolve(); + }); + + expect(mockExtendActiveJobsGrace).not.toHaveBeenCalled(); + expect(mockUseActiveJobs).toHaveBeenLastCalledWith(true, true); + }); + it('does not re-arm for a conversation that is not the one being viewed', async () => { const observedSubmissions: Array = []; mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS); diff --git a/client/src/hooks/SSE/useResumeOnLoad.ts b/client/src/hooks/SSE/useResumeOnLoad.ts index 904833e9d8..06b7e6c2e7 100644 --- a/client/src/hooks/SSE/useResumeOnLoad.ts +++ b/client/src/hooks/SSE/useResumeOnLoad.ts @@ -1,4 +1,4 @@ -import { useEffect, useRef, useState } from 'react'; +import { useEffect, useMemo, useRef, useState } from 'react'; import { useQueryClient } from '@tanstack/react-query'; import { useSetRecoilState, useRecoilValue, useRecoilCallback } from 'recoil'; import { @@ -20,11 +20,19 @@ import { carriedSteerContext, getBranchSiblingIndexesForTarget, } from '~/utils'; +import { + useStreamStatus, + useActiveJobs, + useAgentQueuedTurns, + streamStatusQueryKey, + isQueuedTurnSuccessorOwed, + extendActiveJobsGrace, + ACTIVE_JOBS_SUCCESSOR_GRACE_MS, +} from '~/data-provider'; import { getGenerationProtocolVersion, supportsGenerationProtocolV2, } from '~/data-provider/SSE/protocol'; -import { useStreamStatus, useActiveJobs, streamStatusQueryKey } from '~/data-provider'; import useSteerConvert from '~/hooks/Chat/useSteerConvert'; import store from '~/store'; @@ -270,6 +278,9 @@ export default function useResumeOnLoad( * repeatable, and still one status read per interval at worst. */ const answeredActiveJobRef = useRef<{ conversationId: string; at: number } | null>(null); + /** Receipt state at the last arm, so history is refetched on a transition + * rather than on every heartbeat of an unchanged queued turn. */ + const lastArmedReceiptSignatureRef = useRef(null); /** * Bumped when an announcement is answered. Clearing `processedConvoRef` alone * cannot restart the check below: a ref mutation neither schedules a render @@ -467,12 +478,301 @@ export default function useResumeOnLoad( * and an effect keyed on it would run once and never again. The stamp moves * on every fetch, which is the heartbeat this needs. */ - const { data: activeJobsData, dataUpdatedAt: activeJobsUpdatedAt } = - useActiveJobs(resumableEnabled); + /** + * Positive evidence that the backend owes this conversation a run: a queued + * turn it has taken ownership of. The client never builds a submission for + * one — `useQueueDrain` declines the boundary — so the resulting run has to + * be recognised from the active-job list, and finishing the previous run is + * what empties that list. Saying so outright keeps the listening exact + * instead of betting on how long admission takes. + * + * Read from the server's own receipts rather than the chip projection, so + * admission semantics stay beside the queue's own predicate instead of being + * restated here. `useSteering` owns the fetch; subscribing with + * `enabled: false` observes that cache without competing for it, and an + * unpopulated cache simply falls back to the handover window. + */ + const { data: queuedTurnReceipts, dataUpdatedAt: queuedTurnsObservedAt } = useAgentQueuedTurns( + conversationId ?? '', + false, + ); + /** + * Receipts whose successor this pane has already attached to, per turn and + * per conversation. An `admitted` receipt stays in the cache for a fetch + * after delivery and still reads as owed; without remembering it was + * delivered, the latch would clear on the attachment and be re-recorded from + * the same receipt on the next render, synchronously, until the cache moved + * on. Tracked per turn rather than per receipt set: two turns queued behind + * the same generation deliver one at a time, and marking the whole set on + * the first would let the still-attached first successor pass as delivery of + * the second. Entries are pruned once the conversation's receipts no longer + * mention them. + */ + const [deliveredTurnIdsByConvo, setDeliveredTurnIdsByConvo] = useState< + ReadonlyMap> + >(() => new Map()); + /** + * What the receipts say, reduced to the two facts this hook keys on. The + * signature drives re-arming: a `queued` receipt is refetched every two + * seconds during a long wait and advances `dataUpdatedAt` without changing, + * and re-arming on that heartbeat would invalidate status and history every + * throttle interval while nothing has happened. Only a transition — a status + * change, a receipt appearing or being dropped — is an announcement. + * + * The boundary is the generation the owed turn waits behind, which the + * receipt knows better than this pane does: `expectedPredecessorCreatedAt` + * is stamped from the live epoch at enqueue, and `effectivePredecessorCreatedAt` + * is the boundary admission actually consumed once turns chain. It is what + * lets delivery be judged by generation rather than by "some run exists". + */ + const receiptSignature = Array.isArray(queuedTurnReceipts) + ? queuedTurnReceipts + .map((receipt) => `${receipt.queuedTurnId}:${receipt.status}`) + .sort() + .join('|') + : ''; + const deliveredTurnIds = + conversationId != null ? deliveredTurnIdsByConvo.get(conversationId) : undefined; + const owedReceipts = useMemo( + () => + Array.isArray(queuedTurnReceipts) + ? queuedTurnReceipts.filter( + (receipt) => + !deliveredTurnIds?.has(receipt.queuedTurnId) && isQueuedTurnSuccessorOwed([receipt]), + ) + : [], + [queuedTurnReceipts, deliveredTurnIds], + ); + const successorOwedByReceipt = owedReceipts.length > 0; + /** A turn still `queued`/`claimed` has not produced a run: whatever generation + * is live right now is its predecessor — including a sibling turn's + * successor that this pane is attached to. The boundary it enqueued with + * names the root of the chain, not the link it now waits behind. */ + const owedReceiptStillWaiting = owedReceipts.some( + (receipt) => receipt.status === 'queued' || receipt.status === 'claimed', + ); + const receiptBoundary = owedReceipts.reduce((boundary, receipt) => { + const candidate = + receipt.effectivePredecessorCreatedAt ?? receipt.expectedPredecessorCreatedAt ?? null; + return candidate != null && (boundary == null || candidate > boundary) ? candidate : boundary; + }, null); + + /** + * The latch remembers the generation the owed turn waits behind, because + * the successor is usually announced while that predecessor is still live — + * attached here, or merely listed because another client owns it. Neither + * is delivery. The active-job list carries no generation identity at all, so + * "the conversation is listed" can never prove the successor exists; only + * attachment to a generation other than the predecessor's can. Judging + * delivery by anything weaker clears the latch in exactly the window it is + * for, and a successor that starts and finishes between two list polls is + * then never seen. + * + * `quietSince` is the absolute start of the expiry window. It begins when + * the receipt stops reporting the turn — or when its conversation leaves the + * screen, where no receipt is observed at all — and is never pushed back by + * a remount, so returning later cannot reopen a spent window. + */ + type OwedSuccessor = { predecessorCreatedAt: number | null; quietSince: number | null }; + /** Keyed by conversation: navigating to a conversation that also owes a run + * must not evict another's latch, or its expiry — and the history repair it + * performs — is lost. */ + const [owedSuccessors, setOwedSuccessors] = useState>( + () => new Map(), + ); + const owedSuccessor = + conversationId != null ? (owedSuccessors.get(conversationId) ?? null) : null; + const successorOwed = + !hasLiveSubmissionForThisConvo && (successorOwedByReceipt || owedSuccessor != null); + const { data: activeJobsData, dataUpdatedAt: activeJobsUpdatedAt } = useActiveJobs( + resumableEnabled, + successorOwed, + ); const hasActiveJobForThisConvo = !!conversationId && conversationId !== Constants.NEW_CONVO && activeJobsData?.activeJobIds?.includes(conversationId) === true; + /** + * The predecessor a turn actually waits behind is the LATEST link known: + * receipts can still name the chain's root long after a sibling's successor + * became the live link, and the latch may have learned that link while the + * turn was waiting. Generations are timestamps, so "latest" is the max of + * what the receipt says, what the latch learned, and — while the turn still + * waits — whatever is live right now. + */ + const knownPredecessorCreatedAt = [ + owedReceiptStillWaiting ? attachedGenerationCreatedAt : null, + receiptBoundary, + owedSuccessor?.predecessorCreatedAt ?? null, + ].reduce( + (latest, candidate) => + candidate != null && (latest == null || candidate > latest) ? candidate : latest, + null, + ); + /** Delivery needs a boundary to compare against. With none known, a live + * generation could as easily be the predecessor as the successor, and + * guessing "delivered" would silence the poll while the run is still owed; + * the expiry window is the bounded fallback for that ambiguity. */ + const attachedToSuccessor = + owedSuccessor != null && + attachedGenerationCreatedAt != null && + !owedReceiptStillWaiting && + knownPredecessorCreatedAt != null && + attachedGenerationCreatedAt !== knownPredecessorCreatedAt; + + useEffect(() => { + if (!conversationId) { + return; + } + if (Array.isArray(queuedTurnReceipts)) { + const stillMentioned = new Set(queuedTurnReceipts.map((receipt) => receipt.queuedTurnId)); + setDeliveredTurnIdsByConvo((current) => { + const delivered = current.get(conversationId); + if (!delivered) { + return current; + } + const kept = [...delivered].filter((turnId) => stillMentioned.has(turnId)); + if (kept.length === delivered.size) { + return current; + } + const next = new Map(current); + if (kept.length === 0) { + next.delete(conversationId); + } else { + next.set(conversationId, new Set(kept)); + } + return next; + }); + } + if (attachedToSuccessor) { + setDeliveredTurnIdsByConvo((current) => { + const next = new Map(current); + const merged = new Set(current.get(conversationId) ?? []); + for (const receipt of owedReceipts) { + merged.add(receipt.queuedTurnId); + } + next.set(conversationId, merged); + return next; + }); + /** + * The list may never have seen this run — it can start and finish + * between two polls — so its "recently active" clock is stale and would + * leave nothing polling once the latch clears. A live generation was + * just observed; restart the handover window from it so an unpredicted + * continuation after this run is still discovered. + */ + extendActiveJobsGrace(); + setOwedSuccessors((current) => { + if (!current.has(conversationId)) { + return current; + } + const next = new Map(current); + next.delete(conversationId); + return next; + }); + return; + } + if (!successorOwedByReceipt) { + return; + } + setOwedSuccessors((current) => { + const existing = current.get(conversationId); + if (existing == null) { + /** The receipt's boundary when it has one; otherwise whatever is live + * right now — a turn that is still owed cannot have started, so a live + * generation at the moment it is first seen is its predecessor. */ + const next = new Map(current); + next.set(conversationId, { + predecessorCreatedAt: knownPredecessorCreatedAt ?? attachedGenerationCreatedAt, + quietSince: null, + }); + return next; + } + /** Still reporting. A boundary learned later (admission consumed it, or a + * sibling's successor is now the live link) supersedes what was + * guessed. The window is reopened only by an observation newer than the + * moment it started: a remount first exposes the cached receipt from + * before the turn went quiet, and that stale data must not extend an + * expiry that is meant to be absolute. */ + const refined = knownPredecessorCreatedAt ?? existing.predecessorCreatedAt; + const reopen = existing.quietSince != null && queuedTurnsObservedAt > existing.quietSince; + if (!reopen && refined === existing.predecessorCreatedAt) { + return current; + } + const next = new Map(current); + next.set(conversationId, { + predecessorCreatedAt: refined, + quietSince: reopen ? null : existing.quietSince, + }); + return next; + }); + }, [ + conversationId, + queuedTurnReceipts, + owedReceipts, + successorOwedByReceipt, + attachedToSuccessor, + knownPredecessorCreatedAt, + attachedGenerationCreatedAt, + queuedTurnsObservedAt, + ]); + + /** Only a server observation newer than the window's start counts as the + * turn still reporting; a remount's cached receipt predates it. Reporting + * is observable only for the conversation on screen. */ + const owedIsReporting = + owedSuccessor != null && + successorOwedByReceipt && + (owedSuccessor.quietSince == null || queuedTurnsObservedAt > owedSuccessor.quietSince); + useEffect(() => { + const now = Date.now(); + const timers: ReturnType[] = []; + let needsQuietStamp = false; + for (const [owedConversationId, latch] of owedSuccessors) { + if (owedConversationId === conversationId && owedIsReporting) { + continue; + } + if (latch.quietSince == null) { + needsQuietStamp = true; + continue; + } + const remaining = Math.max(0, ACTIVE_JOBS_SUCCESSOR_GRACE_MS - (now - latch.quietSince)); + timers.push( + setTimeout(() => { + setOwedSuccessors((current) => { + if (!current.has(owedConversationId)) { + return current; + } + const next = new Map(current); + next.delete(owedConversationId); + return next; + }); + /** The window closed without the successor being seen. If it ran + * and finished inside a poll gap, its turns are on the server and + * nowhere else; one refetch is the whole repair, and marking an + * off-screen conversation stale costs nothing until it is opened. */ + queryClient.invalidateQueries({ queryKey: [QueryKeys.messages, owedConversationId] }); + }, remaining), + ); + } + if (needsQuietStamp) { + setOwedSuccessors((current) => { + const next = new Map(current); + for (const [owedConversationId, latch] of current) { + const reporting = owedConversationId === conversationId && owedIsReporting; + if (!reporting && latch.quietSince == null) { + next.set(owedConversationId, { ...latch, quietSince: now }); + } + } + return next; + }); + } + return () => { + for (const timer of timers) { + clearTimeout(timer); + } + }; + }, [owedSuccessors, owedIsReporting, conversationId, queryClient]); const shouldCheck = resumableEnabled && @@ -761,16 +1061,20 @@ export default function useResumeOnLoad( * ever collect those turns. The refetch also re-gates `messagesLoaded`, which * defers the effect above until the authoritative history has landed. */ + /** + * Two announcement sources, one arm. The active-job list is the general one. + * An owed queued turn is the specific one, and it must be able to arm on its + * own: the run it announces can start and finish between two list polls, and + * the status read this arm enables is the authoritative answer either way — + * active attaches, inactive means the history refetch below is the repair. + */ + const successorAnnounced = hasActiveJobForThisConvo || successorOwed; useEffect(() => { - if ( - !resumableEnabled || - !hasActiveJobForThisConvo || - !conversationId || - hasLiveSubmissionForThisConvo - ) { - if (!hasActiveJobForThisConvo) { - answeredActiveJobRef.current = null; - } + if (!successorAnnounced) { + answeredActiveJobRef.current = null; + return; + } + if (!resumableEnabled || !conversationId || hasLiveSubmissionForThisConvo) { return; } @@ -798,17 +1102,31 @@ export default function useResumeOnLoad( setSubmission(null); } queryClient.invalidateQueries({ queryKey: streamStatusQueryKey(conversationId) }); - queryClient.invalidateQueries({ queryKey: [QueryKeys.messages, conversationId] }); + /** + * History is refetched only when this arm can actually change what the + * pane shows: a run is listed (attachment is imminent and must not be + * grafted onto a stale snapshot), or the receipts just transitioned (the + * turn was admitted, or dropped after finishing inside a poll gap). A + * long wait behind an unadmitted turn re-arms status on the heartbeat but + * must not download the conversation again each time. + */ + const receiptsTransitioned = lastArmedReceiptSignatureRef.current !== receiptSignature; + lastArmedReceiptSignatureRef.current = receiptSignature; + if (hasActiveJobForThisConvo || receiptsTransitioned) { + queryClient.invalidateQueries({ queryKey: [QueryKeys.messages, conversationId] }); + } processedConvoRef.current = null; setExternalRunArm((arm) => arm + 1); }, [ conversationId, resumableEnabled, + successorAnnounced, hasActiveJobForThisConvo, hasActiveSubmissionForThisConvo, hasLiveSubmissionForThisConvo, attachedGenerationCreatedAt, activeJobsUpdatedAt, + receiptSignature, setSubmission, queryClient, ]);