CERBERUS/LibreChat
1
0
Fork 0
mirror of https://github.com/danny-avila/LibreChat.git synced 2026-08-04 14:57:42 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

Merge c0e167b3d4 into b807292997

Browse source
This commit is contained in:
Danny Avila 2026-08-04 14:27:37 +02:00 • committed by GitHub
commit 7cc7e23f0b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 3133 additions and 284 deletions
Download patch file Download diff file Expand all files Collapse all files

240
client/src/components/Chat/Input/PendingSteerChips.tsx
View file

@ -1,221 +1,25 @@
import { memo, useMemo, useRef, useState, useCallback } from 'react';
import { memo, useMemo } from 'react';
import { useAtomValue } from 'jotai';
import { useRecoilValue } from 'recoil';
import { useToastContext } from '@librechat/client';
import { X, Zap, Send, Clock, Pencil, Trash2, Paperclip, RotateCcw } from 'lucide-react';
import { X, Zap, Clock, Pencil, RotateCcw } from 'lucide-react';
import type { TMessage } from 'librechat-data-provider';
import type { SteeringControls, QueuedMessageContext } from '~/hooks/Chat/useSteering';
import type { PendingSteer, QueuedMessage } from '~/store/families';
import type { RestoreToComposer } from './InFlightSteers';
import type { PendingSteer } from '~/store/families';
import type { MenuEntry } from './SteerMenu';
import {
RowMenu,
ICON_BTN_CLASS,
PRIMARY_BTN_CLASS,
EscalateNowButton,
useDefaultToggleEntry,
useInterruptToggleEntry,
} from './SteerMenu';
import { QueuedRow, QueuedOutbox, QueueSendingBanner, ROW_CLASS } from './QueuedOutbox';
import { escalatingSteerFamily } from '~/store/steer';
import { useLocalize } from '~/hooks';
import { cn } from '~/utils';
import store from '~/store';
const ROW_CLASS =
'flex w-full items-center gap-2 rounded-xl border border-border-light bg-surface-secondary px-3 py-2 text-sm text-text-primary';
function AttachmentCount({ count, label }: { count: number; label: string }) {
if (count === 0) {
return null;
}
return (
<span className="flex shrink-0 items-center gap-0.5 text-xs text-text-secondary">
<Paperclip className="h-3.5 w-3.5" aria-hidden="true" />
{count}
<span className="sr-only">{label}</span>
</span>
);
}
function QueuedRow({
message,
steering,
conversationId,
interruptPending,
onEditToComposer,
onRestoreToComposer,
}: {
message: QueuedMessage;
steering: SteeringControls;
conversationId: string;
interruptPending: boolean;
onEditToComposer: (
text: string,
files?: TMessage['files'],
context?: QueuedMessageContext,
) => void;
onRestoreToComposer: RestoreToComposer;
}) {
const localize = useLocalize();
const { showToast } = useToastContext();
const toggleEntry = useDefaultToggleEntry(steering);
const interruptToggle = useInterruptToggleEntry();
const fileCount = message.files?.length ?? 0;
const isRecovered = message.recoverySteerId != null;
const actionPendingRef = useRef(false);
const [actionPending, setActionPending] = useState(false);
/** A recovered item has a replayable parked source. Edit/remove must first
* cancel that source by receipt; local-only rows settle synchronously through
* the same control. The ref closes the pre-render double-click window. */
const afterDiscard = useCallback(
(action: () => boolean) => {
if (actionPendingRef.current) {
return;
}
actionPendingRef.current = true;
setActionPending(true);
void (async () => {
let discarded = false;
try {
discarded = await steering.discardQueued(message);
} catch {
// The steering hook reports request failures and leaves the row in
// place. Keep this guard for test/custom control implementations.
}
if (!discarded) {
actionPendingRef.current = false;
setActionPending(false);
return;
}
if (!action()) {
actionPendingRef.current = false;
setActionPending(false);
}
})();
},
[message, steering],
);
// A recovered item is consumed atomically only when it starts a normal
// generation. Re-steering it would leave or duplicate the parked source;
// Edit/remove are safe because `afterDiscard` tombstones that source first.
const canSteerNow = steering.duringRunActive && steering.canSteer && !isRecovered;
const showPrimary = canSteerNow || (!steering.duringRunActive && steering.canSendQueuedNow);
/** `canSteer` is defined as false while paused on approval, but the
* escalation control must stay visible-and-disabled there — hiding it
* during the pause is exactly the discoverability gap this button fixes. */
const showEscalate =
!isRecovered && (steering.pausedOnApproval || (steering.duringRunActive && steering.canSteer));
const entries: MenuEntry[] = [
{
key: 'edit',
label: localize('com_ui_edit_message'),
icon: <Pencil className="h-4 w-4" aria-hidden="true" />,
disabled: actionPending,
onClick: () => {
const context = { quotes: message.quotes, manualSkills: message.manualSkills };
if (!isRecovered) {
steering.removeQueued(message.id);
onEditToComposer(message.text, message.files, {
quotes: message.quotes,
manualSkills: message.manualSkills,
});
return;
}
afterDiscard(() => {
const restored = onRestoreToComposer(
message.text,
message.files,
context,
conversationId,
);
if (!restored) {
showToast({ message: localize('com_ui_steer_edit_queued'), status: 'info' });
return false;
}
steering.removeQueued(message.id);
return true;
});
},
},
];
const preferences: MenuEntry[] = [toggleEntry, interruptToggle];
return (
<div role="listitem" className={ROW_CLASS} data-testid="queued-message-row">
<Clock className="h-4 w-4 shrink-0 text-cyan-500" aria-hidden="true" />
<span className="min-w-0 flex-1 truncate" title={message.text}>
{message.text}
</span>
<AttachmentCount
count={fileCount}
label={localize('com_ui_queued_attachment_count', { 0: String(fileCount) })}
/>
{showPrimary && (
<button
type="button"
className={PRIMARY_BTN_CLASS}
disabled={actionPending}
onClick={() => steering.sendQueuedNow(message)}
>
{canSteerNow ? (
<>
<Zap className="h-4 w-4 text-amber-500" aria-hidden="true" />
{localize('com_ui_steer')}
</>
) : (
<>
<Send className="h-4 w-4" aria-hidden="true" />
{localize('com_ui_send_now')}
</>
)}
</button>
)}
{showEscalate && (
<EscalateNowButton
surface="queued"
messageText={message.text}
disabled={steering.pausedOnApproval || interruptPending || actionPending}
onClick={() => steering.sendQueuedNow(message, { preempt: true })}
/>
)}
<button
type="button"
aria-label={localize('com_ui_remove_queued')}
disabled={actionPending}
onClick={() => {
const remove = () => {
/* Same safety net as the in-flight cancel: once removal is safely
* settled, return the words to the composer when it is free (the
* gated restore refuses rather than clobber a draft). */
onRestoreToComposer(
message.text,
message.files,
{ quotes: message.quotes, manualSkills: message.manualSkills },
conversationId,
);
steering.removeQueued(message.id);
return true;
};
if (!isRecovered) {
remove();
return;
}
afterDiscard(remove);
}}
className={ICON_BTN_CLASS}
>
<Trash2 className="h-4 w-4" aria-hidden="true" />
</button>
<RowMenu
label={localize('com_ui_more_options')}
entries={entries}
preferences={preferences}
/>
</div>
);
}
function FailedSteerRow({
steer,
steering,
@ -332,8 +136,11 @@ function FailedSteerRow({
* (In-flight steers read as messages, not controls — `InFlightSteers` renders
* them as bubbles anchored above the composer box.)
* - Failed steer rows (Zap, red): the POST failed, so the text never entered
* the thread — kept recoverable with retry / edit / queue actions.
* - Queued rows (Clock): client-side follow-ups auto-sent after the run.
* the thread — kept recoverable with retry / edit / queue actions. They stay
* OUTSIDE the queued group: a failure is an action item, so no disclosure
* may ever hide one.
* - Queued rows (Clock): client-side follow-ups auto-sent after the run. A lone
* one stays a plain chip; two or more collapse into `QueuedOutbox`.
*/
function PendingSteerChips({
conversationId,
@ -353,6 +160,7 @@ function PendingSteerChips({
const localize = useLocalize();
const steers = useRecoilValue(store.pendingSteersByConvoId(conversationId));
const queued = useRecoilValue(store.queuedMessagesByConvoId(steering.queueKey));
const drainHold = useRecoilValue(store.queueDrainHoldByConvoId(steering.queueKey));
const failedSteers = useMemo(() => steers.filter((steer) => steer.status === 'failed'), [steers]);
/** Only one interrupt can be in flight: a second preempt while one is
* unresolved would arm a second seal, so escalation buttons disable. The
@ -383,17 +191,35 @@ function PendingSteerChips({
onEditToComposer={onEditToComposer}
/>
))}
{queued.map((message) => (
{drainHold != null && drainHold.status == null && queued.length > 0 && (
<QueueSendingBanner
count={queued.length}
dueAt={drainHold.dueAt}
onUndo={steering.cancelQueueDrain}
/>
)}
{queued.length === 1 && (
<QueuedRow
key={message.id}
message={message}
key={queued[0].id}
message={queued[0]}
steering={steering}
conversationId={conversationId}
interruptPending={interruptPending}
onEditToComposer={onEditToComposer}
canBump={false}
sendPending={drainHold != null && drainHold.status == null}
onRestoreToComposer={onRestoreToComposer}
/>
))}
)}
{queued.length > 1 && (
<QueuedOutbox
queued={queued}
steering={steering}
conversationId={conversationId}
interruptPending={interruptPending}
sendPending={drainHold != null && drainHold.status == null}
onRestoreToComposer={onRestoreToComposer}
/>
)}
</div>
);
}

656
client/src/components/Chat/Input/QueuedOutbox.tsx Normal file
View file

@ -0,0 +1,656 @@
import { memo, useCallback, useEffect, useMemo, useRef, useState } from 'react';
import { useAtom } from 'jotai';
import { useToastContext } from '@librechat/client';
import {
Zap,
Send,
Clock,
Merge,
Pencil,
Trash2,
ChevronUp,
ChevronsUp,
ChevronDown,
Paperclip,
} from 'lucide-react';
import type { SteeringControls } from '~/hooks/Chat/useSteering';
import type { RestoreToComposer } from './InFlightSteers';
import type { QueuedMessage } from '~/store/families';
import type { MenuEntry } from './SteerMenu';
import {
RowMenu,
ICON_BTN_CLASS,
PRIMARY_BTN_CLASS,
EscalateNowButton,
useDefaultToggleEntry,
useInterruptToggleEntry,
} from './SteerMenu';
import { isMergeableQueuedMessage, isSendableQueuedMessage } from '~/utils';
import { queueExpandedFamily } from '~/store/steer';
import { useLocalize } from '~/hooks';
import { cn } from '~/utils';
export const ROW_CLASS =
'flex w-full items-center gap-2 rounded-xl border border-border-light bg-surface-secondary px-3 py-2 text-sm text-text-primary';
/** Enough to show a merged row's paragraphs without the row eating the box. */
const EDITOR_MAX_ROWS = 4;
/** Depth cue for the collapsed stack. Kept INSIDE the row's own footprint (the
* wrapper reserves the peek height) because the composer box clips overflow. */
const PEEK_CLASS =
'pointer-events-none absolute h-full rounded-xl border border-border-light bg-surface-tertiary';
function AttachmentCount({ count, label }: { count: number; label: string }) {
if (count === 0) {
return null;
}
return (
<span className="flex shrink-0 items-center gap-0.5 text-xs text-text-secondary">
<Paperclip className="h-3.5 w-3.5" aria-hidden="true" />
{count}
<span className="sr-only">{label}</span>
</span>
);
}
function QueuedRowBase({
message,
steering,
conversationId,
interruptPending,
canBump,
sendPending,
onRestoreToComposer,
}: {
message: QueuedMessage;
steering: SteeringControls;
conversationId: string;
interruptPending: boolean;
/** False for the front row (already next) and for a lone queued message. */
canBump: boolean;
/** An automatic send is withheld and about to fire for this queue. */
sendPending: boolean;
onRestoreToComposer: RestoreToComposer;
}) {
const localize = useLocalize();
const { showToast } = useToastContext();
const toggleEntry = useDefaultToggleEntry(steering);
const interruptToggle = useInterruptToggleEntry();
const fileCount = message.files?.length ?? 0;
const isRecovered = message.recoverySteerId != null;
/** Not while a send is pending: the words are already on their way, so an
* editor that opened and instantly settled would read as a glitch. Undo
* first, then edit. */
const editable = isMergeableQueuedMessage(message) && !sendPending;
const actionPendingRef = useRef(false);
const [actionPending, setActionPending] = useState(false);
const [draft, setDraft] = useState<string | null>(null);
const editing = draft != null;
const editorRef = useRef<HTMLTextAreaElement>(null);
/** The pre-edit words, so Escape can put them back after the write-through
* below has already replaced them. */
const originalRef = useRef<string>(message.text);
/** A recovered item has a replayable parked source. Edit/remove must first
* cancel that source by receipt; local-only rows settle synchronously through
* the same control. The ref closes the pre-render double-click window. */
const afterDiscard = useCallback(
(action: () => boolean) => {
if (actionPendingRef.current) {
return;
}
actionPendingRef.current = true;
setActionPending(true);
void (async () => {
let discarded = false;
try {
discarded = await steering.discardQueued(message);
} catch {
// The steering hook reports request failures and leaves the row in
// place. Keep this guard for test/custom control implementations.
}
if (!discarded) {
actionPendingRef.current = false;
setActionPending(false);
return;
}
if (!action()) {
actionPendingRef.current = false;
setActionPending(false);
}
})();
},
[message, steering],
);
// A recovered item is consumed atomically only when it starts a normal
// generation. Re-steering it would leave or duplicate the parked source;
// Edit/remove are safe because `afterDiscard` tombstones that source first.
const canSteerNow = steering.duringRunActive && steering.canSteer && !isRecovered;
const showPrimary = canSteerNow || (!steering.duringRunActive && steering.canSendQueuedNow);
/** `canSteer` is defined as false while paused on approval, but the
* escalation control must stay visible-and-disabled there — hiding it
* during the pause is exactly the discoverability gap this button fixes. */
const showEscalate =
!isRecovered && (steering.pausedOnApproval || (steering.duringRunActive && steering.canSteer));
/** The queue holds exactly what is typed, so "nothing to send" is visible in
* the row itself — no shared claim to publish, release, or leak. */
const sendable = isSendableQueuedMessage(message);
/** Focus follows the explicit Edit action rather than mount, so the row can
* never steal focus from the composer on a re-render. */
useEffect(() => {
if (editing) {
editorRef.current?.focus();
}
}, [editing]);
const draftRef = useRef<string | null>(null);
draftRef.current = draft;
const beginEdit = useCallback(() => {
originalRef.current = message.text;
setDraft(message.text);
}, [message.text]);
/**
* Every keystroke writes through to the queue. The row is a living draft and
* the senders — the drain, Send now, the escalate shortcut — all read the
* atom, so anything still only local is a message that sends as its old
* text. Local state stays for display alone, because an emptied field has to
* survive on screen while the queue keeps the last non-empty words.
*
* The cost is one small-subtree render per keystroke (this stack, never the
* thread, which does not subscribe to the queue).
*/
const editDraft = useCallback(
(value: string) => {
setDraft(value);
steering.updateQueuedText(message.id, value);
},
[message.id, steering],
);
/** Leaving the editor settles the row: trimmed if there are words, and back to
* the pre-edit text if there are not. That is what keeps a blank row from
* ever outliving its editor, which is in turn why every reader can trust
* `isSendableQueuedMessage` on a resting queue. */
const closeEdit = useCallback(() => {
const typed = draftRef.current;
if (typed == null) {
return;
}
steering.updateQueuedText(
message.id,
typed.trim().length === 0 ? originalRef.current : typed.trim(),
);
setDraft(null);
}, [message.id, steering]);
const abandonEdit = useCallback(() => {
steering.updateQueuedText(message.id, originalRef.current);
setDraft(null);
}, [message.id, steering]);
const closeEditRef = useRef(closeEdit);
closeEditRef.current = closeEdit;
/** An editor removed by a remount fires no blur, so it settles on the way out
* — otherwise a blank row could survive the group collapsing around it. */
useEffect(() => () => closeEditRef.current(), []);
/** A row whose send is pending closes its editor: the words are already
* written, and a countdown is no moment to keep typing into. An EMPTY editor
* is resolved the way Escape resolves it — the blank was never written, so
* closing it alone would leave the queue holding words the screen no longer
* shows, and the drain would send those. */
useEffect(() => {
if (sendPending) {
closeEditRef.current();
}
}, [sendPending]);
/** An ordinary row is a living draft: it is rewritten in place. A recovered
* row's words are bound to a parked server source matched by exact text, so
* its Edit keeps the existing discard-then-hand-to-composer path. */
const entries: MenuEntry[] = [
{
key: 'edit',
label: localize(editable ? 'com_ui_queue_edit_inline' : 'com_ui_edit_message'),
icon: <Pencil className="h-4 w-4" aria-hidden="true" />,
disabled: actionPending,
onClick: () => {
if (editable) {
beginEdit();
return;
}
const context = { quotes: message.quotes, manualSkills: message.manualSkills };
afterDiscard(() => {
const restored = onRestoreToComposer(
message.text,
message.files,
context,
conversationId,
);
if (!restored) {
showToast({ message: localize('com_ui_steer_edit_queued'), status: 'info' });
return false;
}
steering.removeQueued(message.id);
return true;
});
},
},
];
const preferences: MenuEntry[] = [toggleEntry, interruptToggle];
return (
<div role="listitem" className={ROW_CLASS} data-testid="queued-message-row">
<Clock className="h-4 w-4 shrink-0 text-cyan-500" aria-hidden="true" />
{editing ? (
/* A textarea, not an input: merged rows carry paragraph breaks and an
* input silently flattens them on the first keystroke. Enter commits
* and Shift+Enter adds a line, matching the composer. */
<textarea
ref={editorRef}
value={draft}
rows={Math.min(EDITOR_MAX_ROWS, draft.split('\n').length)}
aria-label={localize('com_ui_queue_edit_inline')}
data-testid="queued-message-edit"
className="min-w-0 flex-1 resize-none bg-transparent text-sm text-text-primary outline-none"
onChange={(event) => editDraft(event.target.value)}
onBlur={closeEdit}
onKeyDown={(event) => {
/** An IME candidate confirmation arrives as an unshifted Enter
* while composition is still active; committing there would save
* half-typed text. Same guard the composer and DynamicTags use. */
if (event.nativeEvent.isComposing || event.keyCode === 229) {
return;
}
if (event.key === 'Enter' && !event.shiftKey) {
event.preventDefault();
closeEdit();
return;
}
if (event.key === 'Escape') {
event.preventDefault();
abandonEdit();
}
}}
/>
) : (
<span
className={cn('min-w-0 flex-1 truncate', editable && 'cursor-text')}
title={editable ? localize('com_ui_queue_edit_inline') : message.text}
onClick={editable ? beginEdit : undefined}
>
{message.text}
</span>
)}
<AttachmentCount
count={fileCount}
label={localize('com_ui_queued_attachment_count', { 0: String(fileCount) })}
/>
{canBump && !editing && (
<button
type="button"
aria-label={localize('com_ui_queue_send_next')}
title={localize('com_ui_queue_send_next')}
data-testid="queued-send-next"
disabled={actionPending}
onClick={() => steering.bumpQueued(message.id)}
className={ICON_BTN_CLASS}
>
<ChevronsUp className="h-4 w-4" aria-hidden="true" />
</button>
)}
{showPrimary && (
<button
type="button"
className={PRIMARY_BTN_CLASS}
disabled={actionPending || !sendable}
title={sendable ? undefined : localize('com_ui_queue_edit_empty')}
onClick={() => steering.sendQueuedNow(message)}
>
{canSteerNow ? (
<>
<Zap className="h-4 w-4 text-amber-500" aria-hidden="true" />
{localize('com_ui_steer')}
</>
) : (
<>
<Send className="h-4 w-4" aria-hidden="true" />
{localize('com_ui_send_now')}
</>
)}
</button>
)}
{showEscalate && (
<EscalateNowButton
surface="queued"
messageText={message.text}
disabled={steering.pausedOnApproval || interruptPending || actionPending || !sendable}
onClick={() => steering.sendQueuedNow(message, { preempt: true })}
/>
)}
<button
type="button"
aria-label={localize('com_ui_remove_queued')}
disabled={actionPending}
onClick={() => {
const remove = () => {
/* Same safety net as the in-flight cancel: once removal is safely
* settled, return the words to the composer when it is free (the
* gated restore refuses rather than clobber a draft).
*
* Skipped when the editor was left empty: the queue still holds the
* pre-edit words, so handing them back would resurrect text the
* user had visibly deleted. Emptying a row and then removing it
* reads as "delete this", and there is nothing to return. */
if (sendable) {
onRestoreToComposer(
message.text,
message.files,
{ quotes: message.quotes, manualSkills: message.manualSkills },
conversationId,
);
}
steering.removeQueued(message.id);
return true;
};
if (!isRecovered) {
remove();
return;
}
afterDiscard(remove);
}}
className={ICON_BTN_CLASS}
>
<Trash2 className="h-4 w-4" aria-hidden="true" />
</button>
<RowMenu
label={localize('com_ui_more_options')}
entries={entries}
preferences={preferences}
/>
</div>
);
}
/** Rows carry local edit state and a sibling's keystroke must not re-render
* the whole stack; the in-flight bubbles are memoized for the same reason. */
export const QueuedRow = memo(QueuedRowBase);
/**
* The withheld automatic send. Progress is announced rather than merely drawn,
* and the bar is a CSS animation so the window costs no re-renders while it
* runs — this surface sits above a streaming thread.
*/
function QueueSendingBannerBase({
count,
dueAt,
onUndo,
}: {
count: number;
dueAt: number;
onUndo: () => void;
}) {
const localize = useLocalize();
const remaining = Math.max(0, dueAt - Date.now());
return (
<div
role="listitem"
className="flex w-full items-center gap-2 rounded-xl border border-border-light bg-surface-tertiary px-3 py-2 text-sm text-text-primary"
data-testid="queue-sending-banner"
>
<Send className="h-4 w-4 shrink-0 text-text-secondary" aria-hidden="true" />
<span className="min-w-0 flex-1 truncate" aria-live="polite">
{localize('com_ui_queue_sending', { 0: String(count) })}
</span>
<span
aria-hidden="true"
className="h-1 w-10 shrink-0 overflow-hidden rounded-full bg-surface-secondary"
>
<span
className="block h-full w-full origin-left bg-text-secondary motion-safe:animate-queue-undo-grace"
style={{ animationDuration: `${remaining}ms` }}
/>
</span>
<button
type="button"
className={PRIMARY_BTN_CLASS}
data-testid="queue-undo-send"
onClick={onUndo}
>
{localize('com_ui_queue_undo_send')}
</button>
</div>
);
}
export const QueueSendingBanner = memo(QueueSendingBannerBase);
/**
* Two or more waiting messages read as one outbox instead of a growing stack:
* collapsed, the group is a single row (the next message to send) over layered
* card edges, so the composer stops inflating with queue depth. Management
* lives inside the expansion, where the rows keep every per-row affordance.
*/
function QueuedOutboxBase({
queued,
steering,
conversationId,
interruptPending,
sendPending,
onRestoreToComposer,
}: {
queued: QueuedMessage[];
steering: SteeringControls;
conversationId: string;
interruptPending: boolean;
sendPending: boolean;
onRestoreToComposer: RestoreToComposer;
}) {
const localize = useLocalize();
const { showToast } = useToastContext();
const [expanded, setExpanded] = useAtom(queueExpandedFamily(steering.queueKey));
/** Every gate below reads the rows this component already holds. That is the
* whole point of recording blank text instead of refusing it: sendability is
* a property of the data, so there is no shared claim to keep in step. */
const allSendable = useMemo(() => queued.every(isSendableQueuedMessage), [queued]);
const mergeable = useMemo(() => queued.every(isMergeableQueuedMessage), [queued]);
/** Folding reads the queue, so an unresolved empty edit would carry the words
* the user just deleted into the merged message. Same standdown the senders
* use — the row's own controls, the drain, and the shortcut proxy. */
const mergeBlockedReason = (() => {
if (!mergeable) {
return localize('com_ui_queue_merge_blocked');
}
return allSendable ? undefined : localize('com_ui_queue_edit_empty');
})();
/** Clear all folds the queue the same way Merge does, so an unresolved empty
* edit would hand words the user deleted back to the composer. Uniform with
* every other reader rather than an exception worth remembering. */
const clearBlockedReason = allSendable ? undefined : localize('com_ui_queue_edit_empty');
/** The shortcut's promise is the NEWEST waiting message, which is not the
* last array slot once a promotion has reordered the queue — so pick by
* stamp. Recovery-bound rows are skipped because steering refuses them
* mid-run, exactly as the expanded rows omit their escalation controls. */
const escalatable = useMemo(() => {
let newest: QueuedMessage | undefined;
for (const item of queued) {
if (item.recoverySteerId != null) {
continue;
}
if (newest == null || item.createdAt > newest.createdAt) {
newest = item;
}
}
return newest;
}, [queued]);
const [next] = queued;
/** Index of the first row a promotion could overtake. Computed once: the
* per-row prefix scan it replaces allocated and walked a slice for every
* row, and the write-through editor re-renders this list on every
* keystroke. */
const firstOvertakeable = useMemo(
() => queued.findIndex((item) => item.priority !== true),
[queued],
);
const clearAll = useCallback(() => {
void (async () => {
const cleared = await steering.clearQueued();
if (cleared == null) {
return;
}
const restored = onRestoreToComposer(
cleared.text,
cleared.files,
{ quotes: cleared.quotes, manualSkills: cleared.manualSkills },
conversationId,
);
/** The gated restore refuses rather than clobber a draft the user has
* since staged. Hand the words back to the queue instead of dropping
* them — one row now, since they were folded on the way out. The ITEM
* goes back whole: rebuilding one from parts dropped its predecessor
* fence once and its interrupt tier once, and the next field would be
* just as quiet. */
if (!restored) {
steering.requeueCleared([cleared]);
showToast({ message: localize('com_ui_steer_edit_queued'), status: 'info' });
}
})();
}, [conversationId, localize, onRestoreToComposer, showToast, steering]);
/* One wrapper for both states, so the disclosure button keeps its position
* in the tree: remounting it on toggle would drop keyboard focus mid-use.
* It is the parent list's item — collapsed, the group IS the only item; the
* rows become a nested list when expanded. */
return (
<div
role="listitem"
className={cn('relative flex flex-col gap-1.5', !expanded && 'pb-1.5')}
data-testid="queue-group"
>
{!expanded && (
<>
<span aria-hidden="true" className={cn(PEEK_CLASS, 'inset-x-3 top-3 opacity-40')} />
<span aria-hidden="true" className={cn(PEEK_CLASS, 'inset-x-1.5 top-1.5 opacity-70')} />
</>
)}
<button
type="button"
aria-expanded={expanded}
/* Deliberately unlabelled: the accessible name comes from the count and
* next-up preview inside, which is the only description of the queue
* while the rows are unmounted. `aria-expanded` carries the show/hide
* state, so a label would only overwrite that summary. */
data-testid="queue-group-toggle"
onClick={() => setExpanded((prev) => !prev)}
className={cn(ROW_CLASS, 'relative text-left hover:bg-surface-hover')}
>
<Clock className="h-4 w-4 shrink-0 text-cyan-500" aria-hidden="true" />
<span className="shrink-0 font-medium">
{localize('com_ui_queue_count', { 0: String(queued.length) })}
</span>
{!expanded && (
<span className="min-w-0 flex-1 truncate text-text-secondary">
{localize('com_ui_queue_next_up', { 0: next.text })}
</span>
)}
<span className="flex-1" />
{expanded ? (
<ChevronUp className="h-4 w-4 shrink-0 text-text-secondary" aria-hidden="true" />
) : (
<ChevronDown className="h-4 w-4 shrink-0 text-text-secondary" aria-hidden="true" />
)}
</button>
{expanded && (
<div
role="list"
/* Named by its own count rather than repeating the outer stack's
* label, which would nest two identically named lists. */
aria-label={localize('com_ui_queue_count', { 0: String(queued.length) })}
data-testid="queue-rows"
/* The composer box is `overflow-hidden`, so a deep queue would clip
* rows with no way to reach them. Cap the ROWS only: the disclosure
* above and the actions below must stay put. Same ceiling the
* in-flight overlay uses. */
className="flex max-h-[35vh] flex-col gap-1.5 overflow-y-auto"
>
{queued.map((message, position) => (
<QueuedRow
key={message.id}
message={message}
steering={steering}
conversationId={conversationId}
interruptPending={interruptPending}
/** A promotion lifts a row to the top of the promotions tier,
* which is still below every interrupt — so offering it when
* only interrupts sit ahead would advertise a no-op. */
canBump={firstOvertakeable !== -1 && position > firstOvertakeable}
sendPending={sendPending}
onRestoreToComposer={onRestoreToComposer}
/>
))}
</div>
)}
{expanded && (
<div className="flex items-center gap-2 px-1 pb-0.5">
<button
type="button"
className={PRIMARY_BTN_CLASS}
data-testid="queue-merge"
disabled={mergeBlockedReason != null}
title={mergeBlockedReason}
onClick={() => steering.mergeQueued()}
>
<Merge className="h-4 w-4" aria-hidden="true" />
{localize('com_ui_queue_merge')}
</button>
<span className="flex-1" />
<button
type="button"
className={PRIMARY_BTN_CLASS}
data-testid="queue-clear-all"
disabled={clearBlockedReason != null}
title={clearBlockedReason}
onClick={clearAll}
>
<Trash2 className="h-4 w-4" aria-hidden="true" />
{localize('com_ui_queue_clear_all')}
</button>
</div>
)}
{/* Keyboard parity. The shortcut prefers a hovered or focused control and
* otherwise takes the LAST queued one in the document — which is the
* last row in DRAIN order, not the newest message, once a promotion has
* reordered the queue. This offscreen control sits last in both states
* so the fallback resolves to the newest eligible message either way,
* while the visible per-row buttons stay clickable. */}
{escalatable != null &&
!interruptPending &&
steering.duringRunActive &&
steering.canSteer && (
<button
type="button"
/** Invoked by the shortcut through its data attribute, so it needs
* no sequential focus — and an invisible tab stop with no focus
* presentation is worse than none. */
tabIndex={-1}
/** Its target's editor is empty, so the row still holds words the
* user deleted. Disabled rather than absent: the shortcut skips
* unavailable controls, so it falls through instead of sending
* them. */
disabled={!isSendableQueuedMessage(escalatable)}
className="sr-only"
data-escalate-steer="queued"
data-testid="queued-escalate-newest"
aria-label={localize('com_ui_queue_escalate_newest')}
onClick={() => steering.sendQueuedNow(escalatable, { preempt: true })}
/>
)}
</div>
);
}
export const QueuedOutbox = memo(QueuedOutboxBase);

751
client/src/components/Chat/Input/__tests__/PendingSteerChips.test.tsx
View file

@ -3,9 +3,9 @@ import { getDefaultStore } from 'jotai';
import userEvent from '@testing-library/user-event';
import { RecoilRoot, useRecoilValue, useSetRecoilState } from 'recoil';
import { render, screen, fireEvent, act, waitFor } from '@testing-library/react';
import type { PendingSteer, QueuedMessage } from '~/store/families';
import type { PendingSteer, QueuedMessage, QueueDrainHold } from '~/store/families';
import type { SteeringControls } from '~/hooks/Chat/useSteering';
import { escalatingSteerFamily } from '~/store/steer';
import { escalatingSteerFamily, queueExpandedFamily } from '~/store/steer';
import PendingSteerChips from '../PendingSteerChips';
import store from '~/store';
@ -37,6 +37,13 @@ function QueueState() {
return <output data-testid="queue-state">{JSON.stringify(queue)}</output>;
}
let setHoldForTest: ((value: QueueDrainHold | null) => void) | undefined;
function HoldState() {
setHoldForTest = useSetRecoilState(store.queueDrainHoldByConvoId(CONVO_ID));
return null;
}
function PreferenceState() {
const defaultAction = useRecoilValue(store.duringRunDefaultAction);
const interrupts = useRecoilValue(store.steerInterruptsByDefault);
@ -64,6 +71,14 @@ const steeringStub = (overrides: Partial<SteeringControls> = {}) =>
...overrides,
}) as unknown as SteeringControls;
/** The outbox disclosure is jotai state on the module-global default store, so
* it outlives a render and would otherwise leak between tests in this file. */
beforeEach(() => {
act(() => {
getDefaultStore().set(queueExpandedFamily(CONVO_ID), false);
});
});
function renderChips(
queued: QueuedMessage[],
options?: {
@ -91,6 +106,7 @@ function renderChips(
/>
<PreferenceState />
<QueueState />
<HoldState />
</RecoilRoot>,
);
}
@ -456,6 +472,9 @@ describe('PendingSteerChips — queued interrupt-now', () => {
renderChips([queuedMessage, { id: 'q2', text: 'urgent two', createdAt: 2 }], {
steering: liveRun,
});
/* Two or more queued messages collapse into the outbox group; the per-row
* escalation controls live in the expansion. */
fireEvent.click(screen.getByTestId('queue-group-toggle'));
const [first, second] = screen.getAllByTestId('queued-interrupt-now');
expect(first).not.toHaveAttribute('aria-keyshortcuts');
@ -478,6 +497,7 @@ describe('PendingSteerChips — queued interrupt-now', () => {
renderChips([queuedMessage, { id: 'q2', text: 'urgent two', createdAt: 2 }], {
steering: liveRun,
});
fireEvent.click(screen.getByTestId('queue-group-toggle'));
expect(
screen.getByRole('button', { name: 'com_ui_interrupt_steer_now: urgent one' }),
@ -541,3 +561,730 @@ describe('PendingSteerChips — queued interrupt-now', () => {
},
);
});
const mockBumpQueued = jest.fn();
/** Mirrors the real writer: records what is typed, blank included, so the rows
* under test see the same queue the app would. */
const mockUpdateQueuedText = jest.fn((id: string, text: string) => {
updateQueueForTest?.((current) =>
current.map((item) => (item.id === id ? { ...item, text } : item)),
);
return true;
});
const mockMergeQueued = jest.fn(() => true);
const mockCancelQueueDrain = jest.fn();
const mockEnqueue = jest.fn();
const mockRequeueCleared = jest.fn();
let mockClearQueued = jest.fn(async (): Promise<QueuedMessage | null> => null);
const outboxSteering = (overrides: Partial<SteeringControls> = {}) => ({
bumpQueued: mockBumpQueued,
updateQueuedText: mockUpdateQueuedText,
mergeQueued: mockMergeQueued,
clearQueued: mockClearQueued,
cancelQueueDrain: mockCancelQueueDrain,
enqueue: mockEnqueue,
requeueCleared: mockRequeueCleared,
...overrides,
});
const twoQueued: QueuedMessage[] = [
{ id: 'q1', text: 'first thought', createdAt: 1 },
{ id: 'q2', text: 'second thought', createdAt: 2 },
];
describe('PendingSteerChips — queued outbox group', () => {
beforeEach(() => {
jest.clearAllMocks();
mockClearQueued = jest.fn(async (): Promise<QueuedMessage | null> => null);
});
it('leaves a lone queued message as a plain chip', () => {
renderChips([twoQueued[0]], { steering: outboxSteering() });
expect(screen.queryByTestId('queue-group')).toBeNull();
expect(screen.getAllByTestId('queued-message-row')).toHaveLength(1);
});
it('collapses two or more into one row showing the count and what sends next', () => {
renderChips(twoQueued, { steering: outboxSteering() });
expect(screen.getByTestId('queue-group')).toBeInTheDocument();
expect(screen.getByTestId('queue-group-toggle')).toHaveAttribute('aria-expanded', 'false');
expect(screen.getByText('com_ui_queue_count')).toBeInTheDocument();
expect(screen.getByText('com_ui_queue_next_up')).toBeInTheDocument();
// The footprint is constant: the rows themselves are not mounted.
expect(screen.queryByTestId('queued-message-row')).toBeNull();
});
it('expands to the full managed list and back', () => {
renderChips(twoQueued, { steering: outboxSteering() });
const toggle = screen.getByTestId('queue-group-toggle');
fireEvent.click(toggle);
expect(toggle).toHaveAttribute('aria-expanded', 'true');
expect(screen.getAllByTestId('queued-message-row')).toHaveLength(2);
fireEvent.click(toggle);
expect(screen.queryByTestId('queued-message-row')).toBeNull();
});
/** The escalate shortcut clicks the LAST queued control in the document, and
* collapsing unmounts the rows — so the newest message keeps one here. */
it('keeps the escalate shortcut pointed at the newest message while collapsed', () => {
renderChips(twoQueued, { steering: outboxSteering({ duringRunActive: true, canSteer: true }) });
const controls = document.querySelectorAll('[data-escalate-steer="queued"]');
expect(controls).toHaveLength(1);
expect(controls[0]).toHaveAttribute('data-testid', 'queued-escalate-newest');
fireEvent.click(screen.getByTestId('queued-escalate-newest'));
expect(mockSendQueuedNow).toHaveBeenCalledWith(expect.objectContaining({ id: 'q2' }), {
preempt: true,
});
});
/** Steering refuses a recovery-bound item mid-run, so a proxy pointed at one
* would make the shortcut silently do nothing. */
it('points the collapsed escalate stand-in at the newest ELIGIBLE message', () => {
renderChips([twoQueued[0], { ...twoQueued[1], recoverySteerId: 'server-source' }], {
steering: outboxSteering({ duringRunActive: true, canSteer: true }),
});
fireEvent.click(screen.getByTestId('queued-escalate-newest'));
expect(mockSendQueuedNow).toHaveBeenCalledWith(expect.objectContaining({ id: 'q1' }), {
preempt: true,
});
});
it('offers no collapsed escalate stand-in when every row is recovery-bound', () => {
renderChips(
twoQueued.map((message, i) => ({ ...message, recoverySteerId: `server-source-${i}` })),
{ steering: outboxSteering({ duringRunActive: true, canSteer: true }) },
);
expect(screen.queryByTestId('queued-escalate-newest')).toBeNull();
expect(document.querySelectorAll('[data-escalate-steer="queued"]')).toHaveLength(0);
});
/** The group stands in for its rows while collapsed, so the accessible list
* must still contain an item — the rows themselves are unmounted. */
it('exposes the collapsed group as the list`s item, and the rows as a nested list', () => {
renderChips(twoQueued, { steering: outboxSteering() });
const list = screen.getByRole('list');
expect(screen.getAllByRole('listitem')).toHaveLength(1);
expect(list).toContainElement(screen.getByTestId('queue-group'));
fireEvent.click(screen.getByTestId('queue-group-toggle'));
// Outer item (the group) plus its two nested rows.
expect(screen.getAllByRole('list')).toHaveLength(2);
expect(screen.getAllByRole('listitem')).toHaveLength(3);
});
/** The shortcut promises the newest waiting message, and a promotion moves
* that row to the FRONT — so the target cannot be read off the array tail. */
it('keeps the escalate stand-in on the newest message after a promotion', () => {
const promoted = [
{ id: 'q2', text: 'newest, promoted', createdAt: 2, bumpedAt: 500 },
{ id: 'q1', text: 'older', createdAt: 1 },
];
renderChips(promoted, {
steering: outboxSteering({ duringRunActive: true, canSteer: true }),
});
fireEvent.click(screen.getByTestId('queued-escalate-newest'));
expect(mockSendQueuedNow).toHaveBeenCalledWith(expect.objectContaining({ id: 'q2' }), {
preempt: true,
});
});
it('keeps the hidden escalate stand-in out of the tab order', () => {
renderChips(twoQueued, { steering: outboxSteering({ duringRunActive: true, canSteer: true }) });
expect(screen.getByTestId('queued-escalate-newest')).toHaveAttribute('tabindex', '-1');
});
/** The shortcut's fallback takes the LAST matching control in the document,
* and the rows render in drain order — so the stand-in has to be last in
* BOTH states, or a promotion silently retargets the shortcut. */
it.each([
['collapsed', false],
['expanded', true],
])('leaves the escalate stand-in last in the document while %s', (_label, expand) => {
renderChips(twoQueued, { steering: outboxSteering({ duringRunActive: true, canSteer: true }) });
if (expand) {
fireEvent.click(screen.getByTestId('queue-group-toggle'));
}
const controls = document.querySelectorAll('[data-escalate-steer="queued"]');
expect(controls).toHaveLength(expand ? 3 : 1);
expect(controls[controls.length - 1]).toHaveAttribute('data-testid', 'queued-escalate-newest');
});
it('still offers every row its own escalation control when expanded', () => {
renderChips(twoQueued, { steering: outboxSteering({ duringRunActive: true, canSteer: true }) });
fireEvent.click(screen.getByTestId('queue-group-toggle'));
expect(screen.getAllByTestId('queued-interrupt-now')).toHaveLength(2);
});
it('offers Send next on every row except the one already next', () => {
renderChips(twoQueued, { steering: outboxSteering() });
fireEvent.click(screen.getByTestId('queue-group-toggle'));
const bumps = screen.getAllByTestId('queued-send-next');
expect(bumps).toHaveLength(1);
fireEvent.click(bumps[0]);
expect(mockBumpQueued).toHaveBeenCalledWith('q2');
});
/** A promotion lifts a row to the top of the promotions tier, which is still
* below every interrupt — so with only interrupts ahead the button would
* advertise a no-op. */
it('offers no Send next when only an unmovable interrupt is ahead', () => {
renderChips(
[
{ id: 'armed', text: 'interrupt', createdAt: 2, priority: true },
{ id: 'plain', text: 'ordinary', createdAt: 1 },
],
{ steering: outboxSteering() },
);
fireEvent.click(screen.getByTestId('queue-group-toggle'));
expect(screen.queryByTestId('queued-send-next')).toBeNull();
});
it('still offers Send next past a movable row that sits behind an interrupt', () => {
renderChips(
[
{ id: 'armed', text: 'interrupt', createdAt: 3, priority: true },
{ id: 'first', text: 'ordinary one', createdAt: 1 },
{ id: 'second', text: 'ordinary two', createdAt: 2 },
],
{ steering: outboxSteering() },
);
fireEvent.click(screen.getByTestId('queue-group-toggle'));
// Only the last row: it can overtake `first`, while `first` can overtake
// nothing but the interrupt.
const bumps = screen.getAllByTestId('queued-send-next');
expect(bumps).toHaveLength(1);
fireEvent.click(bumps[0]);
expect(mockBumpQueued).toHaveBeenCalledWith('second');
});
/** Promotion eligibility is computed in one pass, so verify the rule still
* holds for a queue deep enough that the old per-row prefix scan mattered. */
it('offers Send next on every row past the first movable one, at depth', () => {
const deep = [
{ id: 'armed', text: 'interrupt', createdAt: 9, priority: true },
...Array.from({ length: 5 }, (_, i) => ({
id: `q${i}`,
text: `ordinary ${i}`,
createdAt: i,
})),
];
renderChips(deep, { steering: outboxSteering() });
fireEvent.click(screen.getByTestId('queue-group-toggle'));
// The interrupt and the first ordinary row cannot overtake anything.
expect(screen.getAllByTestId('queued-send-next')).toHaveLength(4);
});
it('merges the batch into one turn', () => {
renderChips(twoQueued, { steering: outboxSteering() });
fireEvent.click(screen.getByTestId('queue-group-toggle'));
fireEvent.click(screen.getByTestId('queue-merge'));
expect(mockMergeQueued).toHaveBeenCalledTimes(1);
});
/** Folding reads the queue, so an emptied editor would carry the words the
* user just deleted into the merged message. */
it('refuses to merge while an inline edit is empty', () => {
renderChips(twoQueued, { steering: outboxSteering() });
fireEvent.click(screen.getByTestId('queue-group-toggle'));
const rows = screen.getAllByTestId('queued-message-row');
fireEvent.click(rows[1].querySelector('span[title]') as HTMLElement);
fireEvent.change(screen.getByTestId('queued-message-edit'), { target: { value: ' ' } });
const merge = screen.getByTestId('queue-merge');
expect(merge).toBeDisabled();
fireEvent.click(merge);
expect(mockMergeQueued).not.toHaveBeenCalled();
// Resolving the edit releases it.
fireEvent.change(screen.getByTestId('queued-message-edit'), { target: { value: 'rewritten' } });
expect(screen.getByTestId('queue-merge')).not.toBeDisabled();
});
/** The queue still holds the pre-edit words, so handing them back would
* resurrect text the user visibly deleted. Emptying a row and removing it
* reads as "delete this". */
it('does not return stale words to the composer when removing an emptied row', () => {
renderChips([twoQueued[0]], { steering: outboxSteering() });
fireEvent.click(screen.getByText('first thought'));
fireEvent.change(screen.getByTestId('queued-message-edit'), { target: { value: '' } });
fireEvent.click(screen.getByLabelText('com_ui_remove_queued'));
expect(mockRestoreToComposer).not.toHaveBeenCalled();
expect(mockRemoveQueued).toHaveBeenCalledWith('q1');
});
it('still returns the words when removing a row that was not being emptied', () => {
renderChips([twoQueued[0]], { steering: outboxSteering() });
fireEvent.click(screen.getByLabelText('com_ui_remove_queued'));
expect(mockRestoreToComposer).toHaveBeenCalledWith(
'first thought',
undefined,
{ quotes: undefined, manualSkills: undefined },
CONVO_ID,
);
expect(mockRemoveQueued).toHaveBeenCalledWith('q1');
});
/** Clear all folds the queue exactly as Merge does, so it takes the same
* standdown rather than being a documented exception. */
it('refuses to clear all while an inline edit is empty', () => {
renderChips(twoQueued, { steering: outboxSteering() });
fireEvent.click(screen.getByTestId('queue-group-toggle'));
const rows = screen.getAllByTestId('queued-message-row');
fireEvent.click(rows[1].querySelector('span[title]') as HTMLElement);
fireEvent.change(screen.getByTestId('queued-message-edit'), { target: { value: '' } });
const clear = screen.getByTestId('queue-clear-all');
expect(clear).toBeDisabled();
fireEvent.click(clear);
expect(mockClearQueued).not.toHaveBeenCalled();
fireEvent.change(screen.getByTestId('queued-message-edit'), { target: { value: 'kept' } });
expect(screen.getByTestId('queue-clear-all')).not.toBeDisabled();
});
/** The composer box clips, so a deep queue needs its own scroll — and the
* disclosure and the actions must stay outside it to remain reachable. */
it('scrolls the expanded rows without clipping the header or the actions', () => {
renderChips(
Array.from({ length: 8 }, (_, i) => ({ id: `q${i}`, text: `queued ${i}`, createdAt: i })),
{ steering: outboxSteering() },
);
fireEvent.click(screen.getByTestId('queue-group-toggle'));
const list = screen.getByTestId('queue-rows');
// Named by its own count, so it does not duplicate the outer stack's label.
expect(list).toHaveAttribute('role', 'list');
expect(list).toHaveAccessibleName('com_ui_queue_count');
expect(list.className).toContain('overflow-y-auto');
expect(list.className).toContain('max-h-[35vh]');
// Outside the scroll container, so they cannot be clipped away.
expect(list).not.toContainElement(screen.getByTestId('queue-group-toggle'));
expect(list).not.toContainElement(screen.getByTestId('queue-merge'));
expect(list).not.toContainElement(screen.getByTestId('queue-clear-all'));
});
/** Collapsing unmounts the rows, so the disclosure's own text is the queue's
* only description — an aria-label would overwrite it. */
it('announces the count and next-up preview as the disclosure name', () => {
renderChips(twoQueued, { steering: outboxSteering() });
const toggle = screen.getByTestId('queue-group-toggle');
expect(toggle).not.toHaveAttribute('aria-label');
expect(toggle).toHaveAccessibleName(/com_ui_queue_count/);
expect(toggle).toHaveAccessibleName(/com_ui_queue_next_up/);
expect(toggle).toHaveAttribute('aria-expanded', 'false');
});
it('refuses to merge while a recovered row holds a parked server source', () => {
renderChips([twoQueued[0], { ...twoQueued[1], recoverySteerId: 'server-source' }], {
steering: outboxSteering(),
});
fireEvent.click(screen.getByTestId('queue-group-toggle'));
expect(screen.getByTestId('queue-merge')).toBeDisabled();
fireEvent.click(screen.getByTestId('queue-merge'));
expect(mockMergeQueued).not.toHaveBeenCalled();
});
it('clear all hands the folded words back to the composer', async () => {
const folded: QueuedMessage = {
id: 'q1',
text: 'first thought\n\nsecond thought',
createdAt: 1,
};
mockClearQueued = jest.fn(async () => folded);
renderChips(twoQueued, { steering: outboxSteering() });
fireEvent.click(screen.getByTestId('queue-group-toggle'));
fireEvent.click(screen.getByTestId('queue-clear-all'));
await waitFor(() => {
expect(mockRestoreToComposer).toHaveBeenCalledWith(
'first thought\n\nsecond thought',
undefined,
{ quotes: undefined, manualSkills: undefined },
CONVO_ID,
);
});
expect(mockEnqueue).not.toHaveBeenCalled();
});
it('returns the words to the queue when the composer refuses them', async () => {
const folded: QueuedMessage = {
id: 'q1',
text: 'not lost',
createdAt: 1,
expectedPredecessorCreatedAt: 4242,
priority: true,
bumpedAt: 99,
};
mockClearQueued = jest.fn(async () => folded);
mockRestoreToComposer.mockReturnValueOnce(false);
renderChips(twoQueued, { steering: outboxSteering() });
fireEvent.click(screen.getByTestId('queue-group-toggle'));
fireEvent.click(screen.getByTestId('queue-clear-all'));
await waitFor(() => {
/** The ITEM goes back whole, so no field can be quietly dropped — the
* fence and the interrupt tier were each lost once when this path
* rebuilt a row from parts. */
expect(mockRequeueCleared).toHaveBeenCalledWith([folded]);
});
expect(mockEnqueue).not.toHaveBeenCalled();
expect(mockShowToast).toHaveBeenCalledWith(
expect.objectContaining({ message: 'com_ui_steer_edit_queued' }),
);
});
});
describe('PendingSteerChips — queued row editing', () => {
beforeEach(() => {
jest.clearAllMocks();
});
it('rewrites a waiting message in place', () => {
renderChips([twoQueued[0]], { steering: outboxSteering() });
fireEvent.click(screen.getByText('first thought'));
const editor = screen.getByTestId('queued-message-edit');
fireEvent.change(editor, { target: { value: 'sharper thought' } });
fireEvent.keyDown(editor, { key: 'Enter' });
expect(mockUpdateQueuedText).toHaveBeenCalledWith('q1', 'sharper thought');
});
/** The front row draining drops the queue below the grouping threshold, which
* remounts the surviving row. Unmounting an input fires no `blur`, so the
* flush has to happen on the way out or the typing is silently lost. */
it('flushes an in-progress edit when the group collapses under it', () => {
renderChips(twoQueued, { steering: outboxSteering() });
fireEvent.click(screen.getByTestId('queue-group-toggle'));
const rows = screen.getAllByTestId('queued-message-row');
fireEvent.click(rows[1].querySelector('span[title]') as HTMLElement);
fireEvent.change(screen.getByTestId('queued-message-edit'), {
target: { value: 'edited while the front row drained' },
});
// The front row drains, leaving a lone message: the group gives way to a
// plain chip and the edited row is remounted elsewhere in the tree.
act(() => {
updateQueueForTest!(() => [twoQueued[1]]);
});
expect(screen.queryByTestId('queue-group')).toBeNull();
expect(mockUpdateQueuedText).toHaveBeenCalledWith('q2', 'edited while the front row drained');
});
/** `steering` is rebuilt at every run boundary. Whatever the write model,
* the invariant is that a run ending under an open editor must not make the
* edit final — Escape still has to put the original words back. */
it('keeps an edit abandonable when the run state changes under it', () => {
const Harness = ({ live }: { live: boolean }) => (
<PendingSteerChips
conversationId={CONVO_ID}
steering={steeringStub(outboxSteering({ duringRunActive: live, canSteer: live }))}
onEditToComposer={mockEditToComposer}
onRestoreToComposer={mockRestoreToComposer}
/>
);
const tree = (live: boolean) => (
<RecoilRoot
initializeState={({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [twoQueued[0]]);
}}
>
<Harness live={live} />
</RecoilRoot>
);
const { rerender } = render(tree(true));
fireEvent.click(screen.getByText('first thought'));
fireEvent.change(screen.getByTestId('queued-message-edit'), {
target: { value: 'still typing' },
});
// The run ends: same row, a freshly built `steering`.
rerender(tree(false));
expect(screen.getByTestId('queued-message-edit')).toBeInTheDocument();
fireEvent.keyDown(screen.getByTestId('queued-message-edit'), { key: 'Escape' });
expect(mockUpdateQueuedText).toHaveBeenLastCalledWith('q1', 'first thought');
});
/** A merged row carries paragraph breaks, and a single-line input flattens
* them on the first keystroke. */
it('keeps paragraph breaks while editing a merged row', () => {
renderChips([{ id: 'merged', text: 'first part\n\nsecond part', createdAt: 1 }], {
steering: outboxSteering(),
});
fireEvent.click(screen.getByTitle('com_ui_queue_edit_inline'));
const editor = screen.getByTestId('queued-message-edit');
expect(editor.tagName).toBe('TEXTAREA');
expect(editor).toHaveValue('first part\n\nsecond part');
fireEvent.change(editor, { target: { value: 'first part\n\nsecond part edited' } });
fireEvent.keyDown(editor, { key: 'Enter' });
expect(mockUpdateQueuedText).toHaveBeenCalledWith('merged', 'first part\n\nsecond part edited');
});
/** The drain reads the atom, not the local draft, so an edit still only local
* when the row sends would go out as the old text. The window opening is the
* last safe moment to settle it. */
it('closes an open editor the moment a send becomes pending', () => {
renderChips([twoQueued[0]], { steering: outboxSteering() });
fireEvent.click(screen.getByText('first thought'));
fireEvent.change(screen.getByTestId('queued-message-edit'), {
target: { value: 'final wording' },
});
// Written through immediately, so whatever sends is what is displayed.
expect(mockUpdateQueuedText).toHaveBeenCalledWith('q1', 'final wording');
// The run ends and the grace window opens.
act(() => {
setHoldForTest!({
runEnd: { conversationId: CONVO_ID, outcome: 'completed', endedAt: 1 },
dueAt: 4_000_000_000_000,
});
});
expect(mockUpdateQueuedText).toHaveBeenCalledWith('q1', 'final wording');
expect(screen.queryByTestId('queued-message-edit')).toBeNull();
});
it('does not offer an edit while a send is already pending', () => {
renderChips([twoQueued[0]], { steering: outboxSteering() });
act(() => {
setHoldForTest!({
runEnd: { conversationId: CONVO_ID, outcome: 'completed', endedAt: 1 },
dueAt: 4_000_000_000_000,
});
});
fireEvent.click(screen.getByText('first thought'));
expect(screen.queryByTestId('queued-message-edit')).toBeNull();
});
/** An IME candidate confirmation arrives as an unshifted Enter while
* composition is still active; committing there saves half-typed text. */
it.each([
['isComposing', { key: 'Enter', isComposing: true }],
['keyCode 229', { key: 'Enter', keyCode: 229 }],
])('ignores Enter reported as %s by an IME', (_label, init) => {
renderChips([twoQueued[0]], { steering: outboxSteering() });
fireEvent.click(screen.getByText('first thought'));
const editor = screen.getByTestId('queued-message-edit');
fireEvent.change(editor, { target: { value: 'partial candidate' } });
fireEvent.keyDown(editor, init);
// Still editing: candidate confirmation stays inside the editor.
expect(screen.getByTestId('queued-message-edit')).toBeInTheDocument();
expect(mockUpdateQueuedText).toHaveBeenLastCalledWith('q1', 'partial candidate');
});
it('adds a line with Shift+Enter instead of committing', () => {
renderChips([twoQueued[0]], { steering: outboxSteering() });
fireEvent.click(screen.getByText('first thought'));
fireEvent.keyDown(screen.getByTestId('queued-message-edit'), { key: 'Enter', shiftKey: true });
expect(mockUpdateQueuedText).not.toHaveBeenCalled();
expect(screen.getByTestId('queued-message-edit')).toBeInTheDocument();
});
/** The write-through refuses blank text, so an emptied editor is the one state
* where the row still holds words the user has visibly deleted. */
it('stands the senders down while the editor is empty', () => {
renderChips([twoQueued[0]], {
steering: outboxSteering({ duringRunActive: true, canSteer: true }),
});
fireEvent.click(screen.getByText('first thought'));
fireEvent.change(screen.getByTestId('queued-message-edit'), { target: { value: ' ' } });
expect(screen.getByText('com_ui_steer').closest('button')).toBeDisabled();
expect(screen.getByTestId('queued-interrupt-now')).toBeDisabled();
// Resolving the edit brings them back.
fireEvent.change(screen.getByTestId('queued-message-edit'), { target: { value: 'rewritten' } });
expect(screen.getByText('com_ui_steer').closest('button')).not.toBeDisabled();
});
/** The blank was never written, so closing the editor alone would leave the
* queue holding words the screen no longer shows — and the drain sends the
* queue. Resolve it the way Escape does. */
it('restores the original when a send arrives on an emptied editor', () => {
renderChips([twoQueued[0]], { steering: outboxSteering() });
fireEvent.click(screen.getByText('first thought'));
fireEvent.change(screen.getByTestId('queued-message-edit'), { target: { value: '' } });
act(() => {
setHoldForTest!({
runEnd: { conversationId: CONVO_ID, outcome: 'completed', endedAt: 1 },
dueAt: 4_000_000_000_000,
});
});
expect(mockUpdateQueuedText).toHaveBeenLastCalledWith('q1', 'first thought');
expect(screen.queryByTestId('queued-message-edit')).toBeNull();
});
/** The shortcut proxy targets this row but lives in the group, and the
* shortcut fires even while a textarea has focus — so emptying the editor has
* to reach it too, or Ctrl/Cmd+Shift+. sends the deleted words. */
it('stands the shortcut proxy down for an empty edit, and brings it back', () => {
renderChips(twoQueued, { steering: outboxSteering({ duringRunActive: true, canSteer: true }) });
fireEvent.click(screen.getByTestId('queue-group-toggle'));
// The newest row is the proxy's target.
const rows = screen.getAllByTestId('queued-message-row');
fireEvent.click(rows[1].querySelector('span[title]') as HTMLElement);
fireEvent.change(screen.getByTestId('queued-message-edit'), { target: { value: '' } });
expect(screen.getByTestId('queued-escalate-newest')).toBeDisabled();
fireEvent.change(screen.getByTestId('queued-message-edit'), { target: { value: 'rewritten' } });
expect(screen.getByTestId('queued-escalate-newest')).not.toBeDisabled();
});
it('releases the empty-edit claim when the editor closes', () => {
renderChips(twoQueued, { steering: outboxSteering({ duringRunActive: true, canSteer: true }) });
fireEvent.click(screen.getByTestId('queue-group-toggle'));
const rows = screen.getAllByTestId('queued-message-row');
fireEvent.click(rows[1].querySelector('span[title]') as HTMLElement);
const editor = screen.getByTestId('queued-message-edit');
fireEvent.change(editor, { target: { value: '' } });
expect(screen.getByTestId('queued-escalate-newest')).toBeDisabled();
// Escape restores the original words, so nothing is held back any more.
fireEvent.keyDown(editor, { key: 'Escape' });
expect(screen.getByTestId('queued-escalate-newest')).not.toBeDisabled();
});
/** The invariant that lets every reader trust the data: a blank row cannot
* outlive its editor, so a resting queue never holds one. */
it('restores the original when an emptied editor is closed rather than abandoned', () => {
renderChips([twoQueued[0]], { steering: outboxSteering() });
fireEvent.click(screen.getByText('first thought'));
const editor = screen.getByTestId('queued-message-edit');
fireEvent.change(editor, { target: { value: '' } });
// Blank while the editor is open, which is what the senders read.
expect(JSON.parse(screen.getByTestId('queue-state').textContent ?? '[]')[0].text).toBe('');
fireEvent.blur(editor);
expect(mockUpdateQueuedText).toHaveBeenLastCalledWith('q1', 'first thought');
expect(screen.queryByTestId('queued-message-edit')).toBeNull();
});
it('trims the words it settles', () => {
renderChips([twoQueued[0]], { steering: outboxSteering() });
fireEvent.click(screen.getByText('first thought'));
const editor = screen.getByTestId('queued-message-edit');
fireEvent.change(editor, { target: { value: ' spaced out ' } });
fireEvent.keyDown(editor, { key: 'Enter' });
expect(mockUpdateQueuedText).toHaveBeenLastCalledWith('q1', 'spaced out');
});
it('puts the original words back on Escape', () => {
renderChips([twoQueued[0]], { steering: outboxSteering() });
fireEvent.click(screen.getByText('first thought'));
const editor = screen.getByTestId('queued-message-edit');
fireEvent.change(editor, { target: { value: 'discard me' } });
expect(mockUpdateQueuedText).toHaveBeenCalledWith('q1', 'discard me');
fireEvent.keyDown(editor, { key: 'Escape' });
expect(mockUpdateQueuedText).toHaveBeenLastCalledWith('q1', 'first thought');
expect(screen.queryByTestId('queued-message-edit')).toBeNull();
});
/** Its parked source is matched by exact text server-side, so the words may
* only change after the discard ladder has downgraded the row. */
it('never edits a recovered row in place', () => {
renderChips([{ ...twoQueued[0], recoverySteerId: 'server-source' }], {
steering: outboxSteering(),
});
fireEvent.click(screen.getByText('first thought'));
expect(screen.queryByTestId('queued-message-edit')).toBeNull();
});
});
describe('PendingSteerChips — withheld automatic send', () => {
beforeEach(() => {
jest.clearAllMocks();
});
const withHold = (queued: QueuedMessage[]) =>
render(
<RecoilRoot
initializeState={({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), queued);
set(store.queueDrainHoldByConvoId(CONVO_ID), {
runEnd: {
conversationId: CONVO_ID,
outcome: 'completed' as const,
endedAt: 1,
},
dueAt: Date.now() + 3000,
});
}}
>
<PendingSteerChips
conversationId={CONVO_ID}
steering={steeringStub(outboxSteering())}
onEditToComposer={mockEditToComposer}
onRestoreToComposer={mockRestoreToComposer}
/>
</RecoilRoot>,
);
it('announces the send and offers to take it back', () => {
withHold(twoQueued);
const banner = screen.getByTestId('queue-sending-banner');
expect(banner).toBeInTheDocument();
expect(banner.querySelector('[aria-live="polite"]')).toBeInTheDocument();
fireEvent.click(screen.getByTestId('queue-undo-send'));
expect(mockCancelQueueDrain).toHaveBeenCalledTimes(1);
});
it('shows nothing to undo once the queue is empty', () => {
withHold([]);
expect(screen.queryByTestId('queue-sending-banner')).toBeNull();
});
});

317
client/src/hooks/Chat/__tests__/useQueueDrain.spec.tsx
View file

@ -3,7 +3,7 @@ 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, QueueDrainHold } from '~/store/families';
import useQueueDrain from '../useQueueDrain';
import store from '~/store';
@ -38,7 +38,7 @@ function setup(
store.queuedMessagesByConvoId(Constants.NEW_CONVO),
);
setters.setInterruptFlag = useSetRecoilState(store.drainAfterAbortByIndex(INDEX));
useQueueDrain(INDEX, activeConversationId, ask);
useQueueDrain(INDEX, activeConversationId, ask, { undoGraceMs: 0 });
return null;
}
@ -97,6 +97,39 @@ describe('useQueueDrain', () => {
expect(ask).toHaveBeenCalledWith({ text: 'first follow-up' }, emptyOverrides);
});
/** A row being edited holds exactly what is typed, blank included. Blank is
* not a message, and skipping to the row behind it would send out of order —
* so this epoch drains nothing and the next run end picks the queue up. */
it('drains nothing when the front row is mid-edit and blank', async () => {
const { ask, setters } = setup(({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [
{ id: 'q1', text: ' ', createdAt: 1 },
queuedMessage('q2', 'behind it'),
]);
});
act(() => {
setters.setRunEnd!(runEnd());
});
await new Promise((resolve) => setTimeout(resolve, 25));
expect(ask).not.toHaveBeenCalled();
// The epoch was consumed, so the effect is not spinning; a later run end
// drains normally once the row has words again.
act(() => {
setters.setQueue!([
queuedMessage('q1', 'now it has words'),
queuedMessage('q2', 'behind it'),
]);
});
act(() => {
setters.setRunEnd!(runEnd({ generationCreatedAt: 77 }));
});
await waitFor(() => expect(ask).toHaveBeenCalledTimes(1));
expect(ask).toHaveBeenCalledWith({ text: 'now it has words' }, emptyOverrides);
});
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')]);
@ -460,7 +493,7 @@ describe('useQueueDrain', () => {
({ activeConversationId }: { activeConversationId: string }) => {
setRunEnd = useSetRecoilState(store.runEndByIndex(INDEX));
setIsSubmitting = useSetRecoilState(store.isSubmittingFamily(INDEX));
useQueueDrain(INDEX, activeConversationId, ask);
useQueueDrain(INDEX, activeConversationId, ask, { undoGraceMs: 0 });
return {
indexEnd: useRecoilValue(store.runEndByIndex(INDEX)),
parkedA: useRecoilValue(store.pendingRunEndByConvoId(CONVO_A)),
@ -599,7 +632,7 @@ describe('useQueueDrain', () => {
setters.setInterruptFlag = useSetRecoilState(store.drainAfterAbortByIndex(INDEX));
state.newConvoQueue = useRecoilValue(store.queuedMessagesByConvoId(Constants.NEW_CONVO));
state.parkedUnderOptimistic = useRecoilValue(store.pendingRunEndByConvoId(OPTIMISTIC_ID));
useQueueDrain(INDEX, Constants.NEW_CONVO as string, ask);
useQueueDrain(INDEX, Constants.NEW_CONVO as string, ask, { undoGraceMs: 0 });
return null;
}
@ -693,3 +726,279 @@ describe('useQueueDrain', () => {
expect(ask).toHaveBeenCalledTimes(1);
});
});
/** A short real window rather than fake timers: the drain rides react-query
* mutations and Recoil propagation, which fake timers would also freeze. */
const GRACE_MS = 30;
const AFTER_GRACE_MS = 120;
describe('useQueueDrain undo grace', () => {
function setupGraced(initialize?: (snapshot: MutableSnapshot) => void) {
const ask = jest.fn();
const state: {
setRunEnd?: (value: RunEnd | null) => void;
setInterruptFlag?: (value: DrainAfterAbort | false) => void;
setIsSubmitting?: (value: boolean) => void;
cancelHold?: () => void;
queue?: QueuedMessage[];
newConvoQueue?: QueuedMessage[];
hold?: QueueDrainHold | null;
} = {};
function Harness() {
state.setRunEnd = useSetRecoilState(store.runEndByIndex(INDEX));
state.setInterruptFlag = useSetRecoilState(store.drainAfterAbortByIndex(INDEX));
const setHold = useSetRecoilState(store.queueDrainHoldByConvoId(CONVO_ID));
/** Mirrors `useSteering.cancelQueueDrain`: a released epoch is
* tombstoned rather than merely dropped. */
state.cancelHold = () =>
setHold((held) => {
if (held == null) {
return null;
}
return held.status === 'released' ? { ...held, status: 'cancelled' } : null;
});
state.setIsSubmitting = useSetRecoilState(store.isSubmittingFamily(INDEX));
state.queue = useRecoilValue(store.queuedMessagesByConvoId(CONVO_ID));
state.newConvoQueue = useRecoilValue(store.queuedMessagesByConvoId(Constants.NEW_CONVO));
state.hold = useRecoilValue(store.queueDrainHoldByConvoId(CONVO_ID));
useQueueDrain(INDEX, CONVO_ID, ask, { undoGraceMs: GRACE_MS });
return null;
}
const queryClient = new QueryClient({ defaultOptions: { mutations: { retry: false } } });
const wrapper = ({ children }: { children: React.ReactNode }) => (
<QueryClientProvider client={queryClient}>
<RecoilRoot initializeState={initialize}>
<Harness />
{children}
</RecoilRoot>
</QueryClientProvider>
);
renderHook(() => null, { wrapper });
return { ask, state };
}
const withQueue =
(...items: QueuedMessage[]) =>
({ set }: MutableSnapshot) => {
set(store.queuedMessagesByConvoId(CONVO_ID), items);
};
it('withholds the send for the grace window, then drains', async () => {
const { ask, state } = setupGraced(withQueue(queuedMessage('q1', 'hold me')));
act(() => {
state.setRunEnd!(runEnd());
});
// Held: the epoch is parked and the queue is untouched, so there is
// nothing to restore if the user cancels.
await waitFor(() => expect(state.hold).not.toBeNull());
expect(ask).not.toHaveBeenCalled();
expect(state.queue).toHaveLength(1);
await waitFor(() => expect(ask).toHaveBeenCalledTimes(1), { timeout: AFTER_GRACE_MS * 4 });
expect(ask).toHaveBeenCalledWith({ text: 'hold me' }, emptyOverrides);
expect(state.hold).toBeNull();
});
it('cancelling the hold sends nothing and leaves the queue intact', async () => {
const { ask, state } = setupGraced(withQueue(queuedMessage('q1', 'never sent')));
act(() => {
state.setRunEnd!(runEnd());
});
await waitFor(() => expect(state.hold).not.toBeNull());
act(() => {
state.cancelHold!();
});
await new Promise((resolve) => setTimeout(resolve, AFTER_GRACE_MS));
expect(ask).not.toHaveBeenCalled();
expect(state.queue).toEqual([expect.objectContaining({ id: 'q1', text: 'never sent' })]);
});
it('skips the grace when interrupt & send armed the drain', async () => {
const { ask, state } = setupGraced(withQueue(queuedMessage('q1', 'now please')));
act(() => {
state.setInterruptFlag!({ conversationId: CONVO_ID, generationCreatedAt: 41 });
});
act(() => {
state.setRunEnd!(runEnd({ outcome: 'aborted' }));
});
await waitFor(() => expect(ask).toHaveBeenCalledTimes(1));
expect(state.hold).toBeNull();
});
it('does not hold when nothing is queued', async () => {
const { ask, state } = setupGraced();
act(() => {
state.setRunEnd!(runEnd());
});
await new Promise((resolve) => setTimeout(resolve, AFTER_GRACE_MS));
expect(state.hold).toBeNull();
expect(ask).not.toHaveBeenCalled();
});
it('parks a second epoch arriving inside the window so its drain is not lost', async () => {
const { ask, state } = setupGraced(
withQueue(queuedMessage('q1', 'first'), queuedMessage('q2', 'second')),
);
act(() => {
state.setRunEnd!(runEnd());
});
await waitFor(() => expect(state.hold).not.toBeNull());
act(() => {
state.setRunEnd!(runEnd({ generationCreatedAt: 99 }));
});
// Both epochs drain their one message each: the second was parked rather
// than overwriting the held one.
await waitFor(() => expect(ask).toHaveBeenCalledTimes(2), { timeout: AFTER_GRACE_MS * 8 });
expect(ask.mock.calls.map(([payload]) => payload.text)).toEqual(['first', 'second']);
});
/** The composer reads the resolved id by run end, so rows left under the
* optimistic key would vanish from the group for the whole window. */
it('migrates a first turn`s queue to the resolved conversation before opening the window', async () => {
const { ask, state } = setupGraced(({ set }) => {
set(store.queuedMessagesByConvoId(Constants.NEW_CONVO), [
queuedMessage('q1', 'queued on the first turn'),
]);
});
act(() => {
state.setRunEnd!(runEnd({ startedAsNewConvo: true }));
});
await waitFor(() => expect(state.hold).not.toBeNull());
expect(state.newConvoQueue).toEqual([]);
expect(state.queue).toEqual([
expect.objectContaining({ id: 'q1', text: 'queued on the first turn' }),
]);
expect(ask).not.toHaveBeenCalled();
await waitFor(() => expect(ask).toHaveBeenCalledTimes(1), { timeout: AFTER_GRACE_MS * 4 });
expect(ask).toHaveBeenCalledWith({ text: 'queued on the first turn' }, emptyOverrides);
});
/** A run started INSIDE the window blocks the drain, so the released epoch
* sits parked with the banner still up. Undo then has to neutralize that
* epoch, or it sends anyway the moment the run goes idle. */
it('undo after the window closed still cancels the parked epoch', async () => {
const { ask, state } = setupGraced(withQueue(queuedMessage('q1', 'must not send')));
act(() => {
state.setRunEnd!(runEnd());
});
await waitFor(() => expect(state.hold).not.toBeNull());
// The user starts another turn before the window closes.
act(() => {
state.setIsSubmitting!(true);
});
// The timer hands the epoch back, but the drain is blocked, so the hold
// stays — marked released, which is what stops Undo being advertised.
await waitFor(() => expect(state.hold?.status).toBe('released'), {
timeout: AFTER_GRACE_MS * 4,
});
expect(ask).not.toHaveBeenCalled();
act(() => {
state.cancelHold!();
});
expect(state.hold?.status).toBe('cancelled');
act(() => {
state.setIsSubmitting!(false);
});
await new Promise((resolve) => setTimeout(resolve, AFTER_GRACE_MS));
expect(ask).not.toHaveBeenCalled();
expect(state.queue).toHaveLength(1);
expect(state.hold).toBeNull();
});
/** The window may belong to an unrelated earlier epoch, and the user aborted
* a run to say this now — waiting out that timer contradicts the rule that
* armed interrupts skip the grace. */
it('lets an armed interrupt drain straight through a standing window', async () => {
const { ask, state } = setupGraced(
withQueue(queuedMessage('q1', 'interrupt text'), queuedMessage('q2', 'ordinary follow-up')),
);
act(() => {
state.setRunEnd!(runEnd());
});
await waitFor(() => expect(state.hold).not.toBeNull());
expect(ask).not.toHaveBeenCalled();
// Interrupt & send: the arm plus the aborted epoch it belongs to.
act(() => {
state.setInterruptFlag!({ conversationId: CONVO_ID, generationCreatedAt: 99 });
});
act(() => {
state.setRunEnd!(runEnd({ outcome: 'aborted', generationCreatedAt: 99 }));
});
// No timer wait: the assertion resolves well inside the grace window.
await waitFor(() => expect(ask).toHaveBeenCalledTimes(1));
expect(ask).toHaveBeenCalledWith({ text: 'interrupt text' }, emptyOverrides);
});
/** A hold owns exactly one epoch. Acting on a NEWER terminal event on its
* behalf loses that signal and leaves the older one to reopen a window. */
it('a cancelled hold discards only its own epoch', async () => {
const { ask, state } = setupGraced(
withQueue(queuedMessage('q1', 'first'), queuedMessage('q2', 'second')),
);
act(() => {
state.setRunEnd!(runEnd({ generationCreatedAt: 41 }));
});
await waitFor(() => expect(state.hold).not.toBeNull());
// A newer run finishes while the window is open, so its epoch queues up
// behind the held one; then the window closes and Undo lands.
act(() => {
state.setIsSubmitting!(true);
});
act(() => {
state.setRunEnd!(runEnd({ generationCreatedAt: 77 }));
});
await waitFor(() => expect(state.hold?.status).toBe('released'));
act(() => {
state.cancelHold!();
});
act(() => {
state.setIsSubmitting!(false);
});
// The cancelled epoch sends nothing; the NEWER completion is still honoured
// and drains one row, rather than being swallowed on the hold's behalf.
await waitFor(() => expect(ask).toHaveBeenCalledTimes(1), { timeout: AFTER_GRACE_MS * 8 });
expect(ask).toHaveBeenCalledWith({ text: 'first' }, emptyOverrides);
expect(state.queue).toHaveLength(1);
});
it('drains immediately when the window already expired while away', async () => {
const { ask } = setupGraced(({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [queuedMessage('q1', 'overdue')]);
set(store.queueDrainHoldByConvoId(CONVO_ID), {
runEnd: runEnd(),
dueAt: Date.now() - 1000,
});
});
await waitFor(() => expect(ask).toHaveBeenCalledTimes(1));
expect(ask).toHaveBeenCalledWith({ text: 'overdue' }, emptyOverrides);
});
});

355
client/src/hooks/Chat/__tests__/useSteering.spec.tsx
View file

@ -2420,3 +2420,358 @@ describe('useSteering', () => {
});
});
});
describe('useSteering — clear all', () => {
beforeEach(() => {
jest.clearAllMocks();
});
function setupClearAll(initialize?: (snapshot: MutableSnapshot) => void) {
const sendNow = jest.fn();
const stopGenerating = jest.fn();
const wrapper = ({ children }: { children: React.ReactNode }) => (
<RecoilRoot initializeState={withActiveGeneration(initialize)}>{children}</RecoilRoot>
);
return renderHook(
() => ({
steering: useSteering({
index: 0,
conversationId: CONVO_ID,
conversation: agentsConversation,
isSubmitting: false,
answerModeActive: false,
sendNow,
stopGenerating,
}),
queue: useQueue(CONVO_ID),
drainHold: useRecoilValue(store.queueDrainHoldByConvoId(CONVO_ID)),
}),
{ wrapper },
);
}
/** Removing the last row leaves nothing to auto-send, and a surviving window
* could later drain a row queued under a NEWER run. */
it('retires the pending send when removal empties the queue', () => {
const { result } = setupClearAll(({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [{ id: 'q1', text: 'only row', createdAt: 1 }]);
set(store.queueDrainHoldByConvoId(CONVO_ID), {
runEnd: { conversationId: CONVO_ID, outcome: 'completed', endedAt: 1 },
dueAt: 4_000_000_000_000,
});
});
act(() => {
result.current.steering.removeQueued('q1');
});
expect(result.current.queue).toEqual([]);
expect(result.current.drainHold).toBeNull();
});
/** Once the timer has handed the epoch back, the hold is the only thing that
* can neutralize it — dropping it outright lets the parked epoch through. */
it('tombstones a released window rather than dropping it', () => {
const { result } = setupClearAll(({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [{ id: 'q1', text: 'only row', createdAt: 1 }]);
set(store.queueDrainHoldByConvoId(CONVO_ID), {
runEnd: { conversationId: CONVO_ID, outcome: 'completed', endedAt: 1 },
dueAt: 1,
status: 'released',
});
});
act(() => {
result.current.steering.removeQueued('q1');
});
expect(result.current.queue).toEqual([]);
expect(result.current.drainHold).toMatchObject({ status: 'cancelled' });
});
/** A withheld epoch grants ONE drain; expediting a row spends it, so leaving
* the window armed for the remaining rows would send twice for one run end —
* and would do it even if that manual run was stopped. */
it('retires the pending send when a row is expedited, rows remaining or not', () => {
const { result } = setupClearAll(({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [
{ id: 'q1', text: 'sent by hand', createdAt: 1 },
{ id: 'q2', text: 'still waiting', createdAt: 2 },
]);
set(store.queueDrainHoldByConvoId(CONVO_ID), {
runEnd: { conversationId: CONVO_ID, outcome: 'completed', endedAt: 1 },
dueAt: 4_000_000_000_000,
});
});
act(() => {
result.current.steering.sendQueuedNow({ id: 'q1', text: 'sent by hand', createdAt: 1 });
});
expect(result.current.queue).toEqual([
expect.objectContaining({ id: 'q2', text: 'still waiting' }),
]);
expect(result.current.drainHold).toBeNull();
});
it('keeps the pending send while other rows remain', () => {
const { result } = setupClearAll(({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [
{ id: 'q1', text: 'first', createdAt: 1 },
{ id: 'q2', text: 'second', createdAt: 2 },
]);
set(store.queueDrainHoldByConvoId(CONVO_ID), {
runEnd: { conversationId: CONVO_ID, outcome: 'completed', endedAt: 1 },
dueAt: 4_000_000_000_000,
});
});
act(() => {
result.current.steering.removeQueued('q1');
});
expect(result.current.queue).toHaveLength(1);
expect(result.current.drainHold).not.toBeNull();
});
it('folds the queue into one payload and empties it', async () => {
const { result } = setupClearAll(({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [
{ id: 'q1', text: 'first', createdAt: 1 },
{ id: 'q2', text: 'second', createdAt: 2 },
]);
});
let cleared: QueuedMessage | null | undefined;
await act(async () => {
cleared = await result.current.steering.clearQueued();
});
expect(cleared?.text).toBe('first\n\nsecond');
expect(result.current.queue).toEqual([]);
});
/** The row must be OUT of the queue for the cancellation round trip: a run
* completing mid-flight would otherwise drain and send the very message the
* user is removing. */
it('holds a recovered row out of the queue while its source cancels', async () => {
let settleCancel: ((value: { removed: boolean }) => void) | undefined;
mockCancelSteer.mockImplementationOnce(
() =>
new Promise<{ removed: boolean }>((resolve) => {
settleCancel = resolve;
}),
);
const recovered = {
id: 'q1',
text: 'being removed',
createdAt: 1,
recoverySteerId: 'server-source',
recoveryClientSteerId: 'client-source',
};
const { result } = setupClearAll(({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [recovered]);
});
let pending: Promise<boolean> | undefined;
act(() => {
pending = result.current.steering.discardQueued(recovered);
});
// Nothing for a drain to find while the receipt is in flight.
expect(result.current.queue).toEqual([]);
let settled = false;
await act(async () => {
settleCancel?.({ removed: true });
settled = (await pending) ?? false;
});
expect(settled).toBe(true);
// Back in its slot, downgraded: the parked copy is gone, so it is an
// ordinary local row the caller can edit or remove.
expect(result.current.queue).toEqual([
expect.objectContaining({ id: 'q1', text: 'being removed' }),
]);
expect(result.current.queue[0].recoverySteerId).toBeUndefined();
expect(result.current.queue[0].recoveryClientSteerId).toBeUndefined();
});
it('returns a recovered row untouched when its source refuses to cancel', async () => {
mockCancelSteer.mockResolvedValueOnce({ removed: false, generationProtocolVersion: 2 });
const recovered = {
id: 'q1',
text: 'stays put',
createdAt: 1,
recoverySteerId: 'server-source',
recoveryClientSteerId: 'client-source',
};
const { result } = setupClearAll(({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [recovered]);
});
let settled = true;
await act(async () => {
settled = await result.current.steering.discardQueued(recovered);
});
expect(settled).toBe(false);
expect(result.current.queue).toEqual([
expect.objectContaining({ id: 'q1', recoverySteerId: 'server-source' }),
]);
});
/** Retiring twice must not resurrect a parked epoch: the second call has to
* leave an existing tombstone standing. */
it('keeps a cancelled tombstone across a second queue action', () => {
const { result } = setupClearAll(({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [
{ id: 'q1', text: 'first', createdAt: 1 },
{ id: 'q2', text: 'second', createdAt: 2 },
]);
set(store.queueDrainHoldByConvoId(CONVO_ID), {
runEnd: { conversationId: CONVO_ID, outcome: 'completed', endedAt: 1 },
dueAt: 1,
status: 'cancelled',
});
});
act(() => {
result.current.steering.removeQueued('q1');
});
expect(result.current.drainHold).toMatchObject({ status: 'cancelled' });
act(() => {
result.current.steering.removeQueued('q2');
});
expect(result.current.drainHold).toMatchObject({ status: 'cancelled' });
});
/** A standing window would otherwise fire on whatever the fallbacks put back
* — sending exactly what the user just cleared. */
it('cancels a pending automatic send as part of clearing', async () => {
const { result } = setupClearAll(({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [{ id: 'q1', text: 'cleared', createdAt: 1 }]);
set(store.queueDrainHoldByConvoId(CONVO_ID), {
runEnd: { conversationId: CONVO_ID, outcome: 'completed', endedAt: 1 },
dueAt: Date.now() + 3000,
});
});
await act(async () => {
await result.current.steering.clearQueued();
});
expect(result.current.drainHold).toBeNull();
});
/** The rows must leave the queue BEFORE the cancellations are awaited: a run
* ending mid-clear would otherwise drain one of the very messages the user
* asked to take back. */
it('empties the queue up front rather than during the cancellations', () => {
let settleCancel: ((value: { removed: boolean }) => void) | undefined;
mockCancelSteer.mockImplementationOnce(
() =>
new Promise<{ removed: boolean }>((resolve) => {
settleCancel = resolve;
}),
);
const { result } = setupClearAll(({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [
{
id: 'q1',
text: 'recovered row',
createdAt: 1,
recoverySteerId: 'server-source',
recoveryClientSteerId: 'client-source',
},
{ id: 'q2', text: 'ordinary row', createdAt: 2 },
]);
});
act(() => {
void result.current.steering.clearQueued();
});
// Nothing left to drain while the cancellation is still in flight.
expect(result.current.queue).toEqual([]);
settleCancel?.({ removed: true });
});
it('puts a row back when its parked source refuses to cancel', async () => {
mockCancelSteer.mockResolvedValueOnce({ removed: false, generationProtocolVersion: 2 });
const { result } = setupClearAll(({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [
{
id: 'q1',
text: 'stuck row',
createdAt: 1,
recoverySteerId: 'server-source',
recoveryClientSteerId: 'client-source',
},
{ id: 'q2', text: 'ordinary row', createdAt: 2 },
]);
});
let cleared: QueuedMessage | null | undefined;
await act(async () => {
cleared = await result.current.steering.clearQueued();
});
// Only the row the client fully owns went to the composer.
expect(cleared?.text).toBe('ordinary row');
expect(result.current.queue).toEqual([
expect.objectContaining({ id: 'q1', recoverySteerId: 'server-source' }),
]);
});
/** Cancelling a parked source is a round trip, so the queue can grow while
* clear-all is in flight; a message the user queued meanwhile is not part
* of what they asked to clear. */
it('leaves a message queued after the clear started untouched', async () => {
let settleCancel: ((value: { removed: boolean }) => void) | undefined;
mockCancelSteer.mockImplementationOnce(
() =>
new Promise<{ removed: boolean }>((resolve) => {
settleCancel = resolve;
}),
);
const { result } = setupClearAll(({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [
{
id: 'q1',
text: 'recovered row',
createdAt: 1,
recoverySteerId: 'server-source',
recoveryClientSteerId: 'client-source',
},
{ id: 'q2', text: 'ordinary row', createdAt: 2 },
]);
});
let pending: Promise<QueuedMessage | null> | undefined;
act(() => {
pending = result.current.steering.clearQueued();
});
// The user queues something else while the cancellation is in flight.
act(() => {
result.current.steering.enqueue('queued mid-clear', { id: 'q3', createdAt: 3 });
});
let cleared: QueuedMessage | null | undefined;
await act(async () => {
settleCancel?.({ removed: true });
cleared = (await pending) ?? null;
});
expect(cleared?.text).toBe('recovered row\n\nordinary row');
expect(result.current.queue).toEqual([
expect.objectContaining({ id: 'q3', text: 'queued mid-clear' }),
]);
});
});

118
client/src/hooks/Chat/useQueueDrain.ts
View file

@ -3,6 +3,7 @@ import { Constants } from 'librechat-data-provider';
import { useRecoilValue, useRecoilCallback } from 'recoil';
import type { DrainAfterAbort, QueuedMessage, QueuedMessageOrigin, RunEnd } from '~/store/families';
import type { TAskFunction } from '~/common';
import { compareQueuedMessages, isSameRunEpoch, isSendableQueuedMessage } from '~/utils';
import { useMarkFilesUsageMutation } from '~/data-provider';
import store from '~/store';
@ -36,8 +37,8 @@ const batchFileIds = (fileIds: string[]): string[][] => {
return batches;
};
const compareQueuedMessages = (a: QueuedMessage, b: QueuedMessage): number =>
Number(b.priority ?? false) - Number(a.priority ?? false) || a.createdAt - b.createdAt;
/** Gmail-style window to take an automatic send back before it fires. */
export const QUEUE_UNDO_GRACE_MS = 3000;
/** Interrupt intent belongs to one generation, never to whichever terminal
* event happens to occupy the shared pane slot next. The NEW_CONVO alias is
@ -67,17 +68,27 @@ const matchesInterruptArm = (armed: DrainAfterAbort | false, end: RunEnd): boole
* the next, so multi-message queues send FIFO in sequence.
* - Migrates a queue keyed under `NEW_CONVO` to the real conversation id when
* the finished run started as a new-conversation submission.
* - Withholds an automatic send for `undoGraceMs`: the terminal epoch is
* parked in `queueDrainHoldByConvoId` and re-posted when the window closes.
* The queue itself is untouched while a hold stands, so "undo" is a pure
* cancel — nothing to restore, nothing to lose. Manual sends and armed
* interrupts skip the grace: both are the user asking for it NOW.
*/
export default function useQueueDrain(
index: string | number,
activeConversationId: string | undefined,
ask: TAskFunction,
options?: { undoGraceMs?: number },
) {
const undoGraceMs = options?.undoGraceMs ?? QUEUE_UNDO_GRACE_MS;
const runEnd = useRecoilValue(store.runEndByIndex(index));
const parkedRunEnd = useRecoilValue(
store.pendingRunEndByConvoId(activeConversationId ?? Constants.NEW_CONVO),
);
const isSubmitting = useRecoilValue(store.isSubmittingFamily(index));
const drainHold = useRecoilValue(
store.queueDrainHoldByConvoId(activeConversationId ?? Constants.NEW_CONVO),
);
const { mutate: markFilesUsage } = useMarkFilesUsageMutation();
const ownQueue = useRecoilValue(
store.queuedMessagesByConvoId(activeConversationId ?? Constants.NEW_CONVO),
@ -205,6 +216,27 @@ export default function useQueueDrain(
if (end == null) {
return null;
}
/** A standing window short-circuits BEFORE anything is consumed or
* written: both signal carriers retain unconsumed epochs, so leaving
* this one in place is lossless — and mutating nothing is what keeps
* the effect from re-running itself into a loop while it waits.
*
* An armed interrupt is exempt. The user aborted a run to say this
* now, and the standing window may belong to an unrelated earlier
* epoch; making it wait out that timer contradicts the rule that
* armed interrupts skip the grace. Read-only, so the no-loop
* guarantee holds — the arm is consumed on the normal path below. */
if (end.conversationId != null) {
const standing = snapshot
.getLoadable(store.queueDrainHoldByConvoId(end.conversationId))
.getValue();
const armedNow = snapshot.getLoadable(store.drainAfterAbortByIndex(index)).getValue();
const interruptWaiting =
matchesInterruptArm(armedNow, end) || end.interruptArmed === true;
if (standing != null && standing.dueAt > Date.now() && !interruptWaiting) {
return null;
}
}
// Consume the signal first — a hard double-fire guard even if the
// effect re-runs before Recoil propagates.
if (fromParked && activeConversationId) {
@ -241,7 +273,50 @@ export default function useQueueDrain(
: ownQueue;
const shouldDrain = end.outcome === 'completed' || interruptArmed;
const next = shouldDrain ? (merged[0] ?? null) : null;
const hold = snapshot.getLoadable(store.queueDrainHoldByConvoId(conversationId)).getValue();
const now = Date.now();
/** A hold owns exactly ONE epoch. A newer terminal event can be the one
* consumed above, and retiring the hold (or discarding that event) on
* its behalf would lose the newer signal and leave the older one to
* reopen a window — so act only on the epoch the hold actually holds. */
const holdOwnsEnd = hold != null && isSameRunEpoch(hold.runEnd, end);
if (holdOwnsEnd) {
set(store.queueDrainHoldByConvoId(conversationId), null);
}
/** Undo landed after the timer handed this epoch back. The epoch is
* consumed above, so discarding it here is what makes the visible
* Undo mean what it says. */
if (holdOwnsEnd && hold?.status === 'cancelled') {
return null;
}
/** Withhold the epoch, never the queue: the rows stay exactly where
* they are, so cancelling is a no-op and a reload costs no text.
* A first turn's queue does have to migrate before the window opens —
* the composer reads the resolved id by then, so leaving the rows
* under `NEW_CONVO` would blank the group for the whole window. */
if (
shouldDrain &&
!interruptArmed &&
hold == null &&
undoGraceMs > 0 &&
merged.length > 0
) {
if (shouldMigrate && newConvoQueue.length > 0) {
set(store.queuedMessagesByConvoId(Constants.NEW_CONVO), []);
set(store.queuedMessagesByConvoId(conversationId), merged);
}
set(store.queueDrainHoldByConvoId(conversationId), {
runEnd: end,
dueAt: now + undoGraceMs,
});
return null;
}
/** A row mid-edit can be blank, and blank is not a message. Skipping to
* the row behind it would send out of order, so this epoch simply
* drains nothing — consumed above, so the effect cannot spin — and the
* next run end picks the queue up again. */
const front = merged[0] ?? null;
const next = shouldDrain && front != null && isSendableQueuedMessage(front) ? front : null;
const remainder = next ? merged.slice(1) : merged;
if (shouldMigrate && newConvoQueue.length > 0) {
@ -267,6 +342,39 @@ export default function useQueueDrain(
[index, activeConversationId],
);
/** Hands the held epoch back to the drain through the same parked slot a
* returning navigation uses, so an expired window resumes identically
* whether or not the user stayed on the conversation. */
const releaseDrainHold = useRecoilCallback(
({ snapshot, set }) =>
(convoId: string) => {
const held = snapshot.getLoadable(store.queueDrainHoldByConvoId(convoId)).getValue();
if (held == null || held.status != null) {
return;
}
/** Marked before the epoch goes back so Undo stops being offered the
* moment the window closes — a new run can block the drain, and an
* Undo that no longer undoes anything is worse than none. */
set(store.queueDrainHoldByConvoId(convoId), { ...held, status: 'released' });
set(store.pendingRunEndByConvoId(convoId), held.runEnd);
},
[],
);
useEffect(() => {
if (drainHold == null || drainHold.status != null) {
return;
}
const convoId = activeConversationId ?? Constants.NEW_CONVO;
const remaining = drainHold.dueAt - Date.now();
if (remaining <= 0) {
releaseDrainHold(convoId);
return;
}
const timer = setTimeout(() => releaseDrainHold(convoId), remaining);
return () => clearTimeout(timer);
}, [drainHold, activeConversationId, releaseDrainHold]);
const restoreQueued = useRecoilCallback(
({ set }) =>
(convoId: string, item: QueuedMessage) => {
@ -334,6 +442,10 @@ export default function useQueueDrain(
}, [
runEnd,
parkedRunEnd,
/** A standing window leaves later epochs untouched, so removing it has to
* bring the drain back to reconsider them — otherwise a follow-up sits
* stranded until some unrelated dependency happens to change. */
drainHold,
isSubmitting,
activeConversationId,
parkForeignRunEnd,

11
client/src/hooks/Chat/useSteerConvert.ts
View file

@ -5,7 +5,12 @@ import type { TPendingSteer } from 'librechat-data-provider';
import type { QueuedMessage, QueuedMessageOrigin } from '~/store/families';
import type { GenerationProtocolVersion } from '~/data-provider';
import type { SteerCarriedContext } from '~/utils';
import { appendAppliedSteerIds, carriedSteerContext, insertQueuedOrigin } from '~/utils';
import {
appendAppliedSteerIds,
carriedSteerContext,
compareQueuedMessages,
insertQueuedOrigin,
} from '~/utils';
import { fetchStreamStatus, getGenerationProtocolVersion } from '~/data-provider';
import store from '~/store';
@ -165,9 +170,7 @@ export default function useSteerConvert() {
* captured position instead of being re-minted under the server id. */
const ordinary = fresh.filter(({ queuedOrigin }) => queuedOrigin == null);
let merged: QueuedMessage[] = [...existing, ...ordinary.map(({ item }) => item)].sort(
(a, b) =>
Number(b.priority ?? false) - Number(a.priority ?? false) ||
a.createdAt - b.createdAt,
compareQueuedMessages,
);
for (const { queuedOrigin } of fresh) {
if (queuedOrigin != null) {

325
client/src/hooks/Chat/useSteering.ts
View file

@ -7,6 +7,18 @@ import type { TMessage, TConversation, TMessageContentParts } from 'librechat-da
import type { RunEnd, PendingSteer, QueuedMessage, QueuedMessageOrigin } from '~/store/families';
import type { GenerationProtocolVersion } from '~/data-provider';
import type { ExtendedFile, FileSetter } from '~/common';
import {
appendAppliedSteerIds,
bumpQueuedMessage,
carriedSteerContext,
clearAllDrafts,
compareQueuedMessages,
insertQueuedOrigin,
isMergeableQueuedMessage,
isSameRunEpoch,
isSendableQueuedMessage,
mergeQueuedMessages,
} from '~/utils';
import {
useGetMessagesByConvoId,
useCancelSteerMutation,
@ -14,12 +26,6 @@ import {
useMarkFilesUsageMutation,
supportsGenerationProtocolV2,
} from '~/data-provider';
import {
appendAppliedSteerIds,
carriedSteerContext,
clearAllDrafts,
insertQueuedOrigin,
} from '~/utils';
import useSteerConvert from '~/hooks/Chat/useSteerConvert';
import { useSetFilesToDelete } from '~/hooks/Files';
import useLocalize from '~/hooks/useLocalize';
@ -83,16 +89,6 @@ function isDefiniteSteerRejection(error: unknown): boolean {
);
}
function isSameRunEpoch(a: RunEnd | null, b: RunEnd): boolean {
if (a == null || a.conversationId !== b.conversationId) {
return false;
}
if (a.generationCreatedAt != null || b.generationCreatedAt != null) {
return a.generationCreatedAt === b.generationCreatedAt;
}
return a.endedAt === b.endedAt;
}
/** True when the latest assistant message carries an unresolved tool approval —
* the run is (or is about to be) paused, so a steer POST would 409. */
function hasLiveToolApproval(messages: TMessage[] | undefined): boolean {
@ -559,11 +555,7 @@ export default function useSteering({
...(options?.front && { priority: true }),
};
set(store.queuedMessagesByConvoId(queueKey), (prev) =>
[...prev, item].sort(
(a, b) =>
Number(b.priority ?? false) - Number(a.priority ?? false) ||
a.createdAt - b.createdAt,
),
[...prev, item].sort(compareQueuedMessages),
);
if (options?.skipUsageMark !== true) {
markQueuedFilesUsage(options?.files);
@ -633,53 +625,128 @@ export default function useSteering({
clearAllDrafts(Constants.PENDING_CONVO);
}, []);
const removeQueued = useRecoilCallback(
({ set }) =>
(id: string) => {
set(store.queuedMessagesByConvoId(queueKey), (prev) =>
prev.filter((item) => item.id !== id),
/**
* Stops a pending automatic send. A window still counting down is simply
* dropped, but once the timer has handed its epoch back the hold is the only
* thing that can neutralize it — so that case leaves a tombstone the drain
* consumes, rather than a null that lets the parked epoch through.
*/
const retireDrainHold = useRecoilCallback(
({ snapshot, set }) =>
() => {
const held = snapshot.getLoadable(store.queueDrainHoldByConvoId(queueKey)).getValue();
if (held == null) {
return;
}
/** Already neutralized and waiting for its parked epoch to be consumed.
* Clearing it here would expose that epoch again, so a second queue
* action must leave the tombstone standing. */
if (held.status === 'cancelled') {
return;
}
set(
store.queueDrainHoldByConvoId(queueKey),
held.status === 'released' ? { ...held, status: 'cancelled' } : null,
);
},
[queueKey],
);
/** Once a parked source is discarded it must never be retried as a recovery
* attempt. Downgrade the row in place so a guarded Edit that finds a newer
* draft can leave the same words, context, identity, and queue position as
* an ordinary local follow-up. */
const downgradeQueuedRecovery = useRecoilCallback(
const removeQueued = useRecoilCallback(
({ snapshot, set }) =>
(id: string): boolean => {
(id: string) => {
const queue = snapshot.getLoadable(store.queuedMessagesByConvoId(queueKey)).getValue();
let found = false;
const next = queue.map((item) => {
if (item.id !== id) {
return item;
}
found = true;
const {
clientRequestId: _clientRequestId,
recoverySteerId: _recoverySteerId,
recoveryClientSteerId: _recoveryClientSteerId,
...ordinary
} = item;
return ordinary;
});
if (found) {
set(store.queuedMessagesByConvoId(queueKey), next);
const next = queue.filter((item) => item.id !== id);
if (next.length === queue.length) {
return;
}
return found;
set(store.queuedMessagesByConvoId(queueKey), next);
/** Nothing left to auto-send: retiring the window here stops a stale
* completion from later draining a row queued under a NEWER run. */
if (next.length === 0) {
retireDrainHold();
}
},
[queueKey, retireDrainHold],
);
/** "Send next". Order lives in the sort key rather than the array so the
* choice survives the next enqueue, drain, or leftover conversion. */
const bumpQueued = useRecoilCallback(
({ set }) =>
(id: string) => {
set(store.queuedMessagesByConvoId(queueKey), (prev) =>
bumpQueuedMessage(prev, id, Date.now()),
);
},
[queueKey],
);
/** In-place rewrite of a waiting row. Refuses a recovery-bound row: its
* parked source is matched by exact text server-side, so the words may only
* change after `discardQueued` has downgraded it to an ordinary row. */
const updateQueuedText = useRecoilCallback(
({ snapshot, set }) =>
(id: string, text: string): boolean => {
const queue = snapshot.getLoadable(store.queuedMessagesByConvoId(queueKey)).getValue();
const target = queue.find((item) => item.id === id);
if (target == null || !isMergeableQueuedMessage(target)) {
return false;
}
/** Blank is recorded, not refused. Refusing it was the root of a whole
* family of bugs: the queue kept words the screen no longer showed, and
* every reader — the drain, Send now, the escalate shortcut, Merge,
* Clear all, the trash — had to be taught about that disagreement.
* Now the row simply is not sendable, which each reader can see for
* itself via `isSendableQueuedMessage`. */
if (target.text === text) {
return true;
}
set(
store.queuedMessagesByConvoId(queueKey),
queue.map((item) => (item.id === id ? { ...item, text } : item)),
);
return true;
},
[queueKey],
);
/** Folds every ordinary waiting row into one turn. All-or-nothing: a batch
* holding a recovery-bound row is refused rather than partially folded, so
* no parked source is ever stranded behind a merged message. */
const mergeQueued = useRecoilCallback(
({ snapshot, set }) =>
(): boolean => {
const queue = snapshot.getLoadable(store.queuedMessagesByConvoId(queueKey)).getValue();
const merged = mergeQueuedMessages(queue);
if (merged == null) {
return false;
}
set(store.queuedMessagesByConvoId(queueKey), [merged]);
return true;
},
[queueKey],
);
/** Cancels a withheld automatic send. The queue was never popped, so this
* only drops the epoch — the rows stay put for a manual send. If the timer
* already handed the epoch back (a new run can block the drain, leaving it
* parked), dropping the hold alone would let it drain anyway: tombstone it
* instead so the drain discards that epoch when it finally runs. */
/** Cancels a withheld automatic send. The queue was never popped, so the
* rows simply stay put for a manual send. */
const cancelQueueDrain = retireDrainHold;
/** Settle a queued row's terminal recovery source before an Edit/Remove.
* Ordinary rows have no server copy. A v2 leftover first uses its durable
* receipt to atomically discard the parked copy, then becomes an ordinary
* local row; the caller decides whether the live composer can consume it.
* The cancel deliberately omits the active epoch: the receipt belongs to the
* terminal source generation, not whichever run now occupies the chat. */
const discardQueued = useCallback(
/** Cancels a row's parked server copy by receipt, independent of whether the
* row is still in the queue — clear-all takes its rows out first, so the
* downgrade step cannot be the thing that reports success. */
const cancelParkedSource = useCallback(
async (item: QueuedMessage): Promise<boolean> => {
if (item.recoverySteerId == null) {
return true;
@ -698,15 +765,99 @@ export default function useSteering({
showToast({ message: localize('com_ui_steer_cancel_failed'), status: 'error' });
return false;
}
return downgradeQueuedRecovery(item.id);
return true;
} catch {
showToast({ message: localize('com_ui_steer_cancel_failed'), status: 'error' });
return false;
}
},
[cancelSteer, conversationId, downgradeQueuedRecovery, hasRealConvoId, localize, showToast],
[cancelSteer, conversationId, hasRealConvoId, localize, showToast],
);
/** Empties the queue and hands back what it held. Removing up front is the
* point: the parked-source cancellations below are round trips, and rows
* left in place could be drained by a run ending mid-clear — sending the
* very messages the user asked to take back. */
const takeAllQueued = useRecoilCallback(
({ snapshot, set }) =>
(): QueuedMessage[] => {
const queue = snapshot.getLoadable(store.queuedMessagesByConvoId(queueKey)).getValue();
if (queue.length > 0) {
set(store.queuedMessagesByConvoId(queueKey), []);
}
return queue;
},
[queueKey],
);
/**
* Puts rows back without clobbering anything queued in the meantime. Also the
* public requeue path for a cleared payload the composer refused: passing the
* ITEM back preserves every field it carries, where rebuilding one from parts
* has twice now lost a field that mattered (the predecessor fence, then the
* interrupt tier).
*/
const restoreQueuedBatch = useRecoilCallback(
({ set }) =>
(items: QueuedMessage[]) => {
if (items.length === 0) {
return;
}
set(store.queuedMessagesByConvoId(queueKey), (prev) => {
const held = new Set(prev.map((item) => item.id));
const missing = items.filter((item) => !held.has(item.id));
return missing.length === 0 ? prev : [...prev, ...missing].sort(compareQueuedMessages);
});
},
[queueKey],
);
/**
* "Clear all": hands the waiting words back to the user rather than deleting
* them. Rows leave the queue immediately, then each parked server copy is
* cancelled by receipt — a surviving copy would be re-offered as a fresh row
* on the next status read. A row whose cancellation fails goes back in the
* queue; its words are never dropped on the floor. The caller owns the
* composer and decides whether it can accept the payload.
*/
const clearQueued = useCallback(async (): Promise<QueuedMessage | null> => {
/** Taking the words back cancels any pending automatic send outright. A
* standing window would otherwise fire on whatever the fallbacks put back
* — a requeued payload the composer refused, or a row whose parked source
* would not cancel — sending exactly what was just cleared. */
cancelQueueDrain();
const taken = takeAllQueued();
if (taken.length === 0) {
return null;
}
const cleared: QueuedMessage[] = [];
const stuck: QueuedMessage[] = [];
for (const item of taken) {
if (item.recoverySteerId == null) {
cleared.push(item);
continue;
}
if (!(await cancelParkedSource(item))) {
stuck.push(item);
continue;
}
/** The parked copy is gone, so the row is an ordinary local follow-up
* now — and only ordinary rows can be folded together. */
const {
clientRequestId: _clientRequestId,
recoverySteerId: _recoverySteerId,
recoveryClientSteerId: _recoveryClientSteerId,
...ordinary
} = item;
cleared.push(ordinary);
}
restoreQueuedBatch(stuck);
if (cleared.length === 0) {
return null;
}
return cleared.length === 1 ? cleared[0] : mergeQueuedMessages(cleared);
}, [cancelQueueDrain, takeAllQueued, cancelParkedSource, restoreQueuedBatch]);
/** Capture-then-remove, including the item's neighbours, so any refused send
* or rejected steer can restore the ORIGINAL item in place even if the run
* drains an adjacent entry while the request is in flight. */
@ -736,11 +887,17 @@ export default function useSteering({
};
queuedOrigins.set(id, origin);
set(store.queuedMessagesByConvoId(queueKey), (prev) =>
prev.filter((item) => item.id !== id),
prev.filter((queued) => queued.id !== id),
);
/** A withheld epoch grants exactly ONE drain, and expediting a row is
* the user spending it themselves — the manually started run's own
* completion drains whatever comes next. Leaving the window armed for
* any remaining rows would send twice for one run end, and would do it
* even when that manual run was stopped or errored. */
retireDrainHold();
return origin;
},
[queueKey],
[queueKey, retireDrainHold],
);
const releaseQueuedOrigin = useCallback(
@ -778,6 +935,40 @@ export default function useSteering({
[queueKey, releaseQueuedOrigin],
);
/**
* Settles a row's parked source for an Edit or Remove, holding the row OUT of
* the queue for the round trip. Leaving it in place let a run completing
* mid-cancellation drain and send the very message being removed — the same
* hazard clear-all avoids by taking its rows up front. Cancellation refused:
* the row returns to its original slot untouched. Cancellation settled: it
* returns already downgraded, so what the caller does next (hand the words to
* the composer, remove it) sees an ordinary local row with no parked copy.
*/
const discardQueued = useCallback(
async (item: QueuedMessage): Promise<boolean> => {
if (item.recoverySteerId == null) {
return true;
}
const origin = takeQueued(item.id);
if (origin == null) {
return false;
}
if (!(await cancelParkedSource(item))) {
restoreQueued(origin);
return false;
}
const {
clientRequestId: _clientRequestId,
recoverySteerId: _recoverySteerId,
recoveryClientSteerId: _recoveryClientSteerId,
...ordinary
} = origin.item;
restoreQueued({ ...origin, item: ordinary });
return true;
},
[cancelParkedSource, restoreQueued, takeQueued],
);
/**
* Re-posts a spent run-end signal so the drain wakes and reconsiders the
* queue. No-op while a signal for THIS conversation is still armed: that
@ -794,7 +985,15 @@ export default function useSteering({
(convoId: string, end: RunEnd) => {
const indexArmed = snapshot.getLoadable(store.runEndByIndex(index)).getValue();
const parkedArmed = snapshot.getLoadable(store.pendingRunEndByConvoId(convoId)).getValue();
if (isSameRunEpoch(indexArmed, end) || isSameRunEpoch(parkedArmed, end)) {
/** A withheld epoch lives in the hold, in neither carrier — without
* this the re-arm appends a copy and the release appends the held one,
* leaving two identical completions to drain two rows. */
const held = snapshot.getLoadable(store.queueDrainHoldByConvoId(convoId)).getValue();
if (
isSameRunEpoch(indexArmed, end) ||
isSameRunEpoch(parkedArmed, end) ||
isSameRunEpoch(held?.runEnd, end)
) {
return;
}
set(store.pendingRunEndByConvoId(convoId), end);
@ -1275,6 +1474,10 @@ export default function useSteering({
if (isSubmitting && (!duringRunActive || !canSteer || item.recoverySteerId != null)) {
return;
}
/** Nothing to send: the row is mid-edit and currently blank. */
if (!isSendableQueuedMessage(item)) {
return;
}
/** UI callers always find the item; a stale/direct caller has no original
* neighbours, so restoration falls back to the queue's priority split. */
const origin = takeQueued(item.id) ?? { item, beforeIds: [], afterIds: [] };
@ -1460,8 +1663,14 @@ export default function useSteering({
convertSteerToQueue,
queueReclaimedSteer,
enqueue,
requeueCleared: restoreQueuedBatch,
removeQueued,
discardQueued,
bumpQueued,
updateQueuedText,
mergeQueued,
clearQueued,
cancelQueueDrain,
sendQueuedNow,
interruptAndSend,
interruptSteer,
@ -1486,8 +1695,14 @@ export default function useSteering({
convertSteerToQueue,
queueReclaimedSteer,
enqueue,
restoreQueuedBatch,
removeQueued,
discardQueued,
bumpQueued,
updateQueuedText,
mergeQueued,
clearQueued,
cancelQueueDrain,
sendQueuedNow,
interruptAndSend,
interruptSteer,

11
client/src/locales/en/translation.json
View file

@ -1617,7 +1617,18 @@
"com_ui_question_failed_description": "The agent couldn't show this question and may retry automatically.",
"com_ui_question_unanswered": "No answer was given",
"com_ui_queue": "Queue",
"com_ui_queue_clear_all": "Clear all",
"com_ui_queue_count": "{{0}} queued",
"com_ui_queue_edit_empty": "Type a message, or press Escape to keep the original",
"com_ui_queue_edit_inline": "Edit this queued message",
"com_ui_queue_escalate_newest": "Interrupt the run and send the newest queued message",
"com_ui_queue_merge": "Merge into one message",
"com_ui_queue_merge_blocked": "Recovered messages can't be merged — remove or send them first",
"com_ui_queue_next_up": "Next: {{0}}",
"com_ui_queue_send": "Queue message for after the response",
"com_ui_queue_send_next": "Send this one next",
"com_ui_queue_sending": "Sending 1 of {{0}}…",
"com_ui_queue_undo_send": "Undo",
"com_ui_queued_attachment_count": "{{0}} attachments queued with this message",
"com_ui_queued_messages": "Queued messages",
"com_ui_quote_selections": "{{0}} selections",

27
client/src/store/families.ts
View file

@ -387,6 +387,10 @@ export type QueuedMessage = {
/** Front-inserted by "Interrupt & send": stays ahead of chronologically
* older items when leftover steers are merged back into the queue. */
priority?: boolean;
/** Set by "Send next": promotes the row above ordinary follow-ups (but never
* above an interrupt), most recent promotion first. Absent on rows the user
* never reordered. */
bumpedAt?: number;
};
/** Snapshot of a queued item's logical position while it is temporarily sent
@ -408,6 +412,28 @@ const queuedMessagesByConvoId = atomFamily<QueuedMessage[], string>({
default: [],
});
/**
* A terminal epoch withheld from the drain for its undo grace. The queue is
* NOT popped while a hold stands — only the epoch is parked — so cancelling
* costs nothing and a reload during the window loses no text. `dueAt` is
* absolute so a hold survives leaving and re-entering the conversation.
*/
export type QueueDrainHold = {
runEnd: RunEnd;
dueAt: number;
/** Absent while the window is open and Undo is offered. `released` means the
* epoch was handed back to the drain (the window is over, so Undo is no
* longer advertised); `cancelled` neutralizes an epoch already handed back,
* for the frame in which Undo and release can race. */
status?: 'released' | 'cancelled';
};
/** Per-conversation undo grace on the next automatic drain. */
const queueDrainHoldByConvoId = atomFamily<QueueDrainHold | null, string>({
key: 'queueDrainHoldByConvoId',
default: null,
});
/**
* One-shot run-termination signal written by the SSE final/error handlers and
* consumed (reset to null) by `useQueueDrain`. Keyed by chat index like
@ -714,6 +740,7 @@ export default {
pendingQuotesByConvoId,
pendingSteersByConvoId,
queuedMessagesByConvoId,
queueDrainHoldByConvoId,
runEndByIndex,
pendingRunEndByConvoId,
drainAfterAbortByIndex,

8
client/src/store/steer.ts
View file

@ -20,3 +20,11 @@ export const steerOverlayHeightFamily = atomFamily((_conversationId: string) =>
* and the chip-derived check cannot see an arm until its response lands.
*/
export const escalatingSteerFamily = atomFamily((_conversationId: string) => atom<boolean>(false));
/**
* Whether the queued-message outbox is expanded, per queue key (so a brand-new
* chat and its resolved id are distinct, matching `useSteering.queueKey`).
* Collapsed by default and deliberately in memory only: the queue itself does
* not survive a reload, so persisting the disclosure would outlive its subject.
*/
export const queueExpandedFamily = atomFamily((_queueKey: string) => atom<boolean>(false));

243
client/src/utils/__tests__/steer.spec.ts
View file

@ -1,13 +1,20 @@
import { Constants, ContentTypes } from 'librechat-data-provider';
import type { TMessage, TSteerAppliedEvent } from 'librechat-data-provider';
import type { QueuedMessage } from '~/store/families';
import {
isSameRunEpoch,
isSendableQueuedMessage,
getSteerPart,
applySteerPart,
resolveRunEndTarget,
bumpQueuedMessage,
mergeQueuedMessages,
findSteerMessageIndex,
appendAppliedSteerIds,
resolveAbortSteerTarget,
compareQueuedMessages,
insertQueuedOrigin,
isMergeableQueuedMessage,
} from '../steer';
const buildEvent = (overrides: Partial<TSteerAppliedEvent> = {}): TSteerAppliedEvent => ({
@ -218,3 +225,239 @@ describe('insertQueuedOrigin', () => {
]);
});
});
const queued = (overrides: Partial<QueuedMessage> & { id: string }): QueuedMessage => ({
text: `text ${overrides.id}`,
createdAt: 1,
...overrides,
});
describe('insertQueuedOrigin — promotions made while a row was away', () => {
/** B leaves for a steer attempt, C is promoted in its absence, then the
* attempt fails: B's captured neighbours must not put it back in front of
* the row the user explicitly chose to send next. */
it('never restores a row ahead of a promotion it did not know about', () => {
const origin = {
item: { id: 'b', text: 'send B', createdAt: 2 },
beforeIds: ['a'],
afterIds: ['c'],
};
const promoted = bumpQueuedMessage(
[
{ id: 'a', text: 'send A', createdAt: 1 },
{ id: 'c', text: 'send C', createdAt: 3 },
],
'c',
500,
);
expect(insertQueuedOrigin(promoted, origin).map(({ id }) => id)).toEqual(['c', 'a', 'b']);
});
it('still honours captured neighbours when nothing was promoted', () => {
const origin = {
item: { id: 'b', text: 'send B', createdAt: 2 },
beforeIds: ['a'],
afterIds: ['c'],
};
const queue = [
{ id: 'a', text: 'send A', createdAt: 1 },
{ id: 'c', text: 'send C', createdAt: 3 },
];
expect(insertQueuedOrigin(queue, origin).map(({ id }) => id)).toEqual(['a', 'b', 'c']);
});
});
describe('isSameRunEpoch', () => {
const epoch = { conversationId: 'c1', outcome: 'completed' as const, endedAt: 10 };
it('matches on the generation stamp when either side carries one', () => {
expect(
isSameRunEpoch(
{ ...epoch, generationCreatedAt: 41 },
{ ...epoch, generationCreatedAt: 41, endedAt: 99 },
),
).toBe(true);
expect(
isSameRunEpoch({ ...epoch, generationCreatedAt: 41 }, { ...epoch, generationCreatedAt: 42 }),
).toBe(false);
expect(isSameRunEpoch({ ...epoch, generationCreatedAt: 41 }, epoch)).toBe(false);
});
it('falls back to the termination time, and never matches across conversations', () => {
expect(isSameRunEpoch(epoch, { ...epoch })).toBe(true);
expect(isSameRunEpoch(epoch, { ...epoch, endedAt: 11 })).toBe(false);
expect(isSameRunEpoch(epoch, { ...epoch, conversationId: 'c2' })).toBe(false);
});
it('treats an absent signal as no match', () => {
expect(isSameRunEpoch(null, epoch)).toBe(false);
expect(isSameRunEpoch(undefined, epoch)).toBe(false);
});
});
describe('compareQueuedMessages', () => {
const ids = (items: QueuedMessage[]) =>
[...items].sort(compareQueuedMessages).map(({ id }) => id);
it('orders by enqueue time by default', () => {
expect(ids([queued({ id: 'b', createdAt: 2 }), queued({ id: 'a', createdAt: 1 })])).toEqual([
'a',
'b',
]);
});
it('keeps interrupt front-inserts ahead of chronologically older rows', () => {
expect(
ids([
queued({ id: 'old', createdAt: 1 }),
queued({ id: 'armed', createdAt: 9, priority: true }),
]),
).toEqual(['armed', 'old']);
});
it('drains the most recently promoted row first', () => {
const items = [
queued({ id: 'first-bump', createdAt: 1, bumpedAt: 100 }),
queued({ id: 'second-bump', createdAt: 2, bumpedAt: 200 }),
queued({ id: 'plain', createdAt: 3 }),
];
expect(ids(items)).toEqual(['second-bump', 'first-bump', 'plain']);
});
/** An interrupt aborted a run to be said now, so it outranks a promotion
* whichever came first — and two interrupts stay FIFO among themselves. */
it('keeps interrupts ahead of promotions in both arrival orders', () => {
expect(
ids([
queued({ id: 'bumped', createdAt: 1, bumpedAt: 500 }),
queued({ id: 'armed', createdAt: 5, priority: true }),
]),
).toEqual(['armed', 'bumped']);
expect(
ids([
queued({ id: 'armed', createdAt: 1, priority: true }),
queued({ id: 'bumped', createdAt: 5, bumpedAt: 500 }),
]),
).toEqual(['armed', 'bumped']);
});
it('keeps two interrupts FIFO, as sequential instructions rather than rivals', () => {
expect(
ids([
queued({ id: 'second', createdAt: 20, priority: true }),
queued({ id: 'first', createdAt: 10, priority: true }),
]),
).toEqual(['first', 'second']);
});
});
describe('bumpQueuedMessage', () => {
const queue = [
queued({ id: 'a', createdAt: 1 }),
queued({ id: 'b', createdAt: 2 }),
queued({ id: 'c', createdAt: 3 }),
];
it('moves the chosen row to the front and preserves the rest of the order', () => {
expect(bumpQueuedMessage(queue, 'c', 500).map(({ id }) => id)).toEqual(['c', 'a', 'b']);
});
it('survives a later enqueue, because the order lives in the sort key', () => {
const bumped = bumpQueuedMessage(queue, 'c', 500);
const afterEnqueue = [...bumped, queued({ id: 'd', createdAt: 4 })].sort(compareQueuedMessages);
expect(afterEnqueue.map(({ id }) => id)).toEqual(['c', 'a', 'b', 'd']);
});
it('returns the same array when the id is gone', () => {
expect(bumpQueuedMessage(queue, 'missing', 500)).toBe(queue);
});
});
describe('isMergeableQueuedMessage', () => {
it('rejects rows bound to a parked server source', () => {
expect(isMergeableQueuedMessage(queued({ id: 'a', recoverySteerId: 'steer-1' }))).toBe(false);
expect(isMergeableQueuedMessage(queued({ id: 'b', clientRequestId: 'req-1' }))).toBe(false);
});
it('accepts ordinary local rows', () => {
expect(isMergeableQueuedMessage(queued({ id: 'c' }))).toBe(true);
});
});
describe('isSendableQueuedMessage', () => {
it('is the whole rule: a row with words is sendable, one without is not', () => {
expect(isSendableQueuedMessage(queued({ id: 'a', text: 'words' }))).toBe(true);
expect(isSendableQueuedMessage(queued({ id: 'b', text: '' }))).toBe(false);
expect(isSendableQueuedMessage(queued({ id: 'c', text: ' \n ' }))).toBe(false);
});
});
describe('mergeQueuedMessages', () => {
it('refuses a batch containing a row being emptied, so a fold cannot bake in a blank', () => {
expect(
mergeQueuedMessages([queued({ id: 'a', text: 'kept' }), queued({ id: 'b', text: ' ' })]),
).toBeNull();
});
it('joins texts in drain order as paragraphs', () => {
const merged = mergeQueuedMessages([
queued({ id: 'a', text: 'first thought', createdAt: 1 }),
queued({ id: 'b', text: 'second thought', createdAt: 2 }),
]);
expect(merged?.text).toBe('first thought\n\nsecond thought');
});
it('keeps the front row`s identity and position', () => {
const merged = mergeQueuedMessages([
queued({ id: 'a', createdAt: 7, priority: true, bumpedAt: 90 }),
queued({ id: 'b', createdAt: 8 }),
]);
expect(merged).toMatchObject({ id: 'a', createdAt: 7, priority: true, bumpedAt: 90 });
});
it('unions attachments, quotes and skill picks without duplicates', () => {
const merged = mergeQueuedMessages([
queued({
id: 'a',
files: [{ file_id: 'f1', filepath: '/f1', type: 'image/png' }],
quotes: ['q1'],
manualSkills: ['s1'],
}),
queued({
id: 'b',
files: [
{ file_id: 'f1', filepath: '/f1', type: 'image/png' },
{ file_id: 'f2', filepath: '/f2', type: 'image/png' },
],
quotes: ['q1', 'q2'],
manualSkills: ['s2'],
}),
]);
expect(merged?.files?.map((file) => file.file_id)).toEqual(['f1', 'f2']);
expect(merged?.quotes).toEqual(['q1', 'q2']);
expect(merged?.manualSkills).toEqual(['s1', 's2']);
});
it('takes the latest predecessor fence so the merged turn is gated on everything it followed', () => {
const merged = mergeQueuedMessages([
queued({ id: 'a', expectedPredecessorCreatedAt: 100 }),
queued({ id: 'b', expectedPredecessorCreatedAt: 400 }),
]);
expect(merged?.expectedPredecessorCreatedAt).toBe(400);
});
it('refuses to merge a recovery-bound row, whose parked source must be discarded first', () => {
expect(
mergeQueuedMessages([
queued({ id: 'a' }),
queued({ id: 'b', recoverySteerId: 'steer-1', recoveryClientSteerId: 'local-1' }),
]),
).toBeNull();
});
it('refuses a batch of fewer than two rows', () => {
expect(mergeQueuedMessages([queued({ id: 'a' })])).toBeNull();
});
});

165
client/src/utils/steer.ts
View file

@ -5,7 +5,7 @@ import type {
TSteerAppliedEvent,
TMessageContentParts,
} from 'librechat-data-provider';
import type { QueuedMessage, QueuedMessageOrigin } from '~/store/families';
import type { QueuedMessage, QueuedMessageOrigin, RunEnd } from '~/store/families';
type SteerPart = Extract<TMessageContentParts, { type: ContentTypes.STEER }>;
@ -141,6 +141,21 @@ export function appendAppliedSteerIds(prev: string[], steerIds: string[]): strin
return [...prev, ...fresh].slice(-APPLIED_STEER_IDS_CAP);
}
/**
* Whether two terminal signals are the same generation. Conversation ids are
* reused by later turns, so identity is the exact epoch: the generation stamp
* when either side carries one, else the termination time.
*/
export function isSameRunEpoch(a: RunEnd | null | undefined, b: RunEnd): boolean {
if (a == null || a.conversationId !== b.conversationId) {
return false;
}
if (a.generationCreatedAt != null || b.generationCreatedAt != null) {
return a.generationCreatedAt === b.generationCreatedAt;
}
return a.endedAt === b.endedAt;
}
export type SteerCarriedContext = { quotes?: string[]; manualSkills?: string[] };
/** Quotes/skill picks are client-only (a steer never sends them to the
@ -155,6 +170,137 @@ export function carriedSteerContext(source?: SteerCarriedContext): SteerCarriedC
};
}
/**
* The queue's one ordering rule, shared by every writer so a reorder cannot be
* undone by the next enqueue re-sorting on a different key. Three tiers:
*
* 0. "Interrupt & send" front-inserts. They aborted a run to be said now, so
* they outrank everything — and stay FIFO among themselves, because two
* interrupts are sequential instructions, not competing ones.
* 1. "Send next" promotions, most recently promoted first.
* 2. Everything else, in enqueue order.
*/
function orderingTier(item: QueuedMessage): number {
if (item.priority === true) {
return 0;
}
return item.bumpedAt != null ? 1 : 2;
}
export function compareQueuedMessages(a: QueuedMessage, b: QueuedMessage): number {
const tier = orderingTier(a) - orderingTier(b);
if (tier !== 0) {
return tier;
}
if (a.bumpedAt != null && b.bumpedAt != null && a.bumpedAt !== b.bumpedAt) {
return b.bumpedAt - a.bumpedAt;
}
return a.createdAt - b.createdAt;
}
/**
* Whether a row can be sent as it stands. A row being edited holds exactly what
* the user has typed, blank included, so "there is nothing to send" is a fact
* about the row rather than a state every sender has to be told about
* separately. Blank rows exist only while their editor is open — closing it
* restores the pre-edit words — so this never describes a resting queue.
*/
export function isSendableQueuedMessage(item: QueuedMessage): boolean {
return item.text.trim().length > 0;
}
/** Queued texts are separate thoughts, so a join reads as paragraphs. */
export const QUEUED_TEXT_SEPARATOR = '\n\n';
/**
* A row bound to a parked server source cannot be merged or rewritten: the
* recovery turn must reproduce that source's exact text and file set, and its
* user-row identity is the receipt id, so two sources cannot become one turn.
* Discarding the parked copy first (`discardQueued`) downgrades the row and
* makes it mergeable like any local follow-up.
*/
export function isMergeableQueuedMessage(item: QueuedMessage): boolean {
return item.recoverySteerId == null && item.clientRequestId == null;
}
const dedupeFiles = (items: QueuedMessage[]): TMessage['files'] => {
const byId = new Map<string, NonNullable<TMessage['files']>[number]>();
for (const item of items) {
for (const file of item.files ?? []) {
const key = file.file_id ?? file.filepath ?? '';
if (key.length > 0 && !byId.has(key)) {
byId.set(key, file);
}
}
}
return byId.size > 0 ? [...byId.values()] : undefined;
};
const dedupeStrings = (values: Array<string[] | undefined>): string[] | undefined => {
const merged = new Set<string>();
for (const list of values) {
for (const value of list ?? []) {
merged.add(value);
}
}
return merged.size > 0 ? [...merged] : undefined;
};
/**
* Folds queued rows into the one turn they were probably always meant to be —
* each extra turn costs a full model round trip and context replay. Keeps the
* front-most row's identity and position so the merged message drains exactly
* where the first of its parts would have, and takes the LATEST predecessor
* fence of the batch so the merged turn is gated on everything it followed.
*/
export function mergeQueuedMessages(items: QueuedMessage[]): QueuedMessage | null {
if (items.length < 2 || items.some((item) => !isMergeableQueuedMessage(item))) {
return null;
}
/** Folding a row mid-edit would bake in whatever is on screen at that instant,
* blank included. Self-protecting here so no caller has to remember. */
if (items.some((item) => !isSendableQueuedMessage(item))) {
return null;
}
const [first] = items;
const files = dedupeFiles(items);
const quotes = dedupeStrings(items.map((item) => item.quotes));
const manualSkills = dedupeStrings(items.map((item) => item.manualSkills));
const fences = items
.map((item) => item.expectedPredecessorCreatedAt)
.filter((value): value is number => value != null);
return {
id: first.id,
text: items.map((item) => item.text).join(QUEUED_TEXT_SEPARATOR),
createdAt: first.createdAt,
...(first.priority === true && { priority: true }),
...(first.bumpedAt != null && { bumpedAt: first.bumpedAt }),
...(fences.length > 0 && { expectedPredecessorCreatedAt: Math.max(...fences) }),
...(files && { files }),
...(quotes && { quotes }),
...(manualSkills && { manualSkills }),
};
}
/**
* Promotes an item to drain next, leaving every other item's order intact.
* Deliberately does NOT set `priority`: that tier belongs to interrupts, which
* aborted a run and must stay ahead of a promotion made before them.
*/
export function bumpQueuedMessage(
queue: QueuedMessage[],
id: string,
bumpedAt: number,
): QueuedMessage[] {
const index = queue.findIndex((item) => item.id === id);
if (index < 0) {
return queue;
}
const next = [...queue];
next[index] = { ...next[index], bumpedAt };
return next.sort(compareQueuedMessages);
}
/** Restore a temporarily removed queue item using surviving original
* neighbours first, then the queue's durable priority/time ordering. */
export function insertQueuedOrigin(
@ -199,18 +345,19 @@ export function insertQueuedOrigin(
}
}
if (index < 0) {
index = queue.findIndex((queued) => {
const itemPriority = Number(restoredItem.priority === true);
const queuedPriority = Number(queued.priority === true);
return (
itemPriority > queuedPriority ||
(itemPriority === queuedPriority && restoredItem.createdAt < queued.createdAt)
);
});
index = queue.findIndex((queued) => compareQueuedMessages(restoredItem, queued) < 0);
if (index < 0) {
index = queue.length;
}
}
/** Captured neighbours describe where the row WAS, and the queue may have
* been reordered while it was away — a "Send this one next" promotion must
* not be undone by a restore landing in front of it. Walk the neighbour
* slot forward past anything that now outranks the row; with no promotion
* the neighbour slot already satisfies the comparator and this is a no-op. */
while (index < queue.length && compareQueuedMessages(queue[index], restoredItem) < 0) {
index += 1;
}
return [...queue.slice(0, index), restoredItem, ...queue.slice(index)];
}

6
client/tailwind.config.cjs
View file

@ -23,6 +23,10 @@ module.exports = {
authPageWidth: '370px',
},
keyframes: {
'queue-undo-grace': {
from: { transform: 'scaleX(1)' },
to: { transform: 'scaleX(0)' },
},
'accordion-down': {
from: { height: 0 },
to: { height: 'var(--radix-accordion-content-height)' },
@ -62,6 +66,8 @@ module.exports = {
'slide-out-left': 'slide-out-left 300ms cubic-bezier(0.25, 0.1, 0.25, 1)',
'slide-out-right': 'slide-out-right 300ms cubic-bezier(0.25, 0.1, 0.25, 1)',
'shortcut-shake': 'shortcut-shake 0.25s ease-in-out',
/* Duration is overridden inline with the window's remaining time. */
'queue-undo-grace': 'queue-undo-grace 3s linear forwards',
},
colors: {
gray: {

184
e2e/specs/mock/queued-outbox.spec.ts Normal file
View file

@ -0,0 +1,184 @@
import { expect, test } from '@playwright/test';
import type { Page } from '@playwright/test';
import {
MOCK_ENDPOINTS,
NEW_CHAT_PATH,
messagesView,
replyPrompt,
replyText,
selectMockEndpoint,
sendMessage,
} from './helpers';
const uniqueLabel = (prefix: string) =>
`${prefix}-${Date.now()}-${Math.floor(Math.random() * 1e4)}`;
const messageInput = (page: Page) => page.getByRole('textbox', { name: 'Message input' });
const duringRunSendButton = (page: Page) => page.getByTestId('during-run-send-button');
const queuedRows = (page: Page) => page.getByTestId('queued-message-row');
const messageTurns = (page: Page) => messagesView(page).locator('.message-render');
const outboxGroup = (page: Page) => page.getByTestId('queue-group');
const outboxToggle = (page: Page) => page.getByTestId('queue-group-toggle');
async function establishConversation(page: Page, label: string) {
const setup = await sendMessage(page, replyPrompt(label));
expect(setup.ok()).toBeTruthy();
await expect(messagesView(page).getByText(replyText(label))).toBeVisible({ timeout: 30000 });
await expect(page).toHaveURL(/\/c\/[0-9a-fA-F-]{36}$/, { timeout: 15000 });
}
/** Fill the composer mid-run: the during-run send button must take the
* send/stop slot (it becomes the form submit target for Enter). */
async function typeDuringRun(page: Page, text: string) {
const input = messageInput(page);
await input.click();
await input.fill(text);
await expect(duringRunSendButton(page)).toBeVisible({ timeout: 5000 });
}
/** Ctrl/Cmd+Enter routes to the non-default during-run action (queue). */
async function queueDuringRun(page: Page, text: string) {
await typeDuringRun(page, text);
await messageInput(page).press('ControlOrMeta+Enter');
}
/** Starts a slow run and parks `texts` in the queue behind it. */
async function queueBehindSlowRun(page: Page, label: string, texts: string[]) {
await page.goto(NEW_CHAT_PATH, { timeout: 10000 });
await selectMockEndpoint(page, MOCK_ENDPOINTS[0]);
await establishConversation(page, `outbox-setup-${label}`);
const run = await sendMessage(page, `E2E_SLOW_REPLY:${label}`);
expect(run.ok()).toBeTruthy();
await expect(messagesView(page).getByText('chunk-010')).toBeVisible({ timeout: 15000 });
for (const text of texts) {
await queueDuringRun(page, text);
}
}
test.describe('queued message outbox', () => {
/**
* Two or more waiting messages collapse into one row so the composer stops
* growing with queue depth; the managed list lives in the expansion.
*/
test('groups queued messages behind one row and expands to manage them', async ({ page }) => {
test.setTimeout(120000);
const label = uniqueLabel('outbox-group');
const first = `First queued ${label}`;
const second = `Second queued ${label}`;
await queueBehindSlowRun(page, label, [first, second]);
// Collapsed: one summary row, the individual rows are not mounted.
await expect(outboxGroup(page)).toBeVisible({ timeout: 10000 });
await expect(outboxToggle(page)).toHaveAttribute('aria-expanded', 'false');
await expect(outboxToggle(page)).toContainText('2 queued');
await expect(outboxToggle(page)).toContainText(first);
await expect(queuedRows(page)).toHaveCount(0);
await outboxToggle(page).click();
await expect(outboxToggle(page)).toHaveAttribute('aria-expanded', 'true');
await expect(queuedRows(page)).toHaveCount(2);
await expect(queuedRows(page).nth(0)).toContainText(first);
await expect(queuedRows(page).nth(1)).toContainText(second);
// Both still drain, in order, after the run completes.
await expect(queuedRows(page)).toHaveCount(0, { timeout: 90000 });
const turns = messageTurns(page);
await expect(turns.filter({ hasText: first })).toHaveCount(1, { timeout: 60000 });
await expect(turns.filter({ hasText: second })).toHaveCount(1, { timeout: 60000 });
});
/** "Send next" promotes a waiting row past the one in front of it. */
test('send next changes which queued message drains first', async ({ page }) => {
test.setTimeout(120000);
const label = uniqueLabel('outbox-bump');
const first = `Typed first ${label}`;
const second = `Wanted first ${label}`;
await queueBehindSlowRun(page, label, [first, second]);
await outboxToggle(page).click();
await expect(queuedRows(page)).toHaveCount(2);
// The front row is already next, so only the second offers the promotion.
const bump = page.getByTestId('queued-send-next');
await expect(bump).toHaveCount(1);
await bump.click();
await expect(queuedRows(page).nth(0)).toContainText(second);
await expect(queuedRows(page).nth(1)).toContainText(first);
// Drain order follows the rendered order: the promoted message sends first.
await expect(queuedRows(page)).toHaveCount(0, { timeout: 90000 });
const userTurns = messagesView(page).locator('.user-turn');
await expect(userTurns.filter({ hasText: first })).toHaveCount(1, { timeout: 60000 });
const sent = await userTurns.allInnerTexts();
const promotedIndex = sent.findIndex((text) => text.includes(second));
const typedIndex = sent.findIndex((text) => text.includes(first));
expect(promotedIndex).toBeGreaterThanOrEqual(0);
expect(promotedIndex).toBeLessThan(typedIndex);
});
/**
* Burst-typed fragments are usually one thought, and every extra turn costs a
* full model round trip — merging sends them as a single turn.
*/
test('merge folds the queue into a single turn', async ({ page }) => {
test.setTimeout(120000);
const label = uniqueLabel('outbox-merge');
const first = `Fragment one ${label}`;
const second = `Fragment two ${label}`;
await queueBehindSlowRun(page, label, [first, second]);
await outboxToggle(page).click();
await page.getByTestId('queue-merge').click();
// One row now holds both texts, so the group collapses back to a chip.
await expect(outboxGroup(page)).toHaveCount(0);
const merged = queuedRows(page);
await expect(merged).toHaveCount(1);
await expect(merged).toContainText(first);
await expect(queuedRows(page)).toHaveCount(0, { timeout: 90000 });
// ONE user turn carries both fragments, rather than two turns.
const mergedTurn = messagesView(page).locator('.user-turn').filter({ hasText: first });
await expect(mergedTurn).toHaveCount(1, { timeout: 60000 });
await expect(mergedTurn).toContainText(second);
});
/**
* The automatic send is withheld briefly at run end. Undo cancels it without
* touching the queue, so the words stay put for a manual send.
*/
test('undo takes back the automatic send and keeps the message queued', async ({ page }) => {
test.setTimeout(120000);
const label = uniqueLabel('outbox-undo');
const queued = `Held back ${label}`;
await queueBehindSlowRun(page, label, [queued]);
await expect(queuedRows(page)).toHaveCount(1, { timeout: 10000 });
// The banner only appears once the run ends and the send is pending. The
// window is deliberately short, so wait on the mutation rather than a
// polling assertion and click the moment it lands.
const undo = page.getByTestId('queue-undo-send');
await undo.waitFor({ state: 'visible', timeout: 90000 });
await undo.click({ timeout: 2000 });
await expect(page.getByTestId('queue-sending-banner')).toHaveCount(0);
// Still queued, and no follow-up turn was sent.
await expect(queuedRows(page)).toHaveCount(1);
await expect(queuedRows(page)).toContainText(queued);
await expect(messagesView(page).locator('.user-turn').filter({ hasText: queued })).toHaveCount(
0,
);
// The words are recoverable on demand: Send now starts the turn.
await page.getByText('Send now', { exact: true }).click();
await expect(messagesView(page).locator('.user-turn').filter({ hasText: queued })).toHaveCount(
1,
{ timeout: 60000 },
);
});
});