mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-10-22 01:53:33 +00:00
🧷 fix: Fence Server-Admitted Queue Boundaries (#15393)
* fix: fence server-admitted queue boundaries * fix: preserve effective queue admission boundaries * fix: fence queue recovery predecessor * fix: unify queued boundary reconciliation * fix: make queued turn receipt settlement monotonic * fix: narrow queued receipt projection status * fix: close queued turn settlement races * fix: make queued admission lineage clock independent * fix: centralize queued turn admission ownership * test: type queued admission outcomes * fix: make queued admission identity durable * fix: carry queued admission identity end to end * style: sort queued turn test imports * fix: unify indeterminate admission ownership * fix: migrate queued admission fence state * fix: migrate every queued admission owner * fix: surface queued admission reconciliation * fix: preserve queued receipt evidence * test: type queued receipt regression * fix: stabilize queued receipt authority * fix: supersede initial receipt fetches
This commit is contained in:
Danny Avila
GitHub
parent
f20b978075
commit
f0f82ec5fa
21 changed files with 3408 additions and 262 deletions
|
|
@ -103,8 +103,18 @@ function QueuedRow({
|
|||
const quoteCount = message.quotes?.length ?? 0;
|
||||
const isRecovered = message.recoverySteerId != null;
|
||||
const isRejected = message.server?.status === 'rejected';
|
||||
const isIndeterminate = message.server?.status === 'indeterminate';
|
||||
const isUnconfirmed =
|
||||
message.server?.status === 'uncertain' && message.server.reconciliationExpired === true;
|
||||
let statusLabel:
|
||||
| 'com_ui_queued_turn_reconciliation_required'
|
||||
| 'com_ui_steer_delivery_unconfirmed'
|
||||
| 'com_ui_queued_turn_failed' = 'com_ui_queued_turn_failed';
|
||||
if (isIndeterminate) {
|
||||
statusLabel = 'com_ui_queued_turn_reconciliation_required';
|
||||
} else if (isUnconfirmed) {
|
||||
statusLabel = 'com_ui_steer_delivery_unconfirmed';
|
||||
}
|
||||
const requiresDiscard = isRecovered || message.server?.id != null;
|
||||
const serverActionable =
|
||||
message.server == null ||
|
||||
|
|
@ -198,7 +208,7 @@ function QueuedRow({
|
|||
|
||||
return (
|
||||
<div role="listitem" className={ROW_CLASS} data-testid="queued-message-row">
|
||||
{isRejected || isUnconfirmed ? (
|
||||
{isRejected || isUnconfirmed || isIndeterminate ? (
|
||||
<TriangleAlert className="h-4 w-4 shrink-0 text-text-warning" aria-hidden="true" />
|
||||
) : (
|
||||
<Clock className="h-4 w-4 shrink-0 text-cyan-500" aria-hidden="true" />
|
||||
|
|
@ -216,12 +226,8 @@ function QueuedRow({
|
|||
0: String(fileCount),
|
||||
})}
|
||||
/>
|
||||
{(isRejected || isUnconfirmed) && (
|
||||
<span className="shrink-0 text-xs text-text-warning">
|
||||
{localize(
|
||||
isUnconfirmed ? 'com_ui_steer_delivery_unconfirmed' : 'com_ui_queued_turn_failed',
|
||||
)}
|
||||
</span>
|
||||
{(isRejected || isUnconfirmed || isIndeterminate) && (
|
||||
<span className="shrink-0 text-xs text-text-warning">{localize(statusLabel)}</span>
|
||||
)}
|
||||
{showPrimary && (
|
||||
<button
|
||||
|
|
|
|||
|
|
@ -181,6 +181,25 @@ describe('PendingSteerChips — terminal server state', () => {
|
|||
expect(screen.getByText('com_ui_queued_turn_failed')).toBeInTheDocument();
|
||||
});
|
||||
|
||||
it('shows indeterminate admission as non-actionable reconciliation work', () => {
|
||||
renderChips([
|
||||
{
|
||||
id: 'indeterminate-turn',
|
||||
text: 'external result unknown',
|
||||
createdAt: 1,
|
||||
server: {
|
||||
id: 'server-indeterminate-1',
|
||||
status: 'indeterminate',
|
||||
errorCode: 'ADMISSION_INDETERMINATE',
|
||||
},
|
||||
},
|
||||
]);
|
||||
|
||||
expect(screen.getByText('com_ui_queued_turn_reconciliation_required')).toBeInTheDocument();
|
||||
expect(screen.getByLabelText('com_ui_remove_queued')).toBeDisabled();
|
||||
expect(screen.queryByText('com_ui_send_now')).toBeNull();
|
||||
});
|
||||
|
||||
it('dismisses an expired ambiguous receipt without restoring or resending it', () => {
|
||||
renderChips([
|
||||
{
|
||||
|
|
|
|||
|
|
@ -1,6 +1,18 @@
|
|||
import { act, renderHook, waitFor } from '@testing-library/react';
|
||||
|
||||
const mockListAgentQueuedTurns = jest.fn();
|
||||
const mockEnqueueAgentQueuedTurn = jest.fn();
|
||||
const mockCancelAgentQueuedTurn = jest.fn();
|
||||
const mockRefetch = jest.fn();
|
||||
const mockCancelQueries = jest.fn().mockResolvedValue(undefined);
|
||||
const mockQueryClient = { cancelQueries: mockCancelQueries };
|
||||
const mockUseQuery = jest.fn((options: unknown) => ({ options, refetch: mockRefetch }));
|
||||
|
||||
jest.mock('@tanstack/react-query', () => ({
|
||||
useQuery: (options: unknown) => mockUseQuery(options),
|
||||
useMutation: jest.fn(),
|
||||
useQueryClient: () => mockQueryClient,
|
||||
}));
|
||||
|
||||
jest.mock('librechat-data-provider', () => ({
|
||||
QueryKeys: { agentQueuedTurns: 'agentQueuedTurns' },
|
||||
|
|
@ -21,6 +33,8 @@ import {
|
|||
fetchAgentQueuedTurns,
|
||||
isDefiniteQueuedTurnRejection,
|
||||
isDefiniteQueuedTurnsUnsupported,
|
||||
shouldPollAgentQueuedTurns,
|
||||
useAgentQueuedTurns,
|
||||
} from '../queuedTurns';
|
||||
|
||||
describe('Agent queued-turn data adapter', () => {
|
||||
|
|
@ -116,4 +130,87 @@ describe('Agent queued-turn data adapter', () => {
|
|||
expect(isDefiniteQueuedTurnRejection(error)).toBe(false);
|
||||
},
|
||||
);
|
||||
|
||||
it('stops background polling for a fail-closed indeterminate claim', () => {
|
||||
expect(
|
||||
shouldPollAgentQueuedTurns([
|
||||
{
|
||||
status: 'claimed',
|
||||
failure: { code: 'ADMISSION_INDETERMINATE' },
|
||||
},
|
||||
]),
|
||||
).toBe(false);
|
||||
expect(shouldPollAgentQueuedTurns([{ status: 'claimed' }])).toBe(true);
|
||||
expect(shouldPollAgentQueuedTurns([{ status: 'queued' }])).toBe(true);
|
||||
});
|
||||
|
||||
it('refreshes stopped reconciliation work on every mount and focus', () => {
|
||||
renderHook(() => useAgentQueuedTurns('conversation/one', true));
|
||||
|
||||
expect(mockUseQuery).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
refetchOnMount: 'always',
|
||||
refetchOnWindowFocus: 'always',
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('keeps one cache authority and refetches when known receipt ids change', async () => {
|
||||
const rendered = renderHook(
|
||||
({ ids }: { ids: string[] }) => useAgentQueuedTurns('conversation/one', true, ids),
|
||||
{ initialProps: { ids: ['request-one'] } },
|
||||
);
|
||||
|
||||
expect(mockUseQuery).toHaveBeenLastCalledWith(
|
||||
expect.objectContaining({
|
||||
queryKey: ['agentQueuedTurns', 'conversation/one'],
|
||||
}),
|
||||
);
|
||||
expect(mockRefetch).not.toHaveBeenCalled();
|
||||
|
||||
rendered.rerender({ ids: [] });
|
||||
|
||||
await waitFor(() =>
|
||||
expect(mockCancelQueries).toHaveBeenCalledWith({
|
||||
queryKey: ['agentQueuedTurns', 'conversation/one'],
|
||||
exact: true,
|
||||
}),
|
||||
);
|
||||
await waitFor(() =>
|
||||
expect(mockRefetch).toHaveBeenCalledWith({
|
||||
cancelRefetch: true,
|
||||
}),
|
||||
);
|
||||
expect(mockCancelQueries.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
mockRefetch.mock.invocationCallOrder[0],
|
||||
);
|
||||
expect(mockUseQuery).toHaveBeenLastCalledWith(
|
||||
expect.objectContaining({
|
||||
queryKey: ['agentQueuedTurns', 'conversation/one'],
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('lets only the newest identity projection refetch after cancellation', async () => {
|
||||
let releaseFirstCancellation: (() => void) | undefined;
|
||||
mockCancelQueries
|
||||
.mockImplementationOnce(
|
||||
() =>
|
||||
new Promise<void>((resolve) => {
|
||||
releaseFirstCancellation = resolve;
|
||||
}),
|
||||
)
|
||||
.mockResolvedValueOnce(undefined);
|
||||
const rendered = renderHook(
|
||||
({ ids }: { ids: string[] }) => useAgentQueuedTurns('conversation/one', true, ids),
|
||||
{ initialProps: { ids: ['request-one'] } },
|
||||
);
|
||||
|
||||
rendered.rerender({ ids: ['request-two'] });
|
||||
rendered.rerender({ ids: ['request-three'] });
|
||||
|
||||
await waitFor(() => expect(mockRefetch).toHaveBeenCalledTimes(1));
|
||||
await act(async () => releaseFirstCancellation?.());
|
||||
expect(mockRefetch).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import { useEffect, useRef } from 'react';
|
||||
import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
|
||||
import { dataService, MutationKeys, QueryKeys } from 'librechat-data-provider';
|
||||
import type {
|
||||
|
|
@ -8,10 +9,12 @@ import type {
|
|||
export type AgentQueuedTurnReceipt = TAgentQueuedTurnReceipt;
|
||||
export type EnqueueAgentQueuedTurnRequest = TEnqueueAgentQueuedTurnRequest;
|
||||
|
||||
export const agentQueuedTurnsQueryKey = (
|
||||
conversationId: string,
|
||||
clientRequestIds: readonly string[] = [],
|
||||
) => [QueryKeys.agentQueuedTurns, conversationId, ...clientRequestIds] as const;
|
||||
/** Keep one cache authority for a conversation. Known receipt ids change the
|
||||
* server projection, not the identity of the durable queue. Giving every id
|
||||
* set its own key can resurrect an older cached projection after an id is
|
||||
* retired. */
|
||||
export const agentQueuedTurnsQueryKey = (conversationId: string) =>
|
||||
[QueryKeys.agentQueuedTurns, conversationId] as const;
|
||||
|
||||
function queuedTurnErrorResponse(error: unknown): { status?: number; code?: string } {
|
||||
const response = (
|
||||
|
|
@ -72,29 +75,85 @@ export async function cancelAgentQueuedTurn(input: {
|
|||
return response.receipt;
|
||||
}
|
||||
|
||||
/** Indeterminate admission deliberately retains its durable claim until exact
|
||||
* source evidence arrives. It can still be refreshed on mount/focus, but a
|
||||
* two-second loop cannot resolve it and disguises permanent quarantine as
|
||||
* ordinary progress. */
|
||||
export function shouldPollAgentQueuedTurns(
|
||||
receipts: unknown,
|
||||
reconcileUntil?: number,
|
||||
observedAt = Date.now(),
|
||||
): boolean {
|
||||
return (
|
||||
(reconcileUntil != null && observedAt < reconcileUntil) ||
|
||||
(Array.isArray(receipts) &&
|
||||
receipts.some(
|
||||
(item: AgentQueuedTurnReceipt) =>
|
||||
item.status === 'queued' ||
|
||||
(item.status === 'claimed' && item.failure?.code !== 'ADMISSION_INDETERMINATE'),
|
||||
))
|
||||
);
|
||||
}
|
||||
|
||||
export function useAgentQueuedTurns(
|
||||
conversationId: string,
|
||||
enabled: boolean,
|
||||
clientRequestIds: string[] = [],
|
||||
reconcileUntil?: number,
|
||||
) {
|
||||
const queryClient = useQueryClient();
|
||||
const knownIds = [...new Set(clientRequestIds)].sort();
|
||||
return useQuery({
|
||||
queryKey: agentQueuedTurnsQueryKey(conversationId, knownIds),
|
||||
const knownIdsSignature = knownIds.join('\u0000');
|
||||
const previousRequest = useRef({ conversationId, knownIdsSignature });
|
||||
const query = useQuery({
|
||||
queryKey: agentQueuedTurnsQueryKey(conversationId),
|
||||
queryFn: () => fetchAgentQueuedTurns(conversationId, knownIds),
|
||||
enabled: enabled && conversationId.length > 0,
|
||||
staleTime: 1_000,
|
||||
refetchOnMount: true,
|
||||
refetchOnWindowFocus: true,
|
||||
refetchInterval: (receipts) => {
|
||||
return (reconcileUntil != null && Date.now() < reconcileUntil) ||
|
||||
(Array.isArray(receipts) &&
|
||||
receipts.some((item) => item.status === 'queued' || item.status === 'claimed'))
|
||||
? 2_000
|
||||
: false;
|
||||
},
|
||||
/** A stopped indeterminate row can receive exact proof at any time. Its
|
||||
* cache may still be inside `staleTime`, so mount/focus must bypass the
|
||||
* ordinary freshness gate rather than waiting for another user cycle. */
|
||||
refetchOnMount: 'always',
|
||||
refetchOnWindowFocus: 'always',
|
||||
refetchInterval: (receipts) =>
|
||||
shouldPollAgentQueuedTurns(receipts, reconcileUntil) ? 2_000 : false,
|
||||
retry: false,
|
||||
});
|
||||
const { refetch } = query;
|
||||
|
||||
useEffect(() => {
|
||||
const previous = previousRequest.current;
|
||||
previousRequest.current = { conversationId, knownIdsSignature };
|
||||
if (
|
||||
!enabled ||
|
||||
conversationId.length === 0 ||
|
||||
previous.conversationId !== conversationId ||
|
||||
previous.knownIdsSignature === knownIdsSignature
|
||||
) {
|
||||
return;
|
||||
}
|
||||
|
||||
let superseded = false;
|
||||
const refreshProjection = async () => {
|
||||
/** React Query cannot replace an initial fetch that has no cached data
|
||||
* through `refetch` alone. Cancel it explicitly so the next request
|
||||
* captures the current reconciliation identities. */
|
||||
await queryClient.cancelQueries({
|
||||
queryKey: agentQueuedTurnsQueryKey(conversationId),
|
||||
exact: true,
|
||||
});
|
||||
if (!superseded) {
|
||||
await refetch({ cancelRefetch: true });
|
||||
}
|
||||
};
|
||||
void refreshProjection();
|
||||
|
||||
return () => {
|
||||
superseded = true;
|
||||
};
|
||||
}, [conversationId, enabled, knownIdsSignature, queryClient, refetch]);
|
||||
|
||||
return query;
|
||||
}
|
||||
|
||||
export function useEnqueueAgentQueuedTurnMutation() {
|
||||
|
|
|
|||
|
|
@ -3,7 +3,12 @@ import { Constants } from 'librechat-data-provider';
|
|||
import { act, renderHook, waitFor } from '@testing-library/react';
|
||||
import { QueryClient, QueryClientProvider } from '@tanstack/react-query';
|
||||
import { RecoilRoot, useRecoilValue, useSetRecoilState, type MutableSnapshot } from 'recoil';
|
||||
import type { DrainAfterAbort, RunEnd, QueuedMessage } from '~/store/families';
|
||||
import type {
|
||||
DrainAfterAbort,
|
||||
RunEnd,
|
||||
QueuedMessage,
|
||||
SettledQueuedTurnReceipt,
|
||||
} from '~/store/families';
|
||||
import useQueueDrain from '../useQueueDrain';
|
||||
import store from '~/store';
|
||||
|
||||
|
|
@ -26,8 +31,12 @@ function setup(
|
|||
setIsSubmitting?: (value: boolean) => void;
|
||||
setQueue?: (value: QueuedMessage[]) => void;
|
||||
setNewConvoQueue?: (value: QueuedMessage[]) => void;
|
||||
setSettledReceipts?: (value: SettledQueuedTurnReceipt[]) => void;
|
||||
setInterruptFlag?: (value: DrainAfterAbort | false) => void;
|
||||
queueRef?: { current: QueuedMessage[] };
|
||||
queue?: QueuedMessage[];
|
||||
newConvoQueue?: QueuedMessage[];
|
||||
settledReceipts?: SettledQueuedTurnReceipt[];
|
||||
runEnd?: RunEnd | null;
|
||||
} = {};
|
||||
|
||||
function Harness() {
|
||||
|
|
@ -37,7 +46,14 @@ function setup(
|
|||
setters.setNewConvoQueue = useSetRecoilState(
|
||||
store.queuedMessagesByConvoId(Constants.NEW_CONVO),
|
||||
);
|
||||
setters.setSettledReceipts = useSetRecoilState(
|
||||
store.settledQueuedTurnReceiptsByConvoId(CONVO_ID),
|
||||
);
|
||||
setters.setInterruptFlag = useSetRecoilState(store.drainAfterAbortByIndex(INDEX));
|
||||
setters.queue = useRecoilValue(store.queuedMessagesByConvoId(CONVO_ID));
|
||||
setters.newConvoQueue = useRecoilValue(store.queuedMessagesByConvoId(Constants.NEW_CONVO));
|
||||
setters.settledReceipts = useRecoilValue(store.settledQueuedTurnReceiptsByConvoId(CONVO_ID));
|
||||
setters.runEnd = useRecoilValue(store.runEndByIndex(INDEX));
|
||||
useQueueDrain(INDEX, activeConversationId, ask);
|
||||
return null;
|
||||
}
|
||||
|
|
@ -150,6 +166,191 @@ describe('useQueueDrain', () => {
|
|||
expect(ask).toHaveBeenCalledWith({ text: 'legacy successor' }, emptyOverrides);
|
||||
});
|
||||
|
||||
it('reacts to a settled admission while another server row still owns the queue', async () => {
|
||||
const admittedServer: QueuedMessage = {
|
||||
...queuedMessage('q-server-1', 'admitted server-owned turn'),
|
||||
server: { id: 'server-queue-1', status: 'claimed', revision: 1 },
|
||||
};
|
||||
const remainingServer: QueuedMessage = {
|
||||
...queuedMessage('q-server-2', 'later server-owned turn'),
|
||||
server: { id: 'server-queue-2', status: 'queued', revision: 2 },
|
||||
};
|
||||
const { ask, setters } = setup(({ set }) => {
|
||||
set(store.queuedMessagesByConvoId(CONVO_ID), [admittedServer, remainingServer]);
|
||||
});
|
||||
|
||||
act(() => {
|
||||
setters.setRunEnd!(runEnd({ generationCreatedAt: 41 }));
|
||||
});
|
||||
await new Promise((resolve) => setTimeout(resolve, 25));
|
||||
expect(setters.runEnd).not.toBeNull();
|
||||
|
||||
act(() => {
|
||||
setters.setQueue!([remainingServer]);
|
||||
setters.setSettledReceipts!([
|
||||
{
|
||||
clientRequestId: 'admitted-request-1',
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 41,
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
await waitFor(() => expect(setters.runEnd).toBeNull());
|
||||
expect(ask).not.toHaveBeenCalled();
|
||||
expect(setters.settledReceipts).toEqual([]);
|
||||
expect(setters.queue).toEqual([remainingServer]);
|
||||
});
|
||||
|
||||
it('discards a predecessor boundary already consumed by server admission', async () => {
|
||||
const { ask, setters } = setup(({ set }) => {
|
||||
set(store.queuedMessagesByConvoId(CONVO_ID), [
|
||||
queuedMessage('q-local', 'must wait for the admitted run'),
|
||||
]);
|
||||
set(store.settledQueuedTurnReceiptsByConvoId(CONVO_ID), [
|
||||
{
|
||||
clientRequestId: 'admission-1',
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 41,
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
act(() => {
|
||||
setters.setRunEnd!(runEnd({ generationCreatedAt: 41 }));
|
||||
});
|
||||
|
||||
await waitFor(() => expect(setters.runEnd).toBeNull());
|
||||
expect(ask).not.toHaveBeenCalled();
|
||||
expect(setters.settledReceipts).toEqual([]);
|
||||
});
|
||||
|
||||
it('consumes one admission when separate receipts share the same predecessor epoch', async () => {
|
||||
const { ask, setters } = setup(({ set }) => {
|
||||
set(store.queuedMessagesByConvoId(CONVO_ID), [
|
||||
queuedMessage('q-local', 'wait for both admitted runs'),
|
||||
]);
|
||||
set(store.settledQueuedTurnReceiptsByConvoId(CONVO_ID), [
|
||||
{
|
||||
clientRequestId: 'admission-1',
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 41,
|
||||
},
|
||||
{
|
||||
clientRequestId: 'admission-2',
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 41,
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
act(() => {
|
||||
setters.setRunEnd!(runEnd({ generationCreatedAt: 41 }));
|
||||
});
|
||||
|
||||
await waitFor(() =>
|
||||
expect(setters.settledReceipts).toEqual([
|
||||
{
|
||||
clientRequestId: 'admission-2',
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 41,
|
||||
},
|
||||
]),
|
||||
);
|
||||
expect(ask).not.toHaveBeenCalled();
|
||||
|
||||
act(() => {
|
||||
setters.setRunEnd!(runEnd({ generationCreatedAt: 41, endedAt: Date.now() + 1 }));
|
||||
});
|
||||
|
||||
await waitFor(() => expect(setters.settledReceipts).toEqual([]));
|
||||
expect(ask).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('migrates the NEW_CONVO queue before discarding a consumed predecessor boundary', async () => {
|
||||
const queuedBeforeResolution = queuedMessage('q-new', 'wait for the admitted successor');
|
||||
const queuedAfterResolution = queuedMessage('q-resolved', 'still ordered after migration');
|
||||
const { ask, setters } = setup(({ set }) => {
|
||||
set(store.queuedMessagesByConvoId(Constants.NEW_CONVO), [queuedBeforeResolution]);
|
||||
set(store.queuedMessagesByConvoId(CONVO_ID), [queuedAfterResolution]);
|
||||
set(store.settledQueuedTurnReceiptsByConvoId(CONVO_ID), [
|
||||
{
|
||||
clientRequestId: 'admission-1',
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 41,
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
act(() => {
|
||||
setters.setRunEnd!(
|
||||
runEnd({
|
||||
generationCreatedAt: 41,
|
||||
startedAsNewConvo: true,
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
await waitFor(() => expect(setters.runEnd).toBeNull());
|
||||
expect(ask).not.toHaveBeenCalled();
|
||||
expect(setters.newConvoQueue).toEqual([]);
|
||||
expect(setters.queue).toEqual([queuedBeforeResolution, queuedAfterResolution]);
|
||||
});
|
||||
|
||||
it('lets the admitted successor terminal boundary release the next turn', async () => {
|
||||
const { ask, setters } = setup(({ set }) => {
|
||||
set(store.queuedMessagesByConvoId(CONVO_ID), [
|
||||
queuedMessage('q-local', 'send after the admitted run'),
|
||||
]);
|
||||
set(store.settledQueuedTurnReceiptsByConvoId(CONVO_ID), [
|
||||
{
|
||||
clientRequestId: 'admission-1',
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 41,
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
act(() => {
|
||||
setters.setRunEnd!(runEnd({ generationCreatedAt: 42 }));
|
||||
});
|
||||
|
||||
await waitFor(() => expect(ask).toHaveBeenCalledTimes(1));
|
||||
expect(ask).toHaveBeenCalledWith({ text: 'send after the admitted run' }, emptyOverrides);
|
||||
});
|
||||
|
||||
it('does not interpret a consumed predecessor as a timestamp range', async () => {
|
||||
const { ask, setters } = setup(({ set }) => {
|
||||
set(store.queuedMessagesByConvoId(CONVO_ID), [
|
||||
queuedMessage('q-local', 'send after the lower-clock successor'),
|
||||
]);
|
||||
set(store.settledQueuedTurnReceiptsByConvoId(CONVO_ID), [
|
||||
{
|
||||
clientRequestId: 'admission-1',
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 42,
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
act(() => {
|
||||
setters.setRunEnd!(runEnd({ generationCreatedAt: 41 }));
|
||||
});
|
||||
|
||||
await waitFor(() => expect(ask).toHaveBeenCalledTimes(1));
|
||||
expect(ask).toHaveBeenCalledWith(
|
||||
{ text: 'send after the lower-clock successor' },
|
||||
emptyOverrides,
|
||||
);
|
||||
expect(setters.settledReceipts).toEqual([
|
||||
{
|
||||
clientRequestId: 'admission-1',
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 42,
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('parks a mismatched signal instead of draining into the wrong conversation', async () => {
|
||||
const { ask, setters } = setup(({ set }) => {
|
||||
set(store.queuedMessagesByConvoId(CONVO_ID), [queuedMessage('q1', 'stay put')]);
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import { RecoilRoot, useRecoilValue, useSetRecoilState, type MutableSnapshot } f
|
|||
import { Constants, ContentTypes, EModelEndpoint, LocalStorageKeys } from 'librechat-data-provider';
|
||||
import type { TConversation, TMessage } from 'librechat-data-provider';
|
||||
import type { QueuedMessage } from '~/store/families';
|
||||
import useQueueDrain from '../useQueueDrain';
|
||||
import useSteering from '../useSteering';
|
||||
import store from '~/store';
|
||||
|
||||
|
|
@ -260,6 +261,8 @@ describe('useSteering', () => {
|
|||
stopGenerating: jest.fn(),
|
||||
}),
|
||||
queue: useQueue(CONVO_ID),
|
||||
settledReceipts: useRecoilValue(store.settledQueuedTurnReceiptsByConvoId(CONVO_ID)),
|
||||
pendingEnqueueIds: useRecoilValue(store.pendingQueuedTurnEnqueueIdsByConvoId(CONVO_ID)),
|
||||
}),
|
||||
{ wrapper },
|
||||
);
|
||||
|
|
@ -335,6 +338,377 @@ describe('useSteering', () => {
|
|||
);
|
||||
});
|
||||
|
||||
it('applies an admitted POST replay to the consumed predecessor set', async () => {
|
||||
mockEnqueueQueuedTurn.mockImplementation((input, options) => {
|
||||
options.onSuccess({
|
||||
...input,
|
||||
queuedTurnId: 'server-admitted-replay',
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 41,
|
||||
revision: 1,
|
||||
createdAt: new Date(100).toISOString(),
|
||||
updatedAt: new Date(200).toISOString(),
|
||||
});
|
||||
});
|
||||
const { result } = setupServerQueue(
|
||||
withActiveGeneration(({ set }) => {
|
||||
set(store.queuedMessagesByConvoId(CONVO_ID), [
|
||||
{
|
||||
id: 'unrelated-server-row',
|
||||
clientRequestId: 'unrelated-request',
|
||||
text: 'keep me',
|
||||
createdAt: 50,
|
||||
server: { id: 'unrelated-server-id', status: 'queued', revision: 1 },
|
||||
},
|
||||
]);
|
||||
}),
|
||||
);
|
||||
|
||||
await act(async () => {
|
||||
result.current.steering.queueFromComposer('already admitted');
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
await waitFor(() =>
|
||||
expect(result.current.settledReceipts).toEqual([
|
||||
{
|
||||
clientRequestId: expect.stringMatching(UUID_V4_PATTERN),
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 41,
|
||||
},
|
||||
]),
|
||||
);
|
||||
expect(result.current.queue).toEqual([
|
||||
expect.objectContaining({ id: 'unrelated-server-row', text: 'keep me' }),
|
||||
]);
|
||||
});
|
||||
|
||||
it('keeps an old-server admitted receipt uncertain until its exact boundary is known', async () => {
|
||||
mockEnqueueQueuedTurn.mockImplementation((input, options) => {
|
||||
options.onSuccess({
|
||||
...input,
|
||||
queuedTurnId: 'legacy-admitted-replay',
|
||||
status: 'admitted',
|
||||
revision: 1,
|
||||
createdAt: new Date(100).toISOString(),
|
||||
updatedAt: new Date(200).toISOString(),
|
||||
});
|
||||
});
|
||||
const { result } = setupServerQueue();
|
||||
|
||||
await act(async () => {
|
||||
result.current.steering.queueFromComposer('await exact admission boundary');
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
await waitFor(() =>
|
||||
expect(result.current.queue).toEqual([
|
||||
expect.objectContaining({
|
||||
text: 'await exact admission boundary',
|
||||
clientRequestId: expect.stringMatching(UUID_V4_PATTERN),
|
||||
server: expect.objectContaining({
|
||||
id: 'legacy-admitted-replay',
|
||||
status: 'uncertain',
|
||||
}),
|
||||
}),
|
||||
]),
|
||||
);
|
||||
expect(result.current.settledReceipts).toEqual([]);
|
||||
expect(mockUseAgentQueuedTurns.mock.calls.at(-1)?.[2]).toEqual([
|
||||
result.current.queue[0].clientRequestId,
|
||||
]);
|
||||
});
|
||||
|
||||
it('does not let a stale queued POST erase incomplete admission evidence', async () => {
|
||||
let releaseQueuedPost = () => undefined;
|
||||
mockEnqueueQueuedTurn.mockImplementation((input, options) => {
|
||||
releaseQueuedPost = () =>
|
||||
options.onSuccess({
|
||||
...input,
|
||||
queuedTurnId: 'legacy-admitted-race',
|
||||
status: 'queued',
|
||||
revision: 1,
|
||||
createdAt: new Date(100).toISOString(),
|
||||
updatedAt: new Date(100).toISOString(),
|
||||
});
|
||||
});
|
||||
const rendered = setupServerQueue();
|
||||
|
||||
await act(async () => {
|
||||
rendered.result.current.steering.queueFromComposer('hold incomplete admission');
|
||||
await Promise.resolve();
|
||||
});
|
||||
await waitFor(() => expect(mockEnqueueQueuedTurn).toHaveBeenCalledTimes(1));
|
||||
const input = mockEnqueueQueuedTurn.mock.calls[0][0];
|
||||
mockServerQueuedTurns = [
|
||||
{
|
||||
...input,
|
||||
queuedTurnId: 'legacy-admitted-race',
|
||||
status: 'admitted',
|
||||
revision: 2,
|
||||
createdAt: new Date(100).toISOString(),
|
||||
updatedAt: new Date(200).toISOString(),
|
||||
},
|
||||
];
|
||||
rendered.rerender();
|
||||
|
||||
await waitFor(() =>
|
||||
expect(rendered.result.current.queue[0].server).toMatchObject({
|
||||
id: 'legacy-admitted-race',
|
||||
status: 'uncertain',
|
||||
}),
|
||||
);
|
||||
expect(rendered.result.current.settledReceipts).toEqual([
|
||||
{ clientRequestId: input.clientRequestId, status: 'admitted_pending_boundary' },
|
||||
]);
|
||||
|
||||
await act(async () => {
|
||||
releaseQueuedPost();
|
||||
await Promise.resolve();
|
||||
});
|
||||
expect(rendered.result.current.queue).toHaveLength(1);
|
||||
expect(rendered.result.current.queue[0].server).toMatchObject({
|
||||
id: 'legacy-admitted-race',
|
||||
status: 'uncertain',
|
||||
});
|
||||
expect(rendered.result.current.pendingEnqueueIds).toEqual([]);
|
||||
expect(rendered.result.current.settledReceipts).toEqual([]);
|
||||
});
|
||||
|
||||
it('does not let a stale queued POST response resurrect a terminal receipt', async () => {
|
||||
let releaseQueuedPost = () => undefined;
|
||||
mockEnqueueQueuedTurn.mockImplementation((input, options) => {
|
||||
releaseQueuedPost = () =>
|
||||
options.onSuccess({
|
||||
...input,
|
||||
queuedTurnId: 'server-stale-post',
|
||||
status: 'queued',
|
||||
revision: 1,
|
||||
createdAt: new Date(100).toISOString(),
|
||||
updatedAt: new Date(100).toISOString(),
|
||||
});
|
||||
});
|
||||
const rendered = setupServerQueue();
|
||||
|
||||
await act(async () => {
|
||||
rendered.result.current.steering.queueFromComposer('settle before POST returns');
|
||||
await Promise.resolve();
|
||||
});
|
||||
await waitFor(() => expect(mockEnqueueQueuedTurn).toHaveBeenCalledTimes(1));
|
||||
const input = mockEnqueueQueuedTurn.mock.calls[0][0];
|
||||
mockServerQueuedTurns = [
|
||||
{
|
||||
...input,
|
||||
queuedTurnId: 'server-terminal-get',
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 41,
|
||||
revision: 2,
|
||||
createdAt: new Date(100).toISOString(),
|
||||
updatedAt: new Date(200).toISOString(),
|
||||
},
|
||||
];
|
||||
rendered.rerender();
|
||||
|
||||
await waitFor(() => expect(rendered.result.current.queue).toEqual([]));
|
||||
await act(async () => {
|
||||
releaseQueuedPost();
|
||||
await Promise.resolve();
|
||||
});
|
||||
|
||||
expect(rendered.result.current.queue).toEqual([]);
|
||||
expect(rendered.result.current.settledReceipts).toEqual([
|
||||
{
|
||||
clientRequestId: input.clientRequestId,
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 41,
|
||||
},
|
||||
]);
|
||||
});
|
||||
|
||||
it('does not let a stale queued POST response resurrect a cancelled row', async () => {
|
||||
let releaseQueuedPost = () => undefined;
|
||||
mockEnqueueQueuedTurn.mockImplementation((input, options) => {
|
||||
releaseQueuedPost = () =>
|
||||
options.onSuccess({
|
||||
...input,
|
||||
queuedTurnId: 'server-cancel-race',
|
||||
status: 'queued',
|
||||
revision: 1,
|
||||
createdAt: new Date(100).toISOString(),
|
||||
updatedAt: new Date(100).toISOString(),
|
||||
});
|
||||
});
|
||||
const rendered = setupServerQueue();
|
||||
|
||||
await act(async () => {
|
||||
rendered.result.current.steering.queueFromComposer('cancel before POST returns');
|
||||
await Promise.resolve();
|
||||
});
|
||||
await waitFor(() => expect(mockEnqueueQueuedTurn).toHaveBeenCalledTimes(1));
|
||||
const input = mockEnqueueQueuedTurn.mock.calls[0][0];
|
||||
mockServerQueuedTurns = [
|
||||
{
|
||||
...input,
|
||||
queuedTurnId: 'server-cancel-race',
|
||||
status: 'queued',
|
||||
revision: 2,
|
||||
createdAt: new Date(100).toISOString(),
|
||||
updatedAt: new Date(200).toISOString(),
|
||||
},
|
||||
];
|
||||
rendered.rerender();
|
||||
await waitFor(() =>
|
||||
expect(rendered.result.current.queue[0].server?.id).toBe('server-cancel-race'),
|
||||
);
|
||||
mockCancelQueuedTurn.mockResolvedValueOnce({
|
||||
...input,
|
||||
queuedTurnId: 'server-cancel-race',
|
||||
status: 'cancelled',
|
||||
revision: 3,
|
||||
createdAt: new Date(100).toISOString(),
|
||||
updatedAt: new Date(300).toISOString(),
|
||||
});
|
||||
|
||||
await act(async () => {
|
||||
await rendered.result.current.steering.discardQueued(rendered.result.current.queue[0]);
|
||||
});
|
||||
expect(rendered.result.current.queue[0].server).toBeUndefined();
|
||||
expect(rendered.result.current.settledReceipts).toEqual([
|
||||
{ clientRequestId: input.clientRequestId, status: 'cancelled' },
|
||||
]);
|
||||
|
||||
await act(async () => {
|
||||
releaseQueuedPost();
|
||||
await Promise.resolve();
|
||||
});
|
||||
expect(rendered.result.current.queue).toEqual([
|
||||
expect.objectContaining({
|
||||
text: 'cancel before POST returns',
|
||||
clientRequestId: input.clientRequestId,
|
||||
}),
|
||||
]);
|
||||
expect(rendered.result.current.queue[0].server).toBeUndefined();
|
||||
expect(rendered.result.current.pendingEnqueueIds).toEqual([]);
|
||||
expect(rendered.result.current.settledReceipts).toEqual([]);
|
||||
});
|
||||
|
||||
it('does not let a stale queued POST response replace a definitive dead snapshot', async () => {
|
||||
let releaseQueuedPost = () => undefined;
|
||||
mockEnqueueQueuedTurn.mockImplementation((input, options) => {
|
||||
releaseQueuedPost = () =>
|
||||
options.onSuccess({
|
||||
...input,
|
||||
queuedTurnId: 'server-dead-race',
|
||||
status: 'queued',
|
||||
revision: 1,
|
||||
createdAt: new Date(100).toISOString(),
|
||||
updatedAt: new Date(100).toISOString(),
|
||||
});
|
||||
});
|
||||
const rendered = setupServerQueue();
|
||||
|
||||
await act(async () => {
|
||||
rendered.result.current.steering.queueFromComposer('fail before POST returns');
|
||||
await Promise.resolve();
|
||||
});
|
||||
await waitFor(() => expect(mockEnqueueQueuedTurn).toHaveBeenCalledTimes(1));
|
||||
const input = mockEnqueueQueuedTurn.mock.calls[0][0];
|
||||
mockServerQueuedTurns = [
|
||||
{
|
||||
...input,
|
||||
queuedTurnId: 'server-dead-race',
|
||||
status: 'dead',
|
||||
revision: 2,
|
||||
failure: { code: 'PREDECESSOR_ABORTED', message: 'The predecessor aborted' },
|
||||
createdAt: new Date(100).toISOString(),
|
||||
updatedAt: new Date(200).toISOString(),
|
||||
},
|
||||
];
|
||||
rendered.rerender();
|
||||
|
||||
await waitFor(() =>
|
||||
expect(rendered.result.current.queue[0].server).toMatchObject({
|
||||
id: 'server-dead-race',
|
||||
status: 'rejected',
|
||||
errorCode: 'PREDECESSOR_ABORTED',
|
||||
}),
|
||||
);
|
||||
expect(rendered.result.current.settledReceipts).toEqual([
|
||||
{ clientRequestId: input.clientRequestId, status: 'dead' },
|
||||
]);
|
||||
|
||||
await act(async () => {
|
||||
releaseQueuedPost();
|
||||
await Promise.resolve();
|
||||
});
|
||||
expect(rendered.result.current.queue).toHaveLength(1);
|
||||
expect(rendered.result.current.queue[0].server).toMatchObject({
|
||||
id: 'server-dead-race',
|
||||
status: 'rejected',
|
||||
errorCode: 'PREDECESSOR_ABORTED',
|
||||
});
|
||||
expect(rendered.result.current.pendingEnqueueIds).toEqual([]);
|
||||
expect(rendered.result.current.settledReceipts).toEqual([]);
|
||||
});
|
||||
|
||||
it('does not let a stale queued POST replace indeterminate GET evidence', async () => {
|
||||
let releaseQueuedPost = () => undefined;
|
||||
mockEnqueueQueuedTurn.mockImplementation((input, options) => {
|
||||
releaseQueuedPost = () =>
|
||||
options.onSuccess({
|
||||
...input,
|
||||
queuedTurnId: 'server-indeterminate-race',
|
||||
status: 'queued',
|
||||
revision: 1,
|
||||
createdAt: new Date(100).toISOString(),
|
||||
updatedAt: new Date(100).toISOString(),
|
||||
});
|
||||
});
|
||||
const rendered = setupServerQueue();
|
||||
|
||||
await act(async () => {
|
||||
rendered.result.current.steering.queueFromComposer('unknown before POST returns');
|
||||
await Promise.resolve();
|
||||
});
|
||||
await waitFor(() => expect(mockEnqueueQueuedTurn).toHaveBeenCalledTimes(1));
|
||||
const input = mockEnqueueQueuedTurn.mock.calls[0][0];
|
||||
mockServerQueuedTurns = [
|
||||
{
|
||||
...input,
|
||||
queuedTurnId: 'server-indeterminate-race',
|
||||
status: 'claimed',
|
||||
revision: 2,
|
||||
failure: { code: 'ADMISSION_INDETERMINATE' },
|
||||
createdAt: new Date(100).toISOString(),
|
||||
updatedAt: new Date(200).toISOString(),
|
||||
},
|
||||
];
|
||||
rendered.rerender();
|
||||
|
||||
await waitFor(() =>
|
||||
expect(rendered.result.current.queue[0].server).toMatchObject({
|
||||
id: 'server-indeterminate-race',
|
||||
status: 'indeterminate',
|
||||
revision: 2,
|
||||
}),
|
||||
);
|
||||
expect(rendered.result.current.settledReceipts).toEqual([
|
||||
{ clientRequestId: input.clientRequestId, status: 'indeterminate' },
|
||||
]);
|
||||
|
||||
await act(async () => {
|
||||
releaseQueuedPost();
|
||||
await Promise.resolve();
|
||||
});
|
||||
expect(rendered.result.current.queue[0].server).toMatchObject({
|
||||
id: 'server-indeterminate-race',
|
||||
status: 'indeterminate',
|
||||
revision: 2,
|
||||
});
|
||||
expect(rendered.result.current.pendingEnqueueIds).toEqual([]);
|
||||
expect(rendered.result.current.settledReceipts).toEqual([]);
|
||||
});
|
||||
|
||||
it('anchors enqueue to the selected branch tail instead of a later hidden sibling', async () => {
|
||||
mockMessages = [
|
||||
{
|
||||
|
|
@ -482,16 +856,144 @@ describe('useSteering', () => {
|
|||
updatedAt: new Date(200).toISOString(),
|
||||
},
|
||||
];
|
||||
const { result } = setupServerQueue();
|
||||
const rendered = setupServerQueue();
|
||||
|
||||
await waitFor(() => expect(result.current.queue).toHaveLength(1));
|
||||
expect(result.current.queue[0]).toMatchObject({
|
||||
await waitFor(() => expect(rendered.result.current.queue).toHaveLength(1));
|
||||
expect(rendered.result.current.queue[0]).toMatchObject({
|
||||
id: 'client-snapshot-1',
|
||||
text: 'restored after reload',
|
||||
server: { id: 'server-snapshot-1', status: 'queued', revision: 3 },
|
||||
});
|
||||
});
|
||||
|
||||
it('fences the predecessor boundary when the server receipt is admitted', async () => {
|
||||
mockServerQueuedTurns = [
|
||||
{
|
||||
queuedTurnId: 'server-admitted-1',
|
||||
clientRequestId: 'client-admitted-1',
|
||||
conversationId: CONVO_ID,
|
||||
parentMessageId: 'visible-assistant-tail',
|
||||
text: 'already started by the server',
|
||||
status: 'admitted',
|
||||
revision: 4,
|
||||
expectedPredecessorCreatedAt: 41,
|
||||
effectivePredecessorCreatedAt: 42,
|
||||
createdAt: new Date(200).toISOString(),
|
||||
updatedAt: new Date(300).toISOString(),
|
||||
},
|
||||
];
|
||||
const { result } = setupServerQueue();
|
||||
|
||||
await waitFor(() =>
|
||||
expect(result.current.settledReceipts).toEqual([
|
||||
{
|
||||
clientRequestId: 'client-admitted-1',
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 42,
|
||||
},
|
||||
]),
|
||||
);
|
||||
expect(result.current.queue).toEqual([]);
|
||||
});
|
||||
|
||||
it('settles an explicit root admission without waiting for a timestamp boundary', async () => {
|
||||
mockServerQueuedTurns = [
|
||||
{
|
||||
queuedTurnId: 'server-root-admitted',
|
||||
clientRequestId: 'client-root-admitted',
|
||||
conversationId: CONVO_ID,
|
||||
parentMessageId: 'visible-assistant-tail',
|
||||
text: 'started from the conversation root',
|
||||
status: 'admitted',
|
||||
revision: 4,
|
||||
rootPredecessor: true,
|
||||
createdAt: new Date(200).toISOString(),
|
||||
updatedAt: new Date(300).toISOString(),
|
||||
},
|
||||
];
|
||||
const { result } = setupServerQueue();
|
||||
|
||||
await waitFor(() => expect(result.current.queue).toEqual([]));
|
||||
expect(result.current.settledReceipts).toEqual([]);
|
||||
});
|
||||
|
||||
it('keeps the terminal boundary when an old-server admission lacks its exact fence', async () => {
|
||||
mockServerQueuedTurns = [
|
||||
{
|
||||
queuedTurnId: 'server-admitted-race',
|
||||
clientRequestId: 'client-admitted-race',
|
||||
conversationId: CONVO_ID,
|
||||
parentMessageId: 'visible-assistant-tail',
|
||||
text: 'server-owned successor',
|
||||
status: 'admitted',
|
||||
revision: 4,
|
||||
expectedPredecessorCreatedAt: 41,
|
||||
createdAt: new Date(200).toISOString(),
|
||||
updatedAt: new Date(300).toISOString(),
|
||||
},
|
||||
];
|
||||
const ask = jest.fn();
|
||||
const sendNow = jest.fn();
|
||||
const wrapper = ({ children }: { children: React.ReactNode }) => (
|
||||
<RecoilRoot
|
||||
initializeState={withActiveGeneration(({ set }) => {
|
||||
set(store.isSubmittingFamily(0), false);
|
||||
set(store.runEndByIndex(0), {
|
||||
conversationId: CONVO_ID,
|
||||
outcome: 'completed',
|
||||
endedAt: 200,
|
||||
generationCreatedAt: 41,
|
||||
});
|
||||
set(store.queuedMessagesByConvoId(CONVO_ID), [
|
||||
{
|
||||
id: 'client-admitted-race',
|
||||
clientRequestId: 'client-admitted-race',
|
||||
text: 'server-owned successor',
|
||||
createdAt: 100,
|
||||
server: { id: 'server-admitted-race', status: 'claimed', revision: 3 },
|
||||
},
|
||||
{ id: 'local-successor', text: 'must wait', createdAt: 101 },
|
||||
]);
|
||||
})}
|
||||
>
|
||||
{children}
|
||||
</RecoilRoot>
|
||||
);
|
||||
const { result } = renderHook(
|
||||
() => {
|
||||
useSteering({
|
||||
index: 0,
|
||||
conversationId: CONVO_ID,
|
||||
conversation: agentsConversation,
|
||||
isSubmitting: false,
|
||||
answerModeActive: false,
|
||||
sendNow,
|
||||
stopGenerating: jest.fn(),
|
||||
});
|
||||
useQueueDrain(0, CONVO_ID, ask);
|
||||
return {
|
||||
queue: useQueue(CONVO_ID),
|
||||
runEnd: useRecoilValue(store.runEndByIndex(0)),
|
||||
};
|
||||
},
|
||||
{ wrapper },
|
||||
);
|
||||
|
||||
await waitFor(() =>
|
||||
expect(result.current.queue[0]).toMatchObject({
|
||||
id: 'client-admitted-race',
|
||||
server: { id: 'server-admitted-race', status: 'uncertain' },
|
||||
}),
|
||||
);
|
||||
expect(result.current.runEnd).not.toBeNull();
|
||||
expect(result.current.queue.map((item) => item.id)).toEqual([
|
||||
'client-admitted-race',
|
||||
'local-successor',
|
||||
]);
|
||||
expect(ask).not.toHaveBeenCalled();
|
||||
expect(sendNow).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('uses server sequence instead of Mongo timestamps for projected queue order', async () => {
|
||||
mockServerQueuedTurns = [
|
||||
{
|
||||
|
|
@ -542,10 +1044,10 @@ describe('useSteering', () => {
|
|||
},
|
||||
];
|
||||
mockCancelQueuedTurn.mockResolvedValueOnce({ status: 'cancelled' });
|
||||
const { result } = setupServerQueue();
|
||||
const rendered = setupServerQueue();
|
||||
|
||||
await waitFor(() => expect(result.current.queue).toHaveLength(1));
|
||||
expect(result.current.queue[0]).toMatchObject({
|
||||
await waitFor(() => expect(rendered.result.current.queue).toHaveLength(1));
|
||||
expect(rendered.result.current.queue[0]).toMatchObject({
|
||||
text: 'recover this failed turn',
|
||||
server: {
|
||||
id: 'server-dead-1',
|
||||
|
|
@ -556,15 +1058,48 @@ describe('useSteering', () => {
|
|||
});
|
||||
|
||||
await act(async () => {
|
||||
await expect(result.current.steering.discardQueued(result.current.queue[0])).resolves.toBe(
|
||||
true,
|
||||
);
|
||||
await expect(
|
||||
rendered.result.current.steering.discardQueued(rendered.result.current.queue[0]),
|
||||
).resolves.toBe(true);
|
||||
});
|
||||
expect(mockCancelQueuedTurn).toHaveBeenCalledWith({
|
||||
conversationId: CONVO_ID,
|
||||
queuedTurnId: 'server-dead-1',
|
||||
});
|
||||
expect(result.current.queue[0].server).toBeUndefined();
|
||||
expect(rendered.result.current.queue[0].server).toBeUndefined();
|
||||
});
|
||||
|
||||
it('projects an indeterminate claim as non-progressing reconciliation work', async () => {
|
||||
mockServerQueuedTurns = [
|
||||
{
|
||||
queuedTurnId: 'server-indeterminate-1',
|
||||
clientRequestId: 'client-indeterminate-1',
|
||||
conversationId: CONVO_ID,
|
||||
parentMessageId: 'visible-assistant-tail',
|
||||
text: 'external result is unknown',
|
||||
status: 'claimed',
|
||||
revision: 4,
|
||||
failure: {
|
||||
code: 'ADMISSION_INDETERMINATE',
|
||||
message: 'Awaiting exact source evidence',
|
||||
},
|
||||
createdAt: new Date(250).toISOString(),
|
||||
updatedAt: new Date(300).toISOString(),
|
||||
},
|
||||
];
|
||||
const rendered = setupServerQueue();
|
||||
|
||||
await waitFor(() => expect(rendered.result.current.queue).toHaveLength(1));
|
||||
expect(rendered.result.current.queue[0]).toMatchObject({
|
||||
text: 'external result is unknown',
|
||||
server: {
|
||||
id: 'server-indeterminate-1',
|
||||
status: 'indeterminate',
|
||||
errorCode: 'ADMISSION_INDETERMINATE',
|
||||
errorMessage: 'Awaiting exact source evidence',
|
||||
},
|
||||
});
|
||||
expect(rendered.result.current.settledReceipts).toEqual([]);
|
||||
});
|
||||
|
||||
it('cancels the durable copy before downgrading a row to local control', async () => {
|
||||
|
|
|
|||
|
|
@ -89,6 +89,13 @@ export default function useQueueDrain(
|
|||
* during the first turn stay keyed here until that run ends. Renewing only
|
||||
* the active id would skip them for the whole of that run. */
|
||||
const newConvoQueue = useRecoilValue(store.queuedMessagesByConvoId(Constants.NEW_CONVO));
|
||||
/** Receipt settlement can consume the parked terminal boundary without
|
||||
* changing whether another server-owned row remains. Subscribe here so the
|
||||
* drain effect observes that durable transition instead of reading it only
|
||||
* through a callback snapshot whose other dependencies stayed unchanged. */
|
||||
const settledQueuedTurnReceipts = useRecoilValue(
|
||||
store.settledQueuedTurnReceiptsByConvoId(activeConversationId ?? Constants.NEW_CONVO),
|
||||
);
|
||||
const hasServerOwnedQueue = [...ownQueue, ...newConvoQueue].some((item) => item.server != null);
|
||||
|
||||
/* Deduped because the two subscriptions are the same atom before migration.
|
||||
|
|
@ -238,6 +245,57 @@ export default function useQueueDrain(
|
|||
: ownQueue;
|
||||
|
||||
const shouldDrain = end.outcome === 'completed' || interruptArmed;
|
||||
const settledReceipts = snapshot
|
||||
.getLoadable(store.settledQueuedTurnReceiptsByConvoId(conversationId))
|
||||
.getValue();
|
||||
const pendingEnqueueIds = snapshot
|
||||
.getLoadable(store.pendingQueuedTurnEnqueueIdsByConvoId(conversationId))
|
||||
.getValue();
|
||||
const consumedReceiptIndex = settledReceipts.findIndex(
|
||||
(receipt) =>
|
||||
receipt.status === 'admitted' &&
|
||||
receipt.boundaryConsumed !== true &&
|
||||
end.generationCreatedAt != null &&
|
||||
receipt.effectivePredecessorCreatedAt === end.generationCreatedAt,
|
||||
);
|
||||
const consumedByServerAdmission = consumedReceiptIndex >= 0;
|
||||
const consumeEnd = () => {
|
||||
if (fromParked && activeConversationId) {
|
||||
set(store.pendingRunEndByConvoId(activeConversationId), null);
|
||||
} else {
|
||||
set(store.runEndByIndex(index), null);
|
||||
}
|
||||
if (matchingIndexArm) {
|
||||
set(store.drainAfterAbortByIndex(index), false);
|
||||
}
|
||||
};
|
||||
if (consumedByServerAdmission) {
|
||||
/** Admission consumed this terminal boundary on the server. A late
|
||||
* client terminal observation cannot authorize a second successor. */
|
||||
if (shouldMigrate && newConvoQueue.length > 0) {
|
||||
set(store.queuedMessagesByConvoId(Constants.NEW_CONVO), []);
|
||||
set(store.queuedMessagesByConvoId(conversationId), merged);
|
||||
}
|
||||
set(store.settledQueuedTurnReceiptsByConvoId(conversationId), (previous) => {
|
||||
let consumed = false;
|
||||
return previous.flatMap((receipt) => {
|
||||
if (
|
||||
!consumed &&
|
||||
receipt.status === 'admitted' &&
|
||||
receipt.boundaryConsumed !== true &&
|
||||
receipt.effectivePredecessorCreatedAt === end.generationCreatedAt
|
||||
) {
|
||||
consumed = true;
|
||||
return pendingEnqueueIds.includes(receipt.clientRequestId)
|
||||
? [{ ...receipt, boundaryConsumed: true }]
|
||||
: [];
|
||||
}
|
||||
return [receipt];
|
||||
});
|
||||
});
|
||||
consumeEnd();
|
||||
return null;
|
||||
}
|
||||
/** A server-owned Agent row means the backend owns the next fresh-turn
|
||||
* admission. Do not let a legacy/recovered local row race or overtake
|
||||
* it. Keep this terminal boundary available until the authoritative
|
||||
|
|
@ -253,14 +311,7 @@ export default function useQueueDrain(
|
|||
|
||||
// Consume only after server authority has yielded the boundary — a
|
||||
// hard double-fire guard even if the effect re-runs before propagation.
|
||||
if (fromParked && activeConversationId) {
|
||||
set(store.pendingRunEndByConvoId(activeConversationId), null);
|
||||
} else {
|
||||
set(store.runEndByIndex(index), null);
|
||||
}
|
||||
if (matchingIndexArm) {
|
||||
set(store.drainAfterAbortByIndex(index), false);
|
||||
}
|
||||
consumeEnd();
|
||||
|
||||
const next = shouldDrain ? (merged[0] ?? null) : null;
|
||||
const remainder = next ? merged.slice(1) : merged;
|
||||
|
|
@ -362,6 +413,7 @@ export default function useQueueDrain(
|
|||
restoreQueued,
|
||||
markFilesUsage,
|
||||
hasServerOwnedQueue,
|
||||
settledQueuedTurnReceipts,
|
||||
ask,
|
||||
]);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -14,7 +14,13 @@ import type {
|
|||
TConversation,
|
||||
TMessageContentParts,
|
||||
} from 'librechat-data-provider';
|
||||
import type { RunEnd, PendingSteer, QueuedMessage, QueuedMessageOrigin } from '~/store/families';
|
||||
import type {
|
||||
RunEnd,
|
||||
PendingSteer,
|
||||
QueuedMessage,
|
||||
QueuedMessageOrigin,
|
||||
SettledQueuedTurnReceipt,
|
||||
} from '~/store/families';
|
||||
import type { AgentQueuedTurnReceipt, GenerationProtocolVersion } from '~/data-provider';
|
||||
import type { ExtendedFile, FileSetter } from '~/common';
|
||||
import {
|
||||
|
|
@ -172,6 +178,76 @@ function queuedTurnCreatedAt(receipt: AgentQueuedTurnReceipt): number {
|
|||
return Number.isFinite(createdAt) ? createdAt : 0;
|
||||
}
|
||||
|
||||
type QueuedTurnReceiptSource = 'snapshot' | 'enqueue' | 'direct';
|
||||
|
||||
function isAdmissionIndeterminateReceipt(receipt: AgentQueuedTurnReceipt): boolean {
|
||||
return (
|
||||
receipt.failure?.code === 'ADMISSION_INDETERMINATE' &&
|
||||
(receipt.status === 'claimed' || receipt.status === 'dead')
|
||||
);
|
||||
}
|
||||
|
||||
function settledEvidenceForReceipt(
|
||||
receipt: AgentQueuedTurnReceipt,
|
||||
): SettledQueuedTurnReceipt | undefined {
|
||||
if (
|
||||
receipt.status === 'admitted' &&
|
||||
(receipt.effectivePredecessorCreatedAt != null || receipt.rootPredecessor === true)
|
||||
) {
|
||||
return {
|
||||
clientRequestId: receipt.clientRequestId,
|
||||
status: 'admitted',
|
||||
...(receipt.effectivePredecessorCreatedAt != null && {
|
||||
effectivePredecessorCreatedAt: receipt.effectivePredecessorCreatedAt,
|
||||
}),
|
||||
...(receipt.rootPredecessor === true && { rootPredecessor: true }),
|
||||
};
|
||||
}
|
||||
if (receipt.status === 'admitted') {
|
||||
return {
|
||||
clientRequestId: receipt.clientRequestId,
|
||||
status: 'admitted_pending_boundary',
|
||||
};
|
||||
}
|
||||
if (isAdmissionIndeterminateReceipt(receipt)) {
|
||||
return { clientRequestId: receipt.clientRequestId, status: 'indeterminate' };
|
||||
}
|
||||
if (receipt.status === 'cancelled') {
|
||||
return { clientRequestId: receipt.clientRequestId, status: 'cancelled' };
|
||||
}
|
||||
if (receipt.status === 'dead') {
|
||||
return { clientRequestId: receipt.clientRequestId, status: 'dead' };
|
||||
}
|
||||
return undefined;
|
||||
}
|
||||
|
||||
/** Client receipt knowledge is a monotonic evidence lattice. Enqueue is the
|
||||
* only source that can carry a pre-scheduling snapshot, so it never weakens
|
||||
* evidence already observed by GET/direct settlement. Indeterminate evidence
|
||||
* can advance only through a later authoritative terminal observation. */
|
||||
function mergeSettledQueuedTurnEvidence(
|
||||
existing: SettledQueuedTurnReceipt | undefined,
|
||||
receipt: AgentQueuedTurnReceipt,
|
||||
source: QueuedTurnReceiptSource,
|
||||
): SettledQueuedTurnReceipt | undefined {
|
||||
const incoming = settledEvidenceForReceipt(receipt);
|
||||
if (existing == null) {
|
||||
return incoming;
|
||||
}
|
||||
if (existing.status === 'admitted_pending_boundary' && incoming?.status === 'admitted') {
|
||||
return incoming;
|
||||
}
|
||||
if (
|
||||
existing.status === 'indeterminate' &&
|
||||
source !== 'enqueue' &&
|
||||
incoming != null &&
|
||||
incoming.status !== 'indeterminate'
|
||||
) {
|
||||
return incoming;
|
||||
}
|
||||
return existing;
|
||||
}
|
||||
|
||||
function toQueuedTurnFileRefs(files: TMessage['files']): TAgentQueuedTurnFileRef[] | undefined {
|
||||
const refs = (files ?? []).flatMap((file): TAgentQueuedTurnFileRef[] => {
|
||||
if (typeof file.file_id !== 'string' || file.file_id.length === 0) {
|
||||
|
|
@ -195,6 +271,8 @@ function toQueuedTurnFileRefs(files: TMessage['files']): TAgentQueuedTurnFileRef
|
|||
function reconcileServerQueuedTurns(
|
||||
previous: QueuedMessage[],
|
||||
receipts: AgentQueuedTurnReceipt[],
|
||||
settledByRequestId: ReadonlyMap<string, SettledQueuedTurnReceipt>,
|
||||
authoritativeSnapshot = true,
|
||||
): QueuedMessage[] {
|
||||
const previousByClientRequestId = new Map(
|
||||
previous.flatMap((item) =>
|
||||
|
|
@ -203,10 +281,30 @@ function reconcileServerQueuedTurns(
|
|||
);
|
||||
const observedClientRequestIds = new Set(receipts.map((receipt) => receipt.clientRequestId));
|
||||
const projected = receipts.flatMap((receipt): QueuedMessage[] => {
|
||||
if (receipt.status === 'admitted' || receipt.status === 'cancelled') {
|
||||
const settled = settledByRequestId.get(receipt.clientRequestId);
|
||||
const admissionIndeterminate = isAdmissionIndeterminateReceipt(receipt);
|
||||
if (settled?.status === 'admitted' || settled?.status === 'cancelled') {
|
||||
return [];
|
||||
}
|
||||
const optimistic = previousByClientRequestId.get(receipt.clientRequestId);
|
||||
if (
|
||||
(settled?.status === 'dead' ||
|
||||
settled?.status === 'admitted_pending_boundary' ||
|
||||
(settled?.status === 'indeterminate' && !admissionIndeterminate)) &&
|
||||
receipt.status !== 'dead' &&
|
||||
receipt.status !== 'admitted'
|
||||
) {
|
||||
return optimistic == null ? [] : [optimistic];
|
||||
}
|
||||
const boundaryPending = receipt.status === 'admitted';
|
||||
let status: NonNullable<QueuedMessage['server']>['status'] = 'rejected';
|
||||
if (admissionIndeterminate) {
|
||||
status = 'indeterminate';
|
||||
} else if (boundaryPending) {
|
||||
status = 'uncertain';
|
||||
} else if (receipt.status === 'queued' || receipt.status === 'claimed') {
|
||||
status = receipt.status;
|
||||
}
|
||||
return [
|
||||
{
|
||||
id: optimistic?.id ?? receipt.clientRequestId,
|
||||
|
|
@ -226,8 +324,11 @@ function reconcileServerQueuedTurns(
|
|||
...(receipt.priority === true && { priority: true }),
|
||||
server: {
|
||||
id: receipt.queuedTurnId,
|
||||
status: receipt.status === 'dead' ? 'rejected' : receipt.status,
|
||||
status,
|
||||
revision: receipt.revision,
|
||||
...(boundaryPending && {
|
||||
uncertainSince: optimistic?.server?.uncertainSince ?? Date.now(),
|
||||
}),
|
||||
...(receipt.position != null && { position: receipt.position }),
|
||||
...(receipt.failure?.code != null && { errorCode: receipt.failure.code }),
|
||||
...(receipt.failure?.message != null && { errorMessage: receipt.failure.message }),
|
||||
|
|
@ -239,12 +340,24 @@ function reconcileServerQueuedTurns(
|
|||
if (item.server == null) {
|
||||
return true;
|
||||
}
|
||||
const settled =
|
||||
item.clientRequestId == null ? undefined : settledByRequestId.get(item.clientRequestId);
|
||||
if (settled?.status === 'admitted' || settled?.status === 'cancelled') {
|
||||
return false;
|
||||
}
|
||||
if (settled?.status === 'dead' || settled?.status === 'admitted_pending_boundary') {
|
||||
return item.clientRequestId == null || !observedClientRequestIds.has(item.clientRequestId);
|
||||
}
|
||||
if (item.clientRequestId == null || observedClientRequestIds.has(item.clientRequestId)) {
|
||||
return false;
|
||||
}
|
||||
if (!authoritativeSnapshot) {
|
||||
return true;
|
||||
}
|
||||
return (
|
||||
item.server.status === 'sending' ||
|
||||
item.server.status === 'uncertain' ||
|
||||
item.server.status === 'indeterminate' ||
|
||||
item.server.status === 'rejected'
|
||||
);
|
||||
});
|
||||
|
|
@ -363,12 +476,93 @@ export default function useSteering({
|
|||
store.activeGenerationProtocolVersionByConvoId(queueKey),
|
||||
);
|
||||
|
||||
const applyQueuedTurnReceipts = useRecoilCallback(
|
||||
({ snapshot, set }) =>
|
||||
(receipts: AgentQueuedTurnReceipt[], source: QueuedTurnReceiptSource = 'snapshot') => {
|
||||
const previousSettled = snapshot
|
||||
.getLoadable(store.settledQueuedTurnReceiptsByConvoId(queueKey))
|
||||
.getValue();
|
||||
const previousPending = snapshot
|
||||
.getLoadable(store.pendingQueuedTurnEnqueueIdsByConvoId(queueKey))
|
||||
.getValue();
|
||||
const completedRequestIds = new Set(
|
||||
source === 'enqueue' ? receipts.map((receipt) => receipt.clientRequestId) : [],
|
||||
);
|
||||
const nextPending = previousPending.filter((id) => !completedRequestIds.has(id));
|
||||
const settledByRequestId = new Map(
|
||||
previousSettled.map((receipt) => [receipt.clientRequestId, receipt]),
|
||||
);
|
||||
for (const receipt of receipts) {
|
||||
const existing = settledByRequestId.get(receipt.clientRequestId);
|
||||
const settled = mergeSettledQueuedTurnEvidence(existing, receipt, source);
|
||||
if (settled == null) {
|
||||
continue;
|
||||
}
|
||||
settledByRequestId.set(receipt.clientRequestId, settled);
|
||||
}
|
||||
const terminalForReconciliation = new Map(settledByRequestId);
|
||||
const pendingRequestIds = new Set(nextPending);
|
||||
const nextSettled = [...settledByRequestId.values()].filter(
|
||||
(receipt) =>
|
||||
(receipt.status === 'admitted' &&
|
||||
receipt.rootPredecessor !== true &&
|
||||
receipt.boundaryConsumed !== true) ||
|
||||
pendingRequestIds.has(receipt.clientRequestId),
|
||||
);
|
||||
if (source === 'enqueue') {
|
||||
set(store.pendingQueuedTurnEnqueueIdsByConvoId(queueKey), nextPending);
|
||||
}
|
||||
if (
|
||||
nextSettled.length !== previousSettled.length ||
|
||||
nextSettled.some((receipt, index) => receipt !== previousSettled[index])
|
||||
) {
|
||||
set(store.settledQueuedTurnReceiptsByConvoId(queueKey), nextSettled);
|
||||
}
|
||||
if (source !== 'direct') {
|
||||
set(store.queuedMessagesByConvoId(queueKey), (previous) =>
|
||||
reconcileServerQueuedTurns(
|
||||
previous,
|
||||
receipts,
|
||||
terminalForReconciliation,
|
||||
source === 'snapshot',
|
||||
),
|
||||
);
|
||||
}
|
||||
},
|
||||
[queueKey],
|
||||
);
|
||||
|
||||
const finishQueuedTurnEnqueue = useRecoilCallback(
|
||||
({ snapshot, set }) =>
|
||||
(clientRequestId: string): boolean => {
|
||||
const settledReceipts = snapshot
|
||||
.getLoadable(store.settledQueuedTurnReceiptsByConvoId(queueKey))
|
||||
.getValue();
|
||||
const settled = settledReceipts.find(
|
||||
(receipt) => receipt.clientRequestId === clientRequestId,
|
||||
);
|
||||
set(store.pendingQueuedTurnEnqueueIdsByConvoId(queueKey), (previous) =>
|
||||
previous.filter((id) => id !== clientRequestId),
|
||||
);
|
||||
if (
|
||||
settled != null &&
|
||||
(settled.status !== 'admitted' || settled.boundaryConsumed === true)
|
||||
) {
|
||||
set(store.settledQueuedTurnReceiptsByConvoId(queueKey), (previous) =>
|
||||
previous.filter((receipt) => receipt.clientRequestId !== clientRequestId),
|
||||
);
|
||||
}
|
||||
return settled != null;
|
||||
},
|
||||
[queueKey],
|
||||
);
|
||||
|
||||
useEffect(() => {
|
||||
if (!serverQueueEnabled || serverQueuedTurns == null) {
|
||||
return;
|
||||
}
|
||||
setQueuedMessages((previous) => reconcileServerQueuedTurns(previous, serverQueuedTurns));
|
||||
}, [serverQueueEnabled, serverQueuedTurns, setQueuedMessages]);
|
||||
applyQueuedTurnReceipts(serverQueuedTurns);
|
||||
}, [serverQueueEnabled, serverQueuedTurns, applyQueuedTurnReceipts]);
|
||||
useEffect(() => {
|
||||
if (!serverQueueEnabled || nextReconciliationExpiry == null) {
|
||||
return;
|
||||
|
|
@ -796,6 +990,9 @@ export default function useSteering({
|
|||
if (!serverOwned || clientRequestId == null) {
|
||||
return;
|
||||
}
|
||||
set(store.pendingQueuedTurnEnqueueIdsByConvoId(queueKey), (previous) =>
|
||||
previous.includes(clientRequestId) ? previous : [...previous, clientRequestId],
|
||||
);
|
||||
const serverFiles = toQueuedTurnFileRefs(item.files);
|
||||
/** Commit the optimistic Recoil row before mutation callbacks can
|
||||
* reconcile it. This also makes synchronous test/adaptor completions
|
||||
|
|
@ -820,29 +1017,12 @@ export default function useSteering({
|
|||
},
|
||||
{
|
||||
onSuccess: (receipt) => {
|
||||
updateQueuedMessage(item.id, (current) => {
|
||||
if (receipt.status === 'admitted' || receipt.status === 'cancelled') {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
...current,
|
||||
text: receipt.text,
|
||||
createdAt: queuedTurnCreatedAt(receipt),
|
||||
parentMessageId: receipt.parentMessageId,
|
||||
server: {
|
||||
id: receipt.queuedTurnId,
|
||||
status: receipt.status === 'dead' ? 'rejected' : receipt.status,
|
||||
revision: receipt.revision,
|
||||
...(receipt.position != null && { position: receipt.position }),
|
||||
...(receipt.failure?.code != null && { errorCode: receipt.failure.code }),
|
||||
...(receipt.failure?.message != null && {
|
||||
errorMessage: receipt.failure.message,
|
||||
}),
|
||||
},
|
||||
};
|
||||
});
|
||||
applyQueuedTurnReceipts([receipt], 'enqueue');
|
||||
},
|
||||
onError: (error) => {
|
||||
if (finishQueuedTurnEnqueue(clientRequestId)) {
|
||||
return;
|
||||
}
|
||||
updateQueuedMessage(item.id, (current) => {
|
||||
if (isDefiniteQueuedTurnsUnsupported(error)) {
|
||||
const {
|
||||
|
|
@ -891,6 +1071,8 @@ export default function useSteering({
|
|||
markQueuedFilesUsage,
|
||||
activeGenerationCreatedAt,
|
||||
enqueueAgentQueuedTurn,
|
||||
applyQueuedTurnReceipts,
|
||||
finishQueuedTurnEnqueue,
|
||||
updateQueuedMessage,
|
||||
],
|
||||
);
|
||||
|
|
@ -1115,6 +1297,7 @@ export default function useSteering({
|
|||
});
|
||||
return false;
|
||||
}
|
||||
applyQueuedTurnReceipts([receipt], 'direct');
|
||||
return downgradeServerQueuedTurn(item.id);
|
||||
} catch {
|
||||
showToast({
|
||||
|
|
@ -1159,6 +1342,7 @@ export default function useSteering({
|
|||
[
|
||||
cancelSteer,
|
||||
cancelAgentQueuedTurn,
|
||||
applyQueuedTurnReceipts,
|
||||
conversationId,
|
||||
downgradeQueuedRecovery,
|
||||
downgradeServerQueuedTurn,
|
||||
|
|
|
|||
|
|
@ -1791,6 +1791,7 @@
|
|||
"com_ui_queued_messages": "Queued messages",
|
||||
"com_ui_queued_quote_count": "{{0}} quoted excerpts included with this message",
|
||||
"com_ui_queued_turn_failed": "Queued message failed",
|
||||
"com_ui_queued_turn_reconciliation_required": "Awaiting reconciliation",
|
||||
"com_ui_quote_selections": "{{0}} selections",
|
||||
"com_ui_quotes_queued": "Quotes added for your next message",
|
||||
"com_ui_ran_n_agents": "Ran {{0}} agents",
|
||||
|
|
|
|||
|
|
@ -402,7 +402,7 @@ export type QueuedMessage = {
|
|||
* after an ambiguous POST could submit the same words twice. */
|
||||
server?: {
|
||||
id?: string;
|
||||
status: 'sending' | 'uncertain' | 'rejected' | 'queued' | 'claimed';
|
||||
status: 'sending' | 'uncertain' | 'indeterminate' | 'rejected' | 'queued' | 'claimed';
|
||||
errorCode?: string;
|
||||
errorMessage?: string;
|
||||
/** Observation time for a transport-ambiguous enqueue. The logical item
|
||||
|
|
@ -459,6 +459,29 @@ const queuedMessagesByConvoId = atomFamily<QueuedMessage[], string>({
|
|||
default: [],
|
||||
});
|
||||
|
||||
export type SettledQueuedTurnReceipt = {
|
||||
clientRequestId: string;
|
||||
status: 'admitted' | 'admitted_pending_boundary' | 'indeterminate' | 'cancelled' | 'dead';
|
||||
effectivePredecessorCreatedAt?: number;
|
||||
rootPredecessor?: true;
|
||||
boundaryConsumed?: boolean;
|
||||
};
|
||||
|
||||
/** Monotonic client knowledge of terminal server queue receipts. Admission
|
||||
* records preserve boundary multiplicity by request identity. Other terminal
|
||||
* records exist only while their original enqueue callback is outstanding. */
|
||||
const settledQueuedTurnReceiptsByConvoId = atomFamily<SettledQueuedTurnReceipt[], string>({
|
||||
key: 'settledQueuedTurnReceiptsByConvoId',
|
||||
default: [],
|
||||
});
|
||||
|
||||
/** Enqueue callbacks that can still race newer GET/cancellation evidence.
|
||||
* Entries retire as soon as that one callback settles. */
|
||||
const pendingQueuedTurnEnqueueIdsByConvoId = atomFamily<string[], string>({
|
||||
key: 'pendingQueuedTurnEnqueueIdsByConvoId',
|
||||
default: [],
|
||||
});
|
||||
|
||||
/**
|
||||
* One-shot run-termination signal written by the SSE final/error handlers and
|
||||
* consumed (reset to null) by `useQueueDrain`. Keyed by chat index like
|
||||
|
|
@ -788,6 +811,8 @@ export default {
|
|||
pendingQuotesByConvoId,
|
||||
pendingSteersByConvoId,
|
||||
queuedMessagesByConvoId,
|
||||
settledQueuedTurnReceiptsByConvoId,
|
||||
pendingQueuedTurnEnqueueIdsByConvoId,
|
||||
runEndByIndex,
|
||||
pendingRunEndByConvoId,
|
||||
drainAfterAbortByIndex,
|
||||
|
|
|
|||
|
|
@ -67,10 +67,21 @@ describe('Agent queued-turn HTTP admission receipts', () => {
|
|||
const enqueueAgentQueuedTurn = jest
|
||||
.fn()
|
||||
.mockResolvedValueOnce({ turn: turn('queued'), replayed: false });
|
||||
const admitted = {
|
||||
...turn('admitted'),
|
||||
terminalReceipt: {
|
||||
outcome: 'admitted' as const,
|
||||
settledAt: new Date('2026-08-30T12:01:00Z'),
|
||||
admissionId: 'client-request-1',
|
||||
generationId: 'generation-1',
|
||||
generationCreatedAt: 43,
|
||||
effectivePredecessorCreatedAt: 42,
|
||||
},
|
||||
};
|
||||
const getAgentQueuedTurnByClientRequestId = jest
|
||||
.fn()
|
||||
.mockResolvedValueOnce(null)
|
||||
.mockResolvedValueOnce(turn('admitted'));
|
||||
.mockResolvedValueOnce(admitted);
|
||||
const schedule = jest.fn().mockRejectedValueOnce(new Error('scheduler unavailable'));
|
||||
const methods = {
|
||||
getConvo: jest.fn(async () => ({
|
||||
|
|
@ -104,6 +115,7 @@ describe('Agent queued-turn HTTP admission receipts', () => {
|
|||
queuedTurnId: 'queued-turn-1',
|
||||
clientRequestId: 'client-request-1',
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 42,
|
||||
},
|
||||
},
|
||||
});
|
||||
|
|
@ -172,6 +184,47 @@ describe('Agent queued-turn HTTP admission receipts', () => {
|
|||
expect(cancel).toHaveBeenCalledWith(expect.objectContaining({ queuedTurnId: 'queued-turn-1' }));
|
||||
});
|
||||
|
||||
it('projects an explicit root admission without a timestamp boundary', async () => {
|
||||
const admitted = {
|
||||
...turn('admitted'),
|
||||
terminalReceipt: {
|
||||
outcome: 'admitted' as const,
|
||||
settledAt: new Date('2026-08-30T12:01:00Z'),
|
||||
admissionId: 'client-request-1',
|
||||
generationId: 'generation-root',
|
||||
generationCreatedAt: 43,
|
||||
lineagePredecessorId: 'root:message-identity',
|
||||
rootPredecessor: true as const,
|
||||
},
|
||||
};
|
||||
const methods = {
|
||||
getConvo: jest.fn(async () => ({ agent_id: 'agent_1', endpoint: 'agents' })),
|
||||
listAgentQueuedTurnReceipts: jest.fn(async () => [admitted]),
|
||||
};
|
||||
const deps = {
|
||||
methods: methods as unknown as AgentQueuedTurnMethods & {
|
||||
getConvo: typeof methods.getConvo;
|
||||
},
|
||||
lifecycle: { schedule: jest.fn(), cancel: jest.fn() },
|
||||
checkAgentAccess: jest.fn(async () => true),
|
||||
} satisfies AgentQueuedTurnHttpDeps;
|
||||
|
||||
await expect(
|
||||
handleAgentQueuedTurnList({ id: USER_ID }, 'conversation-1', deps),
|
||||
).resolves.toMatchObject({
|
||||
status: 200,
|
||||
body: {
|
||||
queuedTurns: [
|
||||
{
|
||||
queuedTurnId: 'queued-turn-1',
|
||||
status: 'admitted',
|
||||
rootPredecessor: true,
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it('retires a cancelled source after its published delivery receipt expires', async () => {
|
||||
const cancelled = {
|
||||
...turn('cancelled'),
|
||||
|
|
|
|||
|
|
@ -90,6 +90,12 @@ function receipt(
|
|||
...(turn.expectedPredecessorCreatedAt != null && {
|
||||
expectedPredecessorCreatedAt: turn.expectedPredecessorCreatedAt,
|
||||
}),
|
||||
...(turn.terminalReceipt?.outcome === 'admitted' &&
|
||||
turn.terminalReceipt.effectivePredecessorCreatedAt != null && {
|
||||
effectivePredecessorCreatedAt: turn.terminalReceipt.effectivePredecessorCreatedAt,
|
||||
}),
|
||||
...(turn.terminalReceipt?.outcome === 'admitted' &&
|
||||
turn.terminalReceipt.rootPredecessor === true && { rootPredecessor: true as const }),
|
||||
status: turn.status,
|
||||
...(position != null && { position }),
|
||||
revision: turn.sequence,
|
||||
|
|
|
|||
|
|
@ -51,6 +51,7 @@ function claim(): AgentQueuedTurnClaim {
|
|||
clientRequestId: 'client-1',
|
||||
fingerprint: 'fingerprint-1',
|
||||
sequence: 1,
|
||||
admissionSlot: true,
|
||||
status: 'claimed',
|
||||
priority: false,
|
||||
text: 'queued words',
|
||||
|
|
@ -107,7 +108,7 @@ function resolverMethods() {
|
|||
getEffectiveAgentQueuedTurnPredecessor: jest.fn(
|
||||
async (
|
||||
..._args: Parameters<AgentQueuedTurnMethods['getEffectiveAgentQueuedTurnPredecessor']>
|
||||
): Promise<number | undefined> => undefined,
|
||||
) => null,
|
||||
),
|
||||
markAgentQueuedTurnAdmitted: jest.fn(async () => ({
|
||||
outcome: 'admitted' as const,
|
||||
|
|
@ -306,6 +307,7 @@ describe('Agent queued-turn continuation', () => {
|
|||
sourceId: 'queued-turn-1',
|
||||
claimId: 'trigger-1',
|
||||
claimBy: 'worker-1',
|
||||
effectivePredecessorCreatedAt: NOW,
|
||||
},
|
||||
{
|
||||
userId: USER_ID,
|
||||
|
|
@ -325,6 +327,7 @@ describe('Agent queued-turn continuation', () => {
|
|||
admissionId: 'trigger-1',
|
||||
generationId: 'conversation-1',
|
||||
generationCreatedAt: NOW + 1,
|
||||
effectivePredecessorCreatedAt: NOW,
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
|
@ -342,6 +345,7 @@ describe('Agent queued-turn continuation', () => {
|
|||
sourceId: 'queued-turn-1',
|
||||
claimId: 'trigger-1',
|
||||
claimBy: 'worker-1',
|
||||
effectivePredecessorCreatedAt: NOW,
|
||||
};
|
||||
const admission = {
|
||||
userId: USER_ID,
|
||||
|
|
@ -361,6 +365,7 @@ describe('Agent queued-turn continuation', () => {
|
|||
admissionId: 'trigger-1',
|
||||
generationId: 'conversation-1',
|
||||
generationCreatedAt: NOW + 1,
|
||||
effectivePredecessorCreatedAt: NOW,
|
||||
});
|
||||
|
||||
spies.hasAgentQueuedTurnAdmissionReceipt.mockResolvedValueOnce(false);
|
||||
|
|
@ -451,7 +456,10 @@ describe('Agent queued-turn continuation', () => {
|
|||
const claimed = claim();
|
||||
claimed.expectedPredecessorCreatedAt = NOW;
|
||||
spies.claimNextAgentQueuedTurn.mockResolvedValueOnce({ outcome: 'acquired', claim: claimed });
|
||||
spies.getEffectiveAgentQueuedTurnPredecessor.mockResolvedValueOnce(NOW + 250);
|
||||
spies.beginAgentQueuedTurnAdmission.mockResolvedValueOnce({
|
||||
outcome: 'started',
|
||||
turn: { ...claimed, admissionEffectivePredecessorCreatedAt: NOW + 250 },
|
||||
});
|
||||
const resolve = createAgentQueuedTurnResolver({
|
||||
methods,
|
||||
getGenerationJob: async () => null,
|
||||
|
|
@ -459,17 +467,53 @@ describe('Agent queued-turn continuation', () => {
|
|||
claimBy: 'worker-1',
|
||||
});
|
||||
|
||||
await expect(resolve(envelope(), { idempotencyKey: 'trigger-1' })).resolves.toMatchObject({
|
||||
const prepared = await resolve(envelope(), { idempotencyKey: 'trigger-1' });
|
||||
expect(prepared).toMatchObject({
|
||||
status: 'ready',
|
||||
expectedPredecessorCreatedAt: NOW + 250,
|
||||
admissionSource: { effectivePredecessorCreatedAt: NOW + 250 },
|
||||
});
|
||||
expect(spies.getEffectiveAgentQueuedTurnPredecessor).toHaveBeenCalledWith({
|
||||
user: new Types.ObjectId(USER_ID),
|
||||
tenantId: 'tenant-1',
|
||||
expect(spies.beginAgentQueuedTurnAdmission).toHaveBeenCalledTimes(1);
|
||||
if (prepared?.status !== 'ready') {
|
||||
throw new Error('Expected a ready queued turn');
|
||||
}
|
||||
await prepared.settleOnAdmission?.({
|
||||
mode: 'continue',
|
||||
status: 'started',
|
||||
conversationId: 'conversation-1',
|
||||
sequence: 1,
|
||||
expectedPredecessorCreatedAt: NOW,
|
||||
streamId: 'stream-2',
|
||||
generationCreatedAt: NOW + 500,
|
||||
});
|
||||
expect(spies.markAgentQueuedTurnAdmitted).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
queuedTurnId: 'queued-turn-1',
|
||||
effectivePredecessorCreatedAt: NOW + 250,
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it('dead-letters a legacy admission order that cannot be reconstructed safely', async () => {
|
||||
const { methods, spies } = resolverMethods();
|
||||
(
|
||||
spies.beginAgentQueuedTurnAdmission as unknown as jest.MockedFunction<
|
||||
AgentQueuedTurnMethods['beginAgentQueuedTurnAdmission']
|
||||
>
|
||||
).mockResolvedValueOnce({
|
||||
outcome: 'order_unavailable',
|
||||
turn: { ...claim(), status: 'dead' },
|
||||
});
|
||||
const resolve = createAgentQueuedTurnResolver({
|
||||
methods,
|
||||
getGenerationJob: async () => null,
|
||||
now: () => NOW,
|
||||
claimBy: 'worker-1',
|
||||
});
|
||||
|
||||
await expect(resolve(envelope(), { idempotencyKey: 'trigger-1' })).resolves.toEqual({
|
||||
status: 'settled',
|
||||
});
|
||||
expect(spies.releaseAgentQueuedTurn).not.toHaveBeenCalled();
|
||||
expect(spies.beginAgentQueuedTurnAdmission).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it.each([
|
||||
|
|
@ -776,10 +820,10 @@ describe('Agent queued-turn delivery scheduling', () => {
|
|||
);
|
||||
});
|
||||
|
||||
it('reconciles quarantined admission receipts during durable recovery', async () => {
|
||||
it('reconciles a claimed indeterminate legacy admission during durable recovery', async () => {
|
||||
const turn = {
|
||||
...queuedTurn('queued-turn-indeterminate', 1),
|
||||
status: 'dead' as const,
|
||||
status: 'claimed' as const,
|
||||
deliveryKey: 'delivery-indeterminate',
|
||||
deliveryState: 'published' as const,
|
||||
admissionId: 'delivery-indeterminate',
|
||||
|
|
@ -841,7 +885,7 @@ describe('Agent queued-turn delivery scheduling', () => {
|
|||
it('rotates source-fenced ambiguity without trusting transient job-store evidence', async () => {
|
||||
const turn = {
|
||||
...queuedTurn('queued-turn-source-fenced', 1),
|
||||
status: 'dead' as const,
|
||||
status: 'claimed' as const,
|
||||
deliveryKey: 'delivery-source-fenced',
|
||||
deliveryState: 'published' as const,
|
||||
admissionId: 'delivery-source-fenced',
|
||||
|
|
|
|||
|
|
@ -153,6 +153,9 @@ function parseQueuedTurnAdmissionSource(raw: unknown): AgentContinuationAdmissio
|
|||
const sourceId = 'sourceId' in raw ? raw.sourceId : undefined;
|
||||
const claimId = 'claimId' in raw ? raw.claimId : undefined;
|
||||
const claimBy = 'claimBy' in raw ? raw.claimBy : undefined;
|
||||
const effectivePredecessorCreatedAt =
|
||||
'effectivePredecessorCreatedAt' in raw ? raw.effectivePredecessorCreatedAt : undefined;
|
||||
const lineagePredecessorId = 'lineagePredecessorId' in raw ? raw.lineagePredecessorId : undefined;
|
||||
if (
|
||||
typeof sourceId !== 'string' ||
|
||||
sourceId.length === 0 ||
|
||||
|
|
@ -162,11 +165,28 @@ function parseQueuedTurnAdmissionSource(raw: unknown): AgentContinuationAdmissio
|
|||
claimId.length > 128 ||
|
||||
typeof claimBy !== 'string' ||
|
||||
claimBy.length === 0 ||
|
||||
claimBy.length > 256
|
||||
claimBy.length > 256 ||
|
||||
(effectivePredecessorCreatedAt != null &&
|
||||
(typeof effectivePredecessorCreatedAt !== 'number' ||
|
||||
!Number.isSafeInteger(effectivePredecessorCreatedAt) ||
|
||||
effectivePredecessorCreatedAt < 0)) ||
|
||||
(lineagePredecessorId != null &&
|
||||
(typeof lineagePredecessorId !== 'string' ||
|
||||
lineagePredecessorId.length === 0 ||
|
||||
lineagePredecessorId.length > 128))
|
||||
) {
|
||||
throw new TypeError('Agent queued turn admission source is invalid');
|
||||
}
|
||||
return { source: AGENT_QUEUED_TURN_SOURCE, sourceId, claimId, claimBy };
|
||||
return {
|
||||
source: AGENT_QUEUED_TURN_SOURCE,
|
||||
sourceId,
|
||||
claimId,
|
||||
claimBy,
|
||||
...(typeof effectivePredecessorCreatedAt === 'number' && {
|
||||
effectivePredecessorCreatedAt,
|
||||
}),
|
||||
...(typeof lineagePredecessorId === 'string' && { lineagePredecessorId }),
|
||||
};
|
||||
}
|
||||
|
||||
/** Commits the source-owned admission receipt only after the controller has
|
||||
|
|
@ -195,6 +215,12 @@ async function settleAgentQueuedTurnExecutionAdmission(
|
|||
admissionMode: 'ordinary',
|
||||
generationId: input.generationId,
|
||||
generationCreatedAt: input.generationCreatedAt,
|
||||
...(source.effectivePredecessorCreatedAt != null && {
|
||||
effectivePredecessorCreatedAt: source.effectivePredecessorCreatedAt,
|
||||
}),
|
||||
...(source.lineagePredecessorId != null && {
|
||||
lineagePredecessorId: source.lineagePredecessorId,
|
||||
}),
|
||||
settledAt: new Date(),
|
||||
});
|
||||
if (settled.outcome === 'conflict') {
|
||||
|
|
@ -226,6 +252,12 @@ async function verifyAgentQueuedTurnExecutionAdmission(
|
|||
admissionId: input.clientRequestId,
|
||||
generationId: input.generationId,
|
||||
generationCreatedAt: input.generationCreatedAt,
|
||||
...(source.effectivePredecessorCreatedAt != null && {
|
||||
effectivePredecessorCreatedAt: source.effectivePredecessorCreatedAt,
|
||||
}),
|
||||
...(source.lineagePredecessorId != null && {
|
||||
lineagePredecessorId: source.lineagePredecessorId,
|
||||
}),
|
||||
});
|
||||
if (!confirmed) {
|
||||
throw new Error('The queued turn execution admission is not yet confirmed');
|
||||
|
|
@ -578,17 +610,6 @@ function createAgentQueuedTurnResolver({
|
|||
return { status: 'settled' };
|
||||
}
|
||||
|
||||
const effectivePredecessorCreatedAt =
|
||||
(await methods.getEffectiveAgentQueuedTurnPredecessor({
|
||||
user: claim.user,
|
||||
...(claim.tenantId != null && { tenantId: claim.tenantId }),
|
||||
conversationId: claim.conversationId,
|
||||
sequence: claim.sequence,
|
||||
...(claim.expectedPredecessorCreatedAt != null && {
|
||||
expectedPredecessorCreatedAt: claim.expectedPredecessorCreatedAt,
|
||||
}),
|
||||
})) ?? claim.expectedPredecessorCreatedAt;
|
||||
|
||||
const admission = await methods.beginAgentQueuedTurnAdmission({
|
||||
user: claim.user,
|
||||
...(claim.tenantId != null && { tenantId: claim.tenantId }),
|
||||
|
|
@ -607,9 +628,11 @@ function createAgentQueuedTurnResolver({
|
|||
deferWithoutAttempt: true,
|
||||
});
|
||||
}
|
||||
if (admission.outcome === 'retired') {
|
||||
if (admission.outcome === 'retired' || admission.outcome === 'order_unavailable') {
|
||||
return { status: 'settled' };
|
||||
}
|
||||
const effectivePredecessorCreatedAt = admission.turn.admissionEffectivePredecessorCreatedAt;
|
||||
const lineagePredecessorId = admission.turn.admissionLineagePredecessorId;
|
||||
|
||||
return {
|
||||
status: 'ready',
|
||||
|
|
@ -626,6 +649,8 @@ function createAgentQueuedTurnResolver({
|
|||
sourceId: claim.queuedTurnId,
|
||||
claimId: claim.claimId,
|
||||
claimBy: claim.claimBy,
|
||||
...(effectivePredecessorCreatedAt != null && { effectivePredecessorCreatedAt }),
|
||||
...(lineagePredecessorId != null && { lineagePredecessorId }),
|
||||
},
|
||||
releaseOnDefiniteFailure: async (error) => {
|
||||
if (
|
||||
|
|
@ -659,6 +684,8 @@ function createAgentQueuedTurnResolver({
|
|||
...(result.generationCreatedAt != null && {
|
||||
generationCreatedAt: result.generationCreatedAt,
|
||||
}),
|
||||
...(effectivePredecessorCreatedAt != null && { effectivePredecessorCreatedAt }),
|
||||
...(lineagePredecessorId != null && { lineagePredecessorId }),
|
||||
settledAt: new Date(now()),
|
||||
});
|
||||
if (settled.outcome === 'conflict') {
|
||||
|
|
@ -794,7 +821,10 @@ function createAgentQueuedTurnScheduler({
|
|||
}),
|
||||
);
|
||||
try {
|
||||
if (turn.status === 'claimed') {
|
||||
const isIndeterminate =
|
||||
turn.terminalReceipt?.outcome === 'dead' &&
|
||||
turn.terminalReceipt.failure?.code === 'ADMISSION_INDETERMINATE';
|
||||
if (turn.status === 'claimed' && !isIndeterminate) {
|
||||
const result = await runAsSystem(() =>
|
||||
methods.deadLetterAgentQueuedTurn({
|
||||
user: turn.user,
|
||||
|
|
|
|||
|
|
@ -43,6 +43,8 @@ export interface AgentContinuationAdmissionSource {
|
|||
sourceId: string;
|
||||
claimId: string;
|
||||
claimBy: string;
|
||||
effectivePredecessorCreatedAt?: number;
|
||||
lineagePredecessorId?: string;
|
||||
}
|
||||
|
||||
export type AgentTriggerContinuePreparation =
|
||||
|
|
|
|||
|
|
@ -94,6 +94,34 @@ describe('agent queued turn schemas', () => {
|
|||
expect(agentQueuedTurnReceiptSchema.parse(receipt)).toEqual(receipt);
|
||||
});
|
||||
|
||||
it('parses the effective boundary consumed by an admitted turn', () => {
|
||||
const receipt = {
|
||||
...request,
|
||||
queuedTurnId: 'turn-1',
|
||||
status: 'admitted',
|
||||
effectivePredecessorCreatedAt: 84,
|
||||
revision: 2,
|
||||
createdAt: '2026-08-30T12:00:00.000Z',
|
||||
updatedAt: '2026-08-30T12:01:00.000Z',
|
||||
};
|
||||
|
||||
expect(agentQueuedTurnReceiptSchema.parse(receipt)).toEqual(receipt);
|
||||
});
|
||||
|
||||
it('parses an explicit root admission without a synthetic timestamp', () => {
|
||||
const receipt = {
|
||||
...request,
|
||||
queuedTurnId: 'turn-root',
|
||||
status: 'admitted',
|
||||
rootPredecessor: true,
|
||||
revision: 2,
|
||||
createdAt: '2026-08-30T12:00:00.000Z',
|
||||
updatedAt: '2026-08-30T12:01:00.000Z',
|
||||
};
|
||||
|
||||
expect(agentQueuedTurnReceiptSchema.parse(receipt)).toEqual(receipt);
|
||||
});
|
||||
|
||||
it('requires durability only when the capability is supported', () => {
|
||||
expect(agentQueuedTurnCapabilitySchema.parse({ supported: false })).toEqual({
|
||||
supported: false,
|
||||
|
|
|
|||
|
|
@ -52,7 +52,13 @@ export type TCancelAgentQueuedTurnRequest = z.infer<typeof cancelAgentQueuedTurn
|
|||
export const agentQueuedTurnReceiptSchema = enqueueAgentQueuedTurnSchema.extend({
|
||||
queuedTurnId: z.string().trim().min(1),
|
||||
status: z.enum(agentQueuedTurnStatuses),
|
||||
/** Effective generation boundary consumed by an admitted turn. This can
|
||||
* advance beyond the originally captured root as queued turns chain. */
|
||||
effectivePredecessorCreatedAt: z.number().int().nonnegative().optional(),
|
||||
/** Explicitly proves that this admission consumed no predecessor boundary. */
|
||||
rootPredecessor: z.literal(true).optional(),
|
||||
position: z.number().int().nonnegative().optional(),
|
||||
/** Immutable queue sequence retained as `revision` for wire compatibility. */
|
||||
revision: z.number().int().nonnegative(),
|
||||
failure: z
|
||||
.object({
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
File diff suppressed because it is too large
Load diff
|
|
@ -34,6 +34,9 @@ const terminalReceiptSchema = new Schema(
|
|||
admissionMode: { type: String, enum: ['warm', 'ordinary'] },
|
||||
generationId: { type: String, maxlength: 256 },
|
||||
generationCreatedAt: { type: Number, min: 0 },
|
||||
effectivePredecessorCreatedAt: { type: Number, min: 0 },
|
||||
lineagePredecessorId: { type: String, maxlength: 128 },
|
||||
rootPredecessor: { type: Boolean, enum: [true] },
|
||||
failure: { type: failureSchema },
|
||||
},
|
||||
{ _id: false },
|
||||
|
|
@ -57,6 +60,7 @@ const queuedTurnSchema: Schema<IAgentQueuedTurnDocument> = new Schema(
|
|||
sequence: { type: Number, min: 1 },
|
||||
reservationWriterId: { type: String, maxlength: 128 },
|
||||
activeSlot: { type: Number, min: 0, max: 99 },
|
||||
admissionSlot: { type: Boolean },
|
||||
status: {
|
||||
type: String,
|
||||
enum: ['reserving', 'queued', 'claimed', 'admitted', 'cancelled', 'dead'],
|
||||
|
|
@ -84,6 +88,8 @@ const queuedTurnSchema: Schema<IAgentQueuedTurnDocument> = new Schema(
|
|||
claimUntil: { type: Date },
|
||||
admissionStartedAt: { type: Date },
|
||||
admissionId: { type: String, maxlength: 128 },
|
||||
admissionEffectivePredecessorCreatedAt: { type: Number, min: 0 },
|
||||
admissionLineagePredecessorId: { type: String, maxlength: 128 },
|
||||
admissionProtocolVersion: { type: Number, enum: [2] },
|
||||
reconciliationAvailableAt: { type: Date },
|
||||
reconciliationClaimId: { type: String, maxlength: 128 },
|
||||
|
|
@ -115,6 +121,33 @@ queuedTurnSchema.index(
|
|||
partialFilterExpression: { activeSlot: { $exists: true } },
|
||||
},
|
||||
);
|
||||
queuedTurnSchema.index(
|
||||
{ tenantId: 1, user: 1, conversationId: 1, laneId: 1, status: 1 },
|
||||
{
|
||||
name: 'agent_queued_turn_claim_lane',
|
||||
unique: true,
|
||||
partialFilterExpression: { status: 'claimed' },
|
||||
},
|
||||
);
|
||||
queuedTurnSchema.index(
|
||||
{ tenantId: 1, user: 1, conversationId: 1, laneId: 1 },
|
||||
{
|
||||
name: 'agent_queued_turn_admission_started_lane',
|
||||
unique: true,
|
||||
/** `admissionStartedAt` is written by both legacy and current workers and
|
||||
* survives legacy `dead/ADMISSION_INDETERMINATE` quarantine. This is the
|
||||
* cross-version fence after a claim crosses the provider boundary. */
|
||||
partialFilterExpression: { admissionStartedAt: { $exists: true } },
|
||||
},
|
||||
);
|
||||
queuedTurnSchema.index(
|
||||
{ tenantId: 1, user: 1, conversationId: 1, laneId: 1, admissionSlot: 1 },
|
||||
{
|
||||
name: 'agent_queued_turn_admission_slot',
|
||||
unique: true,
|
||||
partialFilterExpression: { admissionSlot: true },
|
||||
},
|
||||
);
|
||||
queuedTurnSchema.index(
|
||||
{
|
||||
tenantId: 1,
|
||||
|
|
|
|||
|
|
@ -37,6 +37,12 @@ export interface AgentQueuedTurnTerminalReceipt {
|
|||
admissionMode?: 'warm' | 'ordinary';
|
||||
generationId?: string;
|
||||
generationCreatedAt?: number;
|
||||
/** Effective predecessor consumed by this admission after queued-turn chaining. */
|
||||
effectivePredecessorCreatedAt?: number;
|
||||
/** Durable lineage node: the root-message identity or predecessor queued-turn identity. */
|
||||
lineagePredecessorId?: string;
|
||||
/** This admission consumed no predecessor generation boundary. */
|
||||
rootPredecessor?: true;
|
||||
failure?: AgentQueuedTurnFailure;
|
||||
}
|
||||
|
||||
|
|
@ -57,6 +63,8 @@ export interface IAgentQueuedTurn {
|
|||
reservationWriterId?: string;
|
||||
/** Bounded active-lane capacity token. Present only while queued/claimed. */
|
||||
activeSlot?: number;
|
||||
/** Unique durable admission-order reservation for one conversation lane. */
|
||||
admissionSlot?: boolean;
|
||||
status: AgentQueuedTurnStatus;
|
||||
priority: boolean;
|
||||
text: string;
|
||||
|
|
@ -75,6 +83,10 @@ export interface IAgentQueuedTurn {
|
|||
/** Durable proof that ordinary admission may have crossed the HTTP boundary. */
|
||||
admissionStartedAt?: Date;
|
||||
admissionId?: string;
|
||||
/** Effective predecessor durably fenced before provider admission begins. */
|
||||
admissionEffectivePredecessorCreatedAt?: number;
|
||||
/** Durable lineage node fenced with the effective predecessor. */
|
||||
admissionLineagePredecessorId?: string;
|
||||
/** Version 2 requires accepted and deduplicated execution responses to prove
|
||||
* the exact post-invocation source receipt. */
|
||||
admissionProtocolVersion?: 2;
|
||||
|
|
@ -148,6 +160,7 @@ export type AgentQueuedTurnActiveRecord = Pick<
|
|||
|
||||
export interface AgentQueuedTurnClaim extends AgentQueuedTurnRecord {
|
||||
status: 'claimed';
|
||||
admissionSlot: true;
|
||||
claimId: string;
|
||||
claimBy: string;
|
||||
claimUntil: Date;
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue