CERBERUS/LibreChat
1
0
Fork 0
mirror of https://github.com/danny-avila/LibreChat.git synced 2026-10-23 10:36:09 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 12

🐝 fix: Keep Polling for Owed Agent Successors (#15537)

Browse source 
* 🎯 fix: Ask Because a Successor Is Owed, Not Because One Might Be

The handover window guessed at how long admission takes. For a queued turn it
never had to: the backend reports the turn as `queued` or `claimed`, which is
positive evidence that a run is coming and that this client will not be the one
to start it. Declare that outright and keep the active-job list live for
exactly as long as it holds.

Read from the server's receipts rather than the chip projection, so admission
semantics stay in one place — `shouldPollAgentQueuedTurns`, shared with the
queue itself — instead of being restated here and drifting. `useSteering` owns
that fetch; subscribing with `enabled: false` observes its cache without
competing for it, and an unpopulated cache falls back to the window. It also
keeps this off Recoil, which the client is migrating away from.

The window stays for successors nothing local predicts — a background-tool
continuation has no receipt to report itself — and is now documented as the
fallback rather than the mechanism.

* 🔗 fix: Carry the Owed Successor Across the Receipt Handover

Codex found the gap in the deterministic path: the receipt that announces a
queued turn reports `admitted` for one fetch at most and is then dropped from
the projection — before the active-job list has necessarily observed the run
it started. If admission outlasted the handover window, the wall-clock
fallback was already spent, and a pane keyed on the receipt alone stopped
listening in exactly the window that mattered.

Two structural changes, so this is not another edge patched.

Owed is held, not read. Once a successor is known to be owed it stays owed
until it is delivered — this pane attaches, or the list names the conversation.
A bounded expiry backstops the case where neither ever happens, and it starts
only once the receipt has gone quiet, so a long wait behind an unadmitted turn
does not burn the window before the handover begins. `admitted` now counts as
owed: the receipt poll stops there because nothing is left to wait for, but an
unattached pane is in the opposite position, and it is the strongest evidence
that pane will get.

The receipt arms on its own. The arm had one announcement source, the list,
which can miss a run that starts and finishes between two polls. An owed turn
is a second source, and the status read the arm enables is the authoritative
answer either way: active attaches, inactive means the history refetch the arm
already performs is the repair. Neither source has to wait on the other.

* 🧹 chore: Sort Imports in useResumeOnLoad

* 🧭 fix: Deliver an Owed Successor Only When the Successor Itself Is Seen

Codex: the latch treated the pane's own live attachment as delivery of the
owed successor. But the successor is usually announced while its predecessor is
still attached — the turn was queued behind it — so the latch was cleared, or
never recorded, in exactly the window it exists for. Once that predecessor
closed and the receipt was dropped, a successor that started and finished
between two list polls was never seen.

Remember which generation was live when the turn became owed. Delivery is then
attachment to a different generation, or the list naming the conversation while
nothing of ours is live. The predecessor no longer counts.

Each half is proved against its own defect: reverting to the old predicate
fails the predecessor-attached case; dropping the generation-aware attach fails
the different-epoch case; and forgetting which generation was the predecessor
livelocks — the latch is set, "delivered" by the very generation it should have
excluded, cleared, and set again.

* 🧷 fix: Judge Successor Delivery by Generation, Re-Arm on Transitions, Expire Off-Screen

Three codex findings on the owed-successor latch, each a way the latch could
be cleared, never recorded, or held too long.

Delivery by generation only. The active-job list carries no generation
identity, so "the conversation is listed" was satisfying delivery while the
listed run was the predecessor — running elsewhere, unattached here — and the
latch was never recorded. The receipt knows the predecessor better than this
pane does: `expectedPredecessorCreatedAt` is stamped from the live epoch at
enqueue and `effectivePredecessorCreatedAt` is the boundary admission actually
consumed. Delivery is now attachment to a generation other than that boundary;
a turn still owed cannot have started, so a generation live when it is first
seen is its predecessor and refines the boundary rather than delivering.

Re-arm on transitions, not heartbeats. A `queued` receipt is refetched every
two seconds during a long wait and advances `dataUpdatedAt` without changing,
and the arm was keyed on that stamp — after the throttle it invalidated status
and history again and again while nothing had happened. It now keys on a
signature of receipt ids and statuses, and refetches history only when a run
is listed or the receipts actually transitioned.

Expire on an absolute window, off-screen too. The expiry effect skipped when
the latched conversation was not the one on screen, so navigating away held
the latch indefinitely and returning later reopened a fresh window. The window
now starts from `quietSince` — set when the receipt stops reporting or the
conversation leaves the screen — and is never pushed back by a remount.

Each change is proved against its own defect: re-adding list-based delivery
fails the listed-predecessor case; removing the transition gate fails the
heartbeat case; skipping expiry off-screen fails the absolute-window case; and
forgetting to learn a live predecessor livelocks the latch.

* 🧽 test: Restore the Admin Allowlist Override This Spec Leaks Into the Shard

The spec writes a per-user `mcpSettings.allowedDomains` override for the shared
primary user and never removes it. The list holds only its own fixture's
origin, and allowlist matching is protocol- and port-inclusive, so for the rest
of the shard every other MCP fixture is blocked: `e2e-oauth` fails inspection,
and an agent expecting its tools initializes into
AGENT_EXPECTED_MCP_TOOLS_UNAVAILABLE (503) instead of producing its OAuth
prompt.

Whether that bites depends only on which specs the shard runs afterwards, so it
surfaces as a deterministic failure in an unrelated spec — `mcp-oauth-resume`
failing an assertion that fires before any resume happens — which passes
whenever it runs first. A client-only PR whose path selection reordered the
shard hit it three times across two heads.

Delete the override in a `finally`, and confirm against the stored config
rather than by re-running `reinitialize`: the connection this test established
stays live, so reinitialize keeps succeeding once the allowlist is gone.

* 🩹 test: Read the Config Wrapper and Treat 404 as Cleared in the Override Teardown

`GET /api/admin/config/:principalType/:principalId` returns `{ config }`, and
deleting the last override removes the document, so the follow-up read is a
404 — the cleared state, not a read failure. The poll treated both as
unreadable and timed out.

* 🧷 fix: Remember Delivered Receipts, Restart the Grace on Delivery, Gate Reopening on Newer Data

Three more codex findings on the owed-successor latch.

Delivered receipts are remembered. An `admitted` receipt stays in the cache for
a fetch after the successor attaches, and it still counts as owed — so the
latch cleared on attachment and was re-recorded from the same receipt on the
next render, synchronously, until the cache moved on. That is React's
maximum-update-depth failure, reachable in the ordinary queued-turn handoff.
The receipt signature that delivered is recorded, and a receipt set with that
signature neither latches nor counts as owing a run; a genuinely new turn can.

Delivery restarts the fallback grace. The list may never have seen the run a
queued turn produced — it can start and finish between two polls — so its own
"recently active" clock is stale, and clearing the latch would have removed
the only reason left to poll. An unpredicted continuation after that run (a
background-tool dispatch finishing) would then go unnoticed until a focus or
reload. Observing the successor live now restarts the handover window.

The window reopens only on newer evidence. Returning to a conversation first
exposes the receipt cached before the turn went quiet, while the authoritative
refetch is still in flight; treating that as "still reporting" reset the
start of an expiry meant to be absolute, so repeated navigation could extend
it indefinitely. Only a server observation newer than the window's start
counts, for reopening and for the reporting check alike.

Each is proved against its defect: forgetting delivered receipts livelocks the
handoff test; skipping the grace restart fails its assertion; reopening on any
report fails the stale-remount case.

* 🧷 fix: Track Delivery Per Turn, Ratchet the Boundary, Keep One Latch Per Conversation

Three more codex findings on the owed-successor latch, plus a bug the second
one's regression test exposed.

Delivery is tracked per queued turn. Two turns queued behind the same
generation deliver one at a time; marking the whole receipt set on the first
would let the still-attached first successor pass as delivery of the second.
While any owed turn is still waiting, nothing is delivered at all — a waiting
turn cannot have produced the live run, so that run is its predecessor.

The boundary ratchets forward. A receipt can name the chain's root long after
a sibling's successor became the link the turn actually waits behind, and the
latch learns that link while the turn waits. Taking the receipt's boundary
first would have "delivered" the second turn on its own predecessor. The
predecessor is the latest link known: the max of what the receipt says, what
the latch learned, and, while waiting, what is live. With no boundary known at
all, a live generation is not delivery; the bounded expiry is the fallback.

Latches are keyed by conversation. Navigating to a conversation that also owes
a run replaced the previous one's latch, cancelling its expiry and the history
repair it performs. Every latch keeps its own absolute window now, and expiry
runs for all of them regardless of which conversation is on screen.

The cherry-picked allowlist-override teardown is synced to its final form
(see #15543): cleanup always runs, only definitive answers end its poll, and
`reinitialize` is no longer used as the reverted signal.

Proofs: marking the whole receipt set is indistinguishable only because the
waiting rule already blocks it; forgetting to learn the live link, or letting
the receipt's root outrank it, both fail the two-turn case; a single-slot latch
fails the two-conversation case; delivering without a boundary fails its case.
This commit is contained in:
Danny Avila 2026-09-03 13:23:19 -04:00 • committed by GitHub
parent 5f187d3a6c
commit f0dfeed2b0
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 1269 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

31
client/src/data-provider/SSE/__tests__/queries.test.ts
View file

@ -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);

30
client/src/data-provider/SSE/__tests__/queuedTurns.test.ts
View file

@ -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));

52
client/src/data-provider/SSE/queries.ts
View file

@ -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,
});
}

22
client/src/data-provider/SSE/queuedTurns.ts
View file

@ -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,

817
client/src/hooks/SSE/__tests__/useResumeOnLoad.spec.tsx
View file

@ -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<TSubmission | null> = [];
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<TSubmission | null> = [];
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<TSubmission | null> = [];
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<TSubmission | null> = [];
mockUseStreamStatus.mockReturnValue(INACTIVE_STATUS);

346
client/src/hooks/SSE/useResumeOnLoad.ts
View file

@ -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<string | null>(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<string, ReadonlySet<string>>
>(() => 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<number | null>((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<ReadonlyMap<string, OwedSuccessor>>(
() => 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<number | null>(
(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<typeof setTimeout>[] = [];
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,
]);