🫗 fix: Drain Quoted Excerpts Into Mid-Run Steering (#15175)

* 🧭 fix: Carry Quoted Excerpts Through Mid-Run Steering

"Add to chat" quote chips were dropped by every during-run steer path: the
steer POST had no quotes concept, so a composer-origin steer left the chip
staged (gluing onto the NEXT send) and a queued item steered into the live
run lost its quotes silently.

Quotes now ride the steer protocol end to end:
- POST + admission: `quotes` on the steer body, normalized like the chat
  route's (getReferencedQuotes caps), part of the idempotency fingerprint
  only when present so pre-existing receipts still replay.
- Injection: merged into the model-bound turn as Markdown blockquotes at
  both boundaries (text-only and media paths), mirroring prependQuotes.
- Persistence + replay: the STEER content part stores `quotes` separately
  from the typed text; stampSteerPartMedia re-merges them per turn (even
  with resendFiles off) via the SDK's transient media stamp, with the quote
  block folded into the token budget.
- UI: composer steers/interrupt-steers drain the chips (skill picks stay
  staged — they configure a NEW turn's run); SteerPart and the in-flight
  bubble render the same MessageQuotes reference blocks as user bubbles;
  queued/failed rows show a quote count; reconnect reseeds fall back to the
  server item's quotes when no local chip survives.
- buildMessages keeps its zero-await path to the parallel context kickoff
  via a synchronous stamp-target probe.

* 🧭 fix: Keep Quotes in the Client-Safe Steer Projection

toPendingSteer is the projection behind resume-state pendingSteers, abort
responses, and terminal leftover claims — dropping quotes there would lose
them on exactly the recovery paths the reconnect reseed's server fallback
relies on.

* 🧪 test: In-Flight Steer Bubble Renders Carried Quotes

* 🔁 fix: Re-Stage Quotes When a Pre-Quotes Replica Accepts the Steer

Codex flagged the rolling-deploy window: an old replica 202s a quoted steer
while dropping the excerpts, so the client cleared the chips for context the
model never received.

The 202 (fresh and receipt replay) now echoes quotesAccepted from the
DURABLE item; a missing echo on a quote-bearing composer-origin steer
re-stages the excerpts as composer chips — the pre-steer behavior, so they
ride the next send instead of vanishing — and strips them from the surviving
chip so a later terminal conversion cannot duplicate them. Queued-origin
steers keep quotes on the item, whose restore paths already return it
intact. The residual cross-version lost-ACK retry stays fail-closed as a
409 idempotency conflict (failed chip with retry controls).

* 🔁 fix: Close the Remaining Cross-Version Quote-Loss Windows

Codex round 2:
- Send now of a quoted queued item against a pre-quotes replica now
  re-stages the excerpts too (the row is consumed and the words inject
  bare, so the composer is their only remaining home); the strip clears the
  chip's captured origin copy so reclaims and terminal conversions cannot
  duplicate them.
- A quoted retry whose lost first ACK was accepted by a pre-quotes replica
  now REPLAYS that legacy receipt instead of 409ing: the stored fingerprint
  matching the quote-less hash of the same words proves the cross-version
  case, and the replayed 202's missing echo drives the re-stage. Different
  quotes against a quote-bearing receipt still conflict.
- TSteerAppliedEvent.part gains the quotes field (typed SSE consumers).

* 🧪 test: Drop the Stale Narrow SteerDrainOutput Alias

The spec's local intersection re-declared injectedMessages with
content: string, predating the SDK pin that declares the field natively
(content: string | MessageContentComplex[]). Under CI's clean install the
hook's BaseHookOutput is no longer assignable to that narrower alias; the
plain PostToolBatchHookOutput is the correct type for every drain/boundary
assertion. Verified against the published 3.6.16 dist and the local one.

* 🔁 fix: Honor the Generation Owner's Quote Capability End to End

Codex round 4:
- steerQuotesCapable rides job metadata (createJob + HITL resume rewrite),
  mirroring preemptCapable's owner-recorded pattern: an upgraded admission
  replica no longer stores quotes — or claims them accepted — for a
  generation whose older owning drain would silently drop them at
  injection. The missing echo drives the client re-stage, and a later
  capable handover cannot double-deliver restored context.
- Applied events reconcile dropped quotes: when a quote-less applied part
  settles a quote-bearing chip (the lost-202 ordering the ACK-echo path
  cannot see), resolveSteerChip and both reconnect settle paths re-stage
  the chip's excerpts before removing their only copy. mergeRestagedQuotes
  dedupe keeps every trigger idempotent for the same excerpts.

* 🔁 fix: Re-Read Quote Capability at the Last Moment and Cap Restaged Chips

Codex round 5:
- A HITL resume rewrites steerQuotesCapable without changing the
  generation's createdAt, so the enqueue fence cannot see a
  capable-to-legacy handover landing during admission's awaits. Re-read
  the owner's flag immediately before item construction (paid for only by
  quote-bearing requests); the residual between re-read and enqueue commit
  matches preemptCapable's documented race.
- mergeRestagedQuotes now respects the 10-quote contract with the staged
  chips winning: a restored tail that cannot ride the next send is dropped
  explicitly instead of rendering as a chip the submission would silently
  discard. MAX_QUOTE_COUNT moves to utils/steer as the single client
  source; QuoteButton imports it.

* 🔁 fix: Steer Quote Coverage for Preflights, Memory, and Single-Scan Stamping

Codex round 6:
- Stored-message policy inspection now extracts steer-part quotes as quote
  fragments (path /content/N/quotes/M), so conversation import and shared
  link preflights inspect the newly persisted field exactly like top-level
  message.quotes.
- The memory copy gets its own quote-merge stamp (text only, resendFiles
  false): formatAgentMessages ignores part.quotes, so without it a steer
  whose substance lives in its excerpt reached the chat model but never
  memory extraction.
- collectSteerStampTargets replaces the boolean probe: buildMessages
  collects once and hands the targets to stampSteerPartMedia, keeping the
  zero-await fast path without scanning the history twice.

* 🔁 fix: Redis Quote Plumbing, Conversion-Race Guard, and Quote-Bound Recovery Proof

Codex round 7:
- RedisJobStore.deserializeJob now restores steerQuotesCapable (the explicit
  mapper otherwise dropped it on every read, leaving quote steering inert in
  Redis deployments), with the round-trip spec extended.
- Both Lua parked-steer projections (terminal close + generation
  replacement) forward item.quotes, matching toPendingSteer — a lost final
  no longer strips excerpts from durable recovery in Redis mode.
- The no-echo restage reads the SURVIVING chip (reclaimRejectedChipQuotes):
  a terminal conversion that beat the delayed 202 already moved the quotes
  onto the queued follow-up, and re-staging them again double-delivered.
  Regression-tested with the conversion-before-ACK ordering.
- RecoveredSteerPayload binds normalized, order-significant quotes (builder,
  validator, TS matcher, and the Lua decode+matcher): a stale client
  presenting the same recoverySteerId with altered or missing quotes cannot
  consume the parked source. Quote-less sources keep matching quote-less
  recoveries.

* 🔁 fix: Execution-Bound Quote Capability with an Atomic Enqueue Predicate

Codex round 8:
- steerQuotesCapable becomes a transient assertion translated (at createJob
  and in ApprovalLifecycle.resolve) into steerQuotesExecutionId, valid only
  while it equals the LIVE providerExecutionId. A legacy replica winning a
  HITL resume rewrites the execution id without knowing the marker, so its
  stale assertion self-invalidates — a bare boolean could not be cleared by
  code that predates it.
- The fenced enqueue evaluates that equality atomically (all three Redis
  scripts decode-and-strip like the existing preemptCapable normalization;
  both InMemory sites mirror it) and returns the persisted item, so the
  quotesAccepted echo reflects exactly what was stored even when a handover
  lands between admission's read and the commit. The last-moment re-read is
  gone — the transaction is the authority.
- Tests: capable-resume re-binding, legacy-resume omit-not-clear
  invalidation, the admission-vs-handover race (capability read true, then
  execution rewritten before enqueue), and the Redis round-trip of the
  marker.

* 🔁 fix: Full Redis Parking Coverage and Loss-Moment Quote Restaging

Codex round 9:
- The two remaining Redis parking projections (terminal status CAS and
  stale-running cleanup) forward item.quotes — every field-picked steer
  projection now carries them (audited: 2 Lua 'projected' + 2 Lua
  'clientItem' + toPendingSteer).
- The ordinary no-echo ACK no longer re-stages: the steer has not injected
  yet, so the quotes stay carried on the pending chip. A quote-less applied
  event re-stages them at the actual loss; a terminal leftover conversion
  carries them onto the recovered row, whose normal send delivers quotes on
  any server — re-staging at the ACK let that leftover auto-send bare text
  while the excerpts glued onto an unrelated draft. Only the settled
  receipt replay (already injected, no future event) reclaims immediately.

* 🔁 fix: Legacy-Replayable Receipts with Separate Quote Identity

Codex round 10: an upgraded-first receipt stored a quote-inclusive
fingerprint no pre-quotes replica could recompute, so a lost-ACK retry
routed through one 409'd already-accepted words with duplicate-send
controls.

The durable fingerprint reverts to the quote-independent 3-field hash —
the one shape EVERY deployed version computes, replayable across a rolling
deploy in both directions — and quote identity moves beside it as
requestedQuotesFingerprint (of the REQUESTED quotes, pre any capability
strip, so an incapable-owner acceptance still replays its own retries).
Absent records (legacy-written or quote-less) accept any same-words retry,
preserving the round-5 rule; present records must match exactly, keeping
different-quotes clientSteerId reuse a 409 on quote-aware readers. Under
the keep-on-chip client contract a legacy replay's missing echo is
harmless — the excerpts stay carried on the pending chip.

* 🧪 chore: Re-Trigger CI After Dropped Workflow Events
This commit is contained in:
Danny Avila 2026-08-24 22:29:13 -04:00 committed by GitHub
parent c0a55aa0f5
commit ac2aef00f6
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
50 changed files with 1686 additions and 160 deletions

View file

@ -738,6 +738,7 @@ describe('ResumableAgentController resume metadata', () => {
model: 'gpt-3.5-turbo',
/** The OWNING replica's seal capability, read by the steer route. */
preemptCapable: true,
steerQuotesCapable: true,
agent_id: undefined,
isTemporary: true,
responseMessageId: expect.stringMatching(/^[0-9a-f-]{36}$/),

View file

@ -58,6 +58,7 @@ const {
isSteeringSupported,
isSteerPreemptSupported,
buildSteerMedia,
collectSteerStampTargets,
stampSteerPartMedia,
createActivityLabelWiring,
createActivityPhaseWiring,
@ -387,6 +388,9 @@ class AgentClient extends BaseClient {
...(item.clientSteerId && { clientSteerId: item.clientSteerId }),
createdAt: item.createdAt,
...(item.files?.length && { files: item.files }),
// Persisted separately from the text (mirroring `message.quotes`) so the
// UI renders reference blocks and replay re-merges them per turn.
...(item.quotes?.length && { quotes: item.quotes }),
};
this.contentParts.push(part);
this.steerOffsetState.offset += 1;
@ -1892,22 +1896,33 @@ class AgentClient extends BaseClient {
payload = formattedMessages;
this.modelBoundSteerFileIdsBySourceMessageId = new Map();
if (this.options.resendFiles) {
/** Persisted steer parts of past turns replay with their attachments:
* one batched owner-scoped fetch, re-encoded per turn and stamped as a
* transient `media` array (same resend semantics as message files).
* The stamp lands after the loop above finalized its counts, so the
* re-encoded media (minus the text part the steer part already counted)
* is folded into the budget here large steered attachments must
* shrink the window like any other resent media. */
/** Persisted steer parts of past turns replay with their attachments and
* quotes: one batched owner-scoped fetch, re-encoded per turn and
* stamped as a transient `media` array (same resend semantics as
* message files). Runs regardless of `resendFiles` because quote-bearing
* parts must re-merge their excerpts every turn (mirroring
* `prependQuotes` above); file encoding stays gated on the setting via
* the flag. The stamp lands after the loop above finalized its counts,
* so the re-encoded media (minus the text part the steer part already
* counted) is folded into the budget here large steered attachments
* and quote blocks must shrink the window like any other resent media.
* The synchronous collection keeps steer-free histories on the
* zero-await path to the parallel context kickoff below, and the
* collected targets feed the stamp directly so the history is scanned
* once. */
const resendSteerFiles = this.options.resendFiles === true;
const steerStampTargets = collectSteerStampTargets(payload, resendSteerFiles);
if (steerStampTargets.length > 0) {
const stamped = await stampSteerPartMedia({
client: this,
user: this.options.req?.user,
payload,
targets: steerStampTargets,
// addPreviousAttachments already fetched steer-part refs in its single
// per-turn historical-files query — no second round trip.
docsById: this.authorizedHistoricalFiles,
getFiles: db.getFiles,
resendFiles: resendSteerFiles,
});
for (const { sourceMessageId, fileIds } of stamped) {
if (typeof sourceMessageId !== 'string' || sourceMessageId.length === 0) {
@ -1927,8 +1942,8 @@ class AgentClient extends BaseClient {
for (const { index, media, steerText } of stamped) {
/** Count the FULL stamped content and subtract only the steer body
* (already counted inside the assistant message): extracted file
* context prepended into the text part must hit the budget too, or
* large steered documents bypass pruning. */
* context and merged quote blocks prepended into the text part must
* hit the budget too, or large steered documents bypass pruning. */
const fullTokens = countFormattedMessageTokens({ role: 'user', content: media }, encoding);
const bodyTokens = steerText
? countFormattedMessageTokens(
@ -1949,6 +1964,25 @@ class AgentClient extends BaseClient {
memoryFormattedMessages[i] ?? buildMemoryFormattedMessage(orderedMessages[i]),
);
}
/** The memory copy feeds `processMemory` through the same
* `formatAgentMessages` replay, which reads `part.media`/`part.steer`
* and ignores `part.quotes` so a steer whose substance lives in its
* quote must be quote-merged here too or memory extraction never sees
* it. Quote merge only (`resendFiles: false`): file media is exactly
* what the memory copy exists to exclude, and text-only stamps touch
* no file fetch or encode. Runs after the fill above so late-built
* copies are stamped too. */
const memorySteerTargets = collectSteerStampTargets(memoryPayload, false);
if (memorySteerTargets.length > 0) {
await stampSteerPartMedia({
client: this,
user: this.options.req?.user,
payload: memoryPayload,
targets: memorySteerTargets,
getFiles: db.getFiles,
resendFiles: false,
});
}
}
this.memoryPayload = hasFileContext ? memoryPayload : null;
messages = orderedMessages;

View file

@ -3749,6 +3749,60 @@ describe('AgentClient - titleConvo', () => {
);
});
it('quote-merges historical steer parts into the prompt AND the memory copy', async () => {
const previousFileContext =
'Attached document(s):\n```md\n# "previous.txt"\nPrevious turn file body\n```';
const result = await client.buildMessages(
[
{
messageId: 'msg-1',
parentMessageId: null,
sender: 'User',
text: 'Summarize.',
isCreatedByUser: true,
fileContext: previousFileContext,
},
{
messageId: 'msg-2',
parentMessageId: 'msg-1',
sender: 'Assistant',
text: '',
isCreatedByUser: false,
content: [
{ type: ContentTypes.TEXT, text: 'working on it' },
{
type: ContentTypes.STEER,
[ContentTypes.STEER]: 'remember this',
steerId: 's1',
quotes: ['the important fact'],
},
],
},
{
messageId: 'msg-3',
parentMessageId: 'msg-2',
sender: 'User',
text: 'Continue.',
isCreatedByUser: true,
},
],
'msg-3',
{},
);
const merged = '> the important fact\n\nremember this';
const promptSteer = result.prompt[1].content.find((part) => part.type === ContentTypes.STEER);
expect(promptSteer.media).toEqual([{ type: ContentTypes.TEXT, text: merged }]);
// The memory copy replays through the same formatter, which ignores
// `part.quotes` — it needs its own merged stamp or memory extraction
// never sees the excerpt.
const memorySteer = client.memoryPayload[1].content.find(
(part) => part.type === ContentTypes.STEER,
);
expect(memorySteer.media).toEqual([{ type: ContentTypes.TEXT, text: merged }]);
});
it('persists canonical token counts while counting request file context for the prompt', async () => {
const { countFormattedMessageTokens } = require('@librechat/api');
const currentFile = makeTextFile('current-file', 'current.txt', 'Current turn file body');

View file

@ -444,7 +444,7 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
const isRecoveredSteerRequest = recoveredSteerId != null;
const recoveryUserMessageId = rawOverrideUserMessageId;
const recoveredSteerPayload = isRecoveredSteerRequest
? buildRecoveredSteerPayload(text, req.body?.files)
? buildRecoveredSteerPayload(text, req.body?.files, req.body?.quotes)
: undefined;
/** A recovered steer is handed off as a new ordinary user turn. Edit,
* regenerate, continue, and arbitrary override-id shapes can reuse an
@ -1106,6 +1106,10 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
// route may land on a different replica whose own SDK probe would
// answer for the wrong process during a rolling deploy.
preemptCapable: isSteerPreemptSupported(),
// Same owner-recorded pattern: this build's drain merges queued steer
// quotes into the injected turn. Admission on another replica must
// not store/acknowledge quotes an older owner would drop.
steerQuotesCapable: true,
// Persist the originating agent so a HITL resume can refuse to rebuild this
// paused run on a different agent (see resume.js).
agent_id: endpointOption.agent_id ?? req.body?.agent_id,

View file

@ -1222,6 +1222,9 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
pendingAction.actionId,
{
preemptCapable: isSteerPreemptSupported(),
// The handover owner's quote handling replaces the previous
// replica's flag, mirroring `preemptCapable` above.
steerQuotesCapable: true,
providerExecutionId,
providerDrained: true,
...(resolvedAskUserQuestion && { resolvedAskUserQuestions }),

View file

@ -16,6 +16,7 @@ import {
import FilePreviewDialog from '~/components/Chat/Messages/Content/FilePreviewDialog';
import { supportsGenerationProtocolV2, useArmSteerMutation } from '~/data-provider';
import { steerOverlayHeightFamily, escalatingSteerFamily } from '~/store/steer';
import MessageQuotes from '~/components/Chat/Messages/Content/MessageQuotes';
import MarkdownLite from '~/components/Chat/Messages/Content/MarkdownLite';
import FileContainer from '~/components/Chat/Input/Files/FileContainer';
import { useSteerCancel, useSteerReclaim, useLocalize } from '~/hooks';
@ -482,6 +483,9 @@ const InFlightSteer = memo(function InFlightSteer({
{localize(preempting ? 'com_ui_steer_in_flight_preempt' : 'com_ui_steer_in_flight')}
</span>
<div className="flex min-w-0 flex-col items-start gap-1">
{/* Same reference blocks the applied `SteerPart` shows, outside the
* collapse so the excerpts stay visible while a long steer clips. */}
<MessageQuotes quotes={steer.quotes} />
<div
ref={contentRef}
id={contentId}

View file

@ -26,9 +26,11 @@ const CLOSE_DELAY_MS = 120;
* `autoFocusOnShow` is disabled so opening never pulls focus off the composer;
* `autoFocusOnHide` still returns focus to the trigger when focus was inside.
*
* Reads + writes `pendingQuotesByConvoId` directly; the atom is drained in
* `useChatFunctions.ask` on submit, so chips disappear once the message is sent
* (the excerpts then re-render as `MessageQuotes` on the user bubble).
* Reads + writes `pendingQuotesByConvoId` directly; the atom is drained on
* every submit route `useChatFunctions.ask` for a fresh send, and the
* during-run steer/queue/interrupt paths in `useSteering` so chips disappear
* once the message is sent (the excerpts then re-render as `MessageQuotes` on
* the user bubble, or inside the steer bubble for a mid-run injection).
*/
function PendingQuoteChips({ conversationId }: { conversationId: string }) {
const localize = useLocalize();

View file

@ -2,7 +2,7 @@ import { memo, useMemo, useRef, useState, useCallback } 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, Send, Clock, Pencil, Trash2, Paperclip, RotateCcw, TextQuote } 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';
@ -24,19 +24,47 @@ 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 }) {
function ContextCount({
icon,
count,
label,
}: {
icon: React.ReactNode;
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" />
{icon}
{count}
<span className="sr-only">{label}</span>
</span>
);
}
function AttachmentCount({ count, label }: { count: number; label: string }) {
return (
<ContextCount
icon={<Paperclip className="h-3.5 w-3.5" aria-hidden="true" />}
count={count}
label={label}
/>
);
}
function QuoteCount({ count, label }: { count: number; label: string }) {
return (
<ContextCount
icon={<TextQuote className="h-3.5 w-3.5" aria-hidden="true" />}
count={count}
label={label}
/>
);
}
function QueuedRow({
message,
steering,
@ -61,6 +89,7 @@ function QueuedRow({
const toggleEntry = useDefaultToggleEntry(steering);
const interruptToggle = useInterruptToggleEntry();
const fileCount = message.files?.length ?? 0;
const quoteCount = message.quotes?.length ?? 0;
const isRecovered = message.recoverySteerId != null;
const actionPendingRef = useRef(false);
const [actionPending, setActionPending] = useState(false);
@ -147,6 +176,10 @@ function QueuedRow({
<span className="min-w-0 flex-1 truncate" title={message.text}>
{message.text}
</span>
<QuoteCount
count={quoteCount}
label={localize('com_ui_queued_quote_count', { 0: String(quoteCount) })}
/>
<AttachmentCount
count={fileCount}
label={localize('com_ui_queued_attachment_count', { 0: String(fileCount) })}
@ -275,6 +308,10 @@ function FailedSteerRow({
<span className="min-w-0 flex-1 truncate" title={steer.text}>
{steer.text}
</span>
<QuoteCount
count={steer.quotes?.length ?? 0}
label={localize('com_ui_queued_quote_count', { 0: String(steer.quotes?.length ?? 0) })}
/>
<span className="shrink-0 text-xs text-red-500">
{localize(
steer.deliveryUncertain ? 'com_ui_steer_delivery_unconfirmed' : 'com_ui_steer_failed',

View file

@ -2,18 +2,15 @@ import { memo, useRef, useState, useEffect, useCallback, useLayoutEffect } from
import { createPortal } from 'react-dom';
import { TextQuote } from 'lucide-react';
import { useSetRecoilState } from 'recoil';
import { cn, MAX_QUOTE_COUNT } from '~/utils';
import { mainTextareaId } from '~/common';
import { useLocalize } from '~/hooks';
import { cn } from '~/utils';
import store from '~/store';
/** Only selections fully inside a rendered chat message get the popup. */
const MESSAGE_SELECTOR = '.message-render';
/** Max characters captured per excerpt (backend re-caps as defense-in-depth). */
const MAX_QUOTE_LENGTH = 1500;
/** Max excerpts queued at once; mirrors the backend `QUOTE_MAX_COUNT` cap so
* the composer never shows more quotes than the model actually receives. */
const MAX_QUOTE_COUNT = 10;
/** Vertical gap (px) between the selection and the popup. */
const POPUP_OFFSET = 8;
/** Keep the popup this far (px) from the viewport edges. */

View file

@ -185,6 +185,19 @@ describe('InFlightSteers', () => {
expect(screen.queryByTestId('in-flight-steers')).toBeNull();
});
it('renders carried quotes as the same reference blocks the applied part shows', () => {
renderSteers([
{
steerId: 's-quoted',
text: 'about the selection',
status: 'pending',
createdAt: 1,
quotes: ['the selected excerpt'],
},
]);
expect(screen.getByTestId('message-quotes')).toHaveTextContent('the selected excerpt');
});
it('shows the menu at rest on every pointer, without hover-gating', () => {
renderSteers([
{ steerId: 's-ack', text: 'waiting on boundary', status: 'pending', createdAt: 1 },

View file

@ -80,6 +80,7 @@ const Part = memo(function Part({
<SteerPart
steer={part[ContentTypes.STEER]}
files={part.files}
quotes={part.quotes}
steerId={part.steerId}
createdAt={part.createdAt}
/>

View file

@ -4,6 +4,7 @@ import { InfoHoverCard, ESide } from '@librechat/client';
import type { TFile, TMessage } from 'librechat-data-provider';
import FilePreviewDialog from '~/components/Chat/Messages/Content/FilePreviewDialog';
import MessageTimestamp from '~/components/Chat/Messages/ui/MessageTimestamp';
import MessageQuotes from '~/components/Chat/Messages/Content/MessageQuotes';
import MarkdownLite from '~/components/Chat/Messages/Content/MarkdownLite';
import FileContainer from '~/components/Chat/Input/Files/FileContainer';
import Image from '~/components/Chat/Messages/Content/Image';
@ -25,11 +26,15 @@ import store from '~/store';
const SteerPart = memo(function SteerPart({
steer,
files,
quotes,
steerId,
createdAt,
}: {
steer: string;
files?: TMessage['files'];
/** Quoted excerpts steered with the message; rendered as the same reference
* blocks a user bubble shows for `message.quotes`. */
quotes?: string[];
/** Anchors the part for the message-nav rail (`#steer-<id>` rib target). */
steerId?: string;
createdAt?: number;
@ -83,6 +88,7 @@ const SteerPart = memo(function SteerPart({
<div className="user-turn relative flex w-fit max-w-[90%] flex-col items-end sm:max-w-[85%]">
<h2 className="sr-only">{label}</h2>
<div className="flex max-w-full flex-col items-start gap-2 rounded-theme-surface rounded-br-theme-control bg-surface-tertiary px-theme-normal py-2.5">
<MessageQuotes quotes={quotes} />
{(imageFiles.length > 0 || otherFiles.length > 0) && (
<div className="flex flex-wrap gap-2">
{otherFiles.map((file) => (

View file

@ -165,4 +165,25 @@ describe('SteerPart presentation', () => {
fireEvent.click(screen.getByTestId('steer-file'));
expect(screen.getByTestId('steer-file-preview')).toHaveTextContent('notes.pdf');
});
it('renders quoted excerpts as reference blocks inside the bubble', () => {
render(
<RecoilRoot initializeState={({ set }) => set(store.user, SEEDED_USER as never)}>
<SteerPart
steer="steered words"
steerId="s1"
createdAt={1}
quotes={['the selected excerpt']}
/>
</RecoilRoot>,
);
const quotes = screen.getByTestId('message-quotes');
expect(quotes).toHaveTextContent('the selected excerpt');
expect(quotes.closest('.bg-surface-tertiary')).not.toBeNull();
});
it('renders no quote block when the steer carried none', () => {
renderPart();
expect(screen.queryByTestId('message-quotes')).toBeNull();
});
});

View file

@ -171,6 +171,10 @@ export interface SteerMessageParams {
text: string;
/** Attachment refs steered with the message (already uploaded). */
files?: TMessage['files'];
/** Quoted excerpts steered with the message ("Add to chat" selections). The
* server normalizes them like a normal send's quotes and merges them into
* the model-bound turn at the injection boundary. */
quotes?: string[];
/**
* Ask the server to seal the live model stream at the next provider-safe
* boundary rather than waiting for a tool step. Never a rejection reason:
@ -190,6 +194,10 @@ export interface SteerMessageResponse {
/** Whether the seal request was actually armed; see {@link SteerMessageParams.preempt}. */
preempt?: boolean;
preemptRevision?: number;
/** Echoed when the durable item carries the sent quotes. Absent on a
* pre-quotes server (which 202s while dropping them) the client then
* re-stages the excerpts as composer chips instead of losing them. */
quotesAccepted?: boolean;
/** Receipt replay after this item already left the durable queue. */
settled?: boolean;
/** Settled specifically by terminal drain; restore as a queued follow-up. */

View file

@ -2052,6 +2052,7 @@ describe('useSteering', () => {
chips: useRecoilValue(store.pendingSteersByConvoId(CONVO_ID)),
pendingQuotes: useRecoilValue(store.pendingQuotesByConvoId(CONVO_ID)),
pendingSkills: useRecoilValue(store.pendingManualSkillsByConvoId(CONVO_ID)),
markApplied: useSetRecoilState(store.appliedSteerIdsByConvoId(CONVO_ID)),
}),
{ wrapper },
);
@ -2100,16 +2101,55 @@ describe('useSteering', () => {
expect(result.current.pendingSkills).toEqual([]);
});
it('leaves staged context untouched on the steer path (steers do not carry it)', () => {
it('steerFromComposer drains the quote chips into the POST, leaving skill picks staged', () => {
const { result } = setupWithContext({}, stageContext);
act(() => {
result.current.steering.steerFromComposer('steer text');
});
expect(mockMutate).toHaveBeenCalledTimes(1);
expect(result.current.pendingQuotes).toEqual(['quoted excerpt']);
expect(mockMutate).toHaveBeenCalledWith(
expect.objectContaining({ text: 'steer text', quotes: ['quoted excerpt'] }),
expect.anything(),
);
expect(mockMutate.mock.calls[0][0]).not.toHaveProperty('manualSkills');
// Consumed like a normal send's quotes; the excerpts now ride the steer.
expect(result.current.pendingQuotes).toEqual([]);
expect(result.current.chips[0]).toMatchObject({ quotes: ['quoted excerpt'] });
// A skill pick configures a NEW turn's run — it keeps waiting for one.
expect(result.current.pendingSkills).toEqual(['skill-1']);
});
it('interruptSteer carries the staged quotes the same way', () => {
const { result } = setupWithContext({}, stageContext);
act(() => {
result.current.steering.interruptSteer('stop and use this');
});
expect(mockMutate).toHaveBeenCalledWith(
expect.objectContaining({ quotes: ['quoted excerpt'], preempt: true }),
expect.anything(),
);
expect(result.current.pendingQuotes).toEqual([]);
expect(result.current.pendingSkills).toEqual(['skill-1']);
});
it("sendQueuedNow posts a queued item's quotes when steering it into the live run", () => {
const item: QueuedMessage = {
id: 'q-live',
text: 'queued with quotes',
createdAt: 1_000,
quotes: ['queued excerpt'],
};
const { result } = setupWithContext({}, ({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [item]);
});
act(() => {
result.current.steering.sendQueuedNow(item);
});
expect(mockMutate).toHaveBeenCalledWith(
expect.objectContaining({ text: 'queued with quotes', quotes: ['queued excerpt'] }),
expect.anything(),
);
});
it('queues without quotes/skills fields when nothing is staged', () => {
const { result } = setupWithContext();
act(() => {
@ -2207,7 +2247,13 @@ describe('useSteering', () => {
it('carries a queued-origin context onto the sending chip and the 202 ACK chip', () => {
mockMutate.mockImplementationOnce((_params, { onSuccess }) => {
onSuccess({ steerId: 'srv-ctx', status: 'queued', position: 1, conversationId: CONVO_ID });
onSuccess({
steerId: 'srv-ctx',
status: 'queued',
position: 1,
conversationId: CONVO_ID,
quotesAccepted: true,
});
});
const { result } = setupWithContext();
act(() => {
@ -2226,6 +2272,145 @@ describe('useSteering', () => {
]);
});
it('keeps rejected quotes carried on the pending chip (old-server ACK)', () => {
// A pre-quotes replica 202s the words without their excerpts, but the
// steer has NOT injected yet — the quotes must stay attached to the
// words: a later quote-less applied event re-stages them at the actual
// loss, while a terminal leftover conversion carries them onto the
// recovered row (whose normal send delivers quotes on any server).
// Re-staging at the ACK would let that leftover auto-send bare text
// while the excerpts glue onto an unrelated composer draft.
mockMutate.mockImplementationOnce((_params, { onSuccess }) => {
onSuccess({ steerId: 'srv-old', status: 'queued', position: 1, conversationId: CONVO_ID });
});
const { result } = setupWithContext({}, stageContext);
act(() => {
result.current.steering.steerFromComposer('quoted for an old server');
});
expect(result.current.pendingQuotes).toEqual([]);
expect(result.current.chips).toEqual([
expect.objectContaining({
steerId: 'srv-old',
status: 'pending',
quotes: ['quoted excerpt'],
}),
]);
});
it('keeps quotes drained when the ACK confirms they were accepted', () => {
mockMutate.mockImplementationOnce((_params, { onSuccess }) => {
onSuccess({
steerId: 'srv-new',
status: 'queued',
position: 1,
conversationId: CONVO_ID,
quotesAccepted: true,
});
});
const { result } = setupWithContext({}, stageContext);
act(() => {
result.current.steering.steerFromComposer('quoted for a new server');
});
expect(result.current.pendingQuotes).toEqual([]);
expect(result.current.chips[0]).toMatchObject({
steerId: 'srv-new',
quotes: ['quoted excerpt'],
});
});
it("keeps a queued item's quotes on its chip when Send now hits an old server", () => {
// The pending chip and its captured origin retain the quotes so every
// later outcome preserves them with the words: a quote-less applied
// event re-stages, a leftover conversion restores the exact row, and a
// reclaim hands them back to the composer.
mockMutate.mockImplementationOnce((_params, { onSuccess }) => {
onSuccess({
steerId: 'srv-q-old',
status: 'queued',
position: 1,
conversationId: CONVO_ID,
});
});
const item: QueuedMessage = {
id: 'q-old-server',
text: 'queued quoted words',
createdAt: 1_000,
quotes: ['queued excerpt'],
};
const { result } = setupWithContext({}, ({ set }) => {
set(store.queuedMessagesByConvoId(CONVO_ID), [item]);
});
act(() => {
result.current.steering.sendQueuedNow(item);
});
expect(result.current.queue).toEqual([]);
expect(result.current.pendingQuotes).toEqual([]);
const chip = result.current.chips[0];
expect(chip).toMatchObject({
steerId: 'srv-q-old',
status: 'pending',
quotes: ['queued excerpt'],
});
expect(chip.queuedOrigin?.item.quotes).toEqual(['queued excerpt']);
});
it('never re-stages quotes a terminal conversion already moved to the queue', () => {
// A pre-quotes server's run-end leftover event converts the chip (with
// its quotes) into a queued follow-up and marks the ids applied BEFORE
// the delayed no-echo 202 lands. The reclaim must find no surviving chip
// and leave the queued copy as the single owner of the excerpts.
let deferredOnSuccess: ((response: unknown) => void) | undefined;
mockMutate.mockImplementationOnce((_params, { onSuccess }) => {
deferredOnSuccess = onSuccess;
});
const { result } = setupWithContext({}, stageContext);
act(() => {
result.current.steering.steerFromComposer('converted before ACK');
});
const localId = result.current.chips[0].steerId;
act(() => {
result.current.markApplied((prev) => [...prev, localId]);
result.current.steering.convertSteerToQueue(localId, 'converted before ACK', undefined, {
quotes: ['quoted excerpt'],
});
});
expect(result.current.queue[0]).toMatchObject({ quotes: ['quoted excerpt'] });
act(() => {
deferredOnSuccess?.({
steerId: 'srv-converted',
status: 'queued',
position: 1,
conversationId: CONVO_ID,
});
});
// No re-mint, no re-stage: the queued follow-up remains the only copy.
expect(result.current.chips).toEqual([]);
expect(result.current.pendingQuotes).toEqual([]);
expect(result.current.queue).toHaveLength(1);
});
it('re-stages quotes on a settled receipt replay that never carried them', () => {
// Lost first ACK against an old replica; the retry's receipt replay
// (settled, already injected) proves the excerpts never attached.
mockMutate.mockImplementationOnce((_params, { onSuccess }) => {
onSuccess({
steerId: 'srv-replayed',
status: 'queued',
position: 1,
conversationId: CONVO_ID,
settled: true,
replayed: true,
generationProtocolVersion: 2,
});
});
const { result } = setupWithContext({}, stageContext);
act(() => {
result.current.steering.steerFromComposer('replayed without quotes');
});
expect(result.current.pendingQuotes).toEqual(['quoted excerpt']);
expect(result.current.chips).toEqual([]);
});
it('restores the carried context when a late ACK converts straight to queued', () => {
// The run ended before the 202 landed: the ACK's queued conversion is
// the only surviving copy of the steer, so it must keep quotes + skills.
@ -2325,6 +2510,7 @@ describe('useSteering', () => {
status: 'queued',
position: 1,
conversationId: CONVO_ID,
quotesAccepted: true,
});
});
act(() => {
@ -2346,7 +2532,7 @@ describe('useSteering', () => {
]);
});
it('leaves composer atoms staged when a composer-origin steer degrades', () => {
it('requeues a degraded composer-origin steer with its drained quotes', () => {
mockMutate.mockImplementationOnce((_params, { onError }) => {
onError({ response: { data: { code: 'RUN_PAUSED' } } });
});
@ -2354,12 +2540,14 @@ describe('useSteering', () => {
act(() => {
result.current.steering.steerFromComposer('degraded steer');
});
// Degrades to a text-only queued item; the staged chips stay put for
// the user's next composer send.
expect(result.current.queue).toEqual([expect.objectContaining({ text: 'degraded steer' })]);
expect(result.current.queue[0].quotes).toBeUndefined();
// The quotes were consumed into the steer, so its queued fallback must
// carry them — dropping them here would lose the user's references.
expect(result.current.queue).toEqual([
expect.objectContaining({ text: 'degraded steer', quotes: ['quoted excerpt'] }),
]);
expect(result.current.queue[0].manualSkills).toBeUndefined();
expect(result.current.pendingQuotes).toEqual(['quoted excerpt']);
expect(result.current.pendingQuotes).toEqual([]);
// Skill picks were never consumed and stay staged for the next send.
expect(result.current.pendingSkills).toEqual(['skill-1']);
});
});

View file

@ -64,8 +64,10 @@ export default function useSteerConvert() {
.getLoadable(store.activeGenerationProtocolVersionByConvoId(conversationId))
.getValue();
const bindRecoverySource = negotiatedVersion === 2;
// Quotes/skill picks never ride the server steer; restore them from
// the local chip (matched by id) before the chips are dropped below.
// Restore quotes/skill picks from the local chip (matched by id)
// before the chips are dropped below: the chip is the only carrier of
// skill picks, and of quotes accepted by an older server whose queue
// items did not persist them yet.
const localChips = snapshot
.getLoadable(store.pendingSteersByConvoId(conversationId))
.getValue();

View file

@ -20,6 +20,7 @@ import {
clearAllDrafts,
getPendingDraftId,
insertQueuedOrigin,
mergeRestagedQuotes,
} from '~/utils';
import useSteerConvert from '~/hooks/Chat/useSteerConvert';
import { useSetFilesToDelete } from '~/hooks/Files';
@ -622,6 +623,69 @@ export default function useSteering({
[conversationId],
);
/** Quotes-only drain for composer-origin steers: the excerpts ride the steer
* POST into the live run, while manual skill picks stay staged a skill
* pick configures a NEW turn's agent run and cannot apply to a mid-run
* injection, so it keeps waiting for the next full submission. */
const takeComposerQuotes = useRecoilCallback(
({ snapshot, reset }) =>
(): QueuedMessageContext => {
const quotes = snapshot
.getLoadable(store.pendingQuotesByConvoId(conversationId))
.getValue();
if (quotes.length === 0) {
return {};
}
reset(store.pendingQuotesByConvoId(conversationId));
return { quotes };
},
[conversationId],
);
/** Returns rejected excerpts to the composer chips from the SURVIVING steer
* chip used ONLY for a settled receipt replay, where the steer already
* injected in the source generation and no future applied event or
* terminal conversion will ever re-home the chip's quotes. An ordinary
* no-echo ACK deliberately does NOT reclaim: its steer is still queued, so
* the quotes stay carried on the pending chip the applied-event
* reconciliation re-stages them at the actual moment of loss, while a
* terminal leftover conversion moves them onto the recovered row, which
* sends via `ask` where quotes work on any server. When no chip remains,
* another path already owns the excerpts and re-staging would
* double-deliver. The chip is stripped in the same update (including its
* captured queue-origin copy) so the restaged composer chips stay their
* single representation. */
const reclaimRejectedChipQuotes = useRecoilCallback(
({ snapshot, set }) =>
(convoId: string, steerIds: string[]) => {
const chips = snapshot.getLoadable(store.pendingSteersByConvoId(convoId)).getValue();
const chip = chips.find((steer) => steerIds.includes(steer.steerId));
const quotes = chip?.quotes ?? chip?.queuedOrigin?.item.quotes;
if (quotes == null || quotes.length === 0) {
return;
}
set(store.pendingQuotesByConvoId(convoId), (prev) => mergeRestagedQuotes(prev, quotes));
set(store.pendingSteersByConvoId(convoId), (prev) => {
let changed = false;
const next = prev.map((steer) => {
const originHasQuotes = steer.queuedOrigin?.item.quotes != null;
if (!steerIds.includes(steer.steerId) || (steer.quotes == null && !originHasQuotes)) {
return steer;
}
changed = true;
const { quotes: _quotes, ...rest } = steer;
if (!originHasQuotes || rest.queuedOrigin == null) {
return rest;
}
const { quotes: _originQuotes, ...originItem } = rest.queuedOrigin.item;
return { ...rest, queuedOrigin: { ...rest.queuedOrigin, item: originItem } };
});
return changed ? next : prev;
});
},
[],
);
/** Consumes the composer's autosaved draft once its text has been taken into
* a steer or queued item. The composer clears via the form's `reset()`,
* which is programmatic and never fires the `input` event `useAutoSave`
@ -853,11 +917,12 @@ export default function useSteering({
[index, queueKey, activeGenerationCreatedAt],
);
/** POSTs a steer (text + files only; the server never carries quotes or
* skill picks). `context` is the RESTORE payload for a queued-origin steer:
* every degradation path threads it back into the requeue/send fallback so
* the item's quotes and manual skills survive. Composer-origin steers pass
* nothing, leaving their context staged in the composer atoms. */
/** POSTs a steer (text + files + quotes; the server merges the quotes into
* the model-bound turn at the injection boundary). `context` doubles as the
* RESTORE payload: every degradation path threads it back into the
* requeue/send fallback so the item's quotes and manual skills survive.
* Skill picks never ride the POST they configure a NEW turn's run, so a
* queued-origin steer only carries them for restoration. */
const submitSteer = useCallback(
(
text: string,
@ -951,6 +1016,7 @@ export default function useSteering({
clientSteerId: localId,
text: trimmed,
...(files && { files }),
...(carried.quotes && { quotes: carried.quotes }),
...(preempt && { preempt }),
...(targetGenerationCreatedAt != null && {
generationCreatedAt: targetGenerationCreatedAt,
@ -959,6 +1025,16 @@ export default function useSteering({
{
onSuccess: (response) => {
try {
/** A 202 without the echo means a pre-quotes replica queued
* the words without their excerpts. The quotes are NOT
* re-staged here the steer has not injected yet, so they
* stay carried on the pending chip: a quote-less applied
* event re-stages them at the actual loss, and a terminal
* leftover conversion carries them onto the recovered row
* instead (its normal send delivers quotes on any server).
* Only a settled replay an already-injected steer with no
* future event to re-home the chip reclaims immediately. */
const quotesRejected = carried.quotes != null && response.quotesAccepted !== true;
const canUseV2Receipt =
targetGenerationProtocolVersion === 2 && supportsGenerationProtocolV2(response);
if (canUseV2Receipt && response.settled === true) {
@ -978,6 +1054,9 @@ export default function useSteering({
...carried,
});
} else {
if (quotesRejected) {
reclaimRejectedChipQuotes(conversationId, [localId, response.steerId]);
}
settleReceiptReplay(conversationId, localId, response.steerId);
}
return;
@ -1005,6 +1084,9 @@ export default function useSteering({
...carried,
} satisfies PendingSteer;
if (acknowledgeSteer(conversationId, localId, acknowledged)) {
/** Terminal conversion re-homes the words as a queued
* follow-up that sends via `ask` the carried quotes ride
* it there, so nothing is re-staged. */
queueRecoveredSteer(acknowledged);
}
} finally {
@ -1128,6 +1210,7 @@ export default function useSteering({
acknowledgeSteer,
settleReceiptReplay,
queueRecoveredSteer,
reclaimRejectedChipQuotes,
steerMessage,
sendNow,
enqueue,
@ -1143,22 +1226,26 @@ export default function useSteering({
],
);
/** Composer-originated steer: consumes the composer's attachments so they
* ride the steer as one unit (the server re-fetches + encodes them at the
* injection boundary). Files are taken only after the guards pass. */
/** Composer-originated steer: consumes the composer's attachments and quote
* chips so they ride the steer as one unit (the server re-fetches + encodes
* files and merges quotes at the injection boundary). Both are taken only
* after the guards pass with `canSteer` true, `submitSteer` cannot
* refuse, so the drained context can never be stranded. */
const steerFromComposer = useCallback(
(text: string, preempt = false): boolean => {
const trimmed = text.trim();
if (trimmed.length === 0 || filesLoading || !canSteer) {
return false;
}
const consumed = submitSteer(trimmed, takeComposerFiles(), undefined, { preempt });
const consumed = submitSteer(trimmed, takeComposerFiles(), takeComposerQuotes(), {
preempt,
});
if (consumed) {
takeComposerDraft();
}
return consumed;
},
[filesLoading, canSteer, takeComposerFiles, takeComposerDraft, submitSteer],
[filesLoading, canSteer, takeComposerFiles, takeComposerQuotes, takeComposerDraft, submitSteer],
);
/** Composer-originated queue: carries the composer's attachments, quote
@ -1398,7 +1485,9 @@ export default function useSteering({
if (!hasRealConvoId) {
return interruptAndSend(trimmed);
}
const consumed = submitSteer(trimmed, takeComposerFiles(), undefined, { preempt: true });
const consumed = submitSteer(trimmed, takeComposerFiles(), takeComposerQuotes(), {
preempt: true,
});
if (consumed) {
takeComposerDraft();
}
@ -1411,6 +1500,7 @@ export default function useSteering({
hasRealConvoId,
interruptAndSend,
takeComposerFiles,
takeComposerQuotes,
takeComposerDraft,
submitSteer,
],

View file

@ -2429,8 +2429,22 @@ describe('useResumableSSE', () => {
});
}
expect(mockResolveSteerChip).toHaveBeenNthCalledWith(1, CONV_ID, 'server-1', 'client-1');
expect(mockResolveSteerChip).toHaveBeenNthCalledWith(2, CONV_ID, 'server-2', 'client-2');
// 4th arg: the applied part's quotes (absent here) — see the pre-quotes
// server restage in resolveSteerChip.
expect(mockResolveSteerChip).toHaveBeenNthCalledWith(
1,
CONV_ID,
'server-1',
'client-1',
undefined,
);
expect(mockResolveSteerChip).toHaveBeenNthCalledWith(
2,
CONV_ID,
'server-2',
'client-2',
undefined,
);
expect(requestFrame).toHaveBeenCalledTimes(2);
await act(async () => {

View file

@ -54,6 +54,8 @@ import {
findReasoningLabelMessageIndex,
appendAppliedSteerIds,
collectAppliedSteerIds,
collectDroppedSteerQuotes,
mergeRestagedQuotes,
removeConvoFromAllQueries,
upsertConvoInAllQueries,
countTaggedApprovalParts,
@ -857,9 +859,32 @@ export default function useResumableSSE(
* part becomes the durable record), and records the id so a 202 ACK that
* arrives AFTER the applied event drops its chip instead of re-minting it. */
const resolveSteerChip = useRecoilCallback(
({ set }) =>
(conversationId: string, steerId: string, clientSteerId?: string) => {
({ snapshot, set }) =>
(
conversationId: string,
steerId: string,
clientSteerId?: string,
appliedPartQuotes?: string[],
) => {
const settledIds = clientSteerId ? [steerId, clientSteerId] : [steerId];
/** A part applied by a pre-quotes server carries no quotes while the
* chip being settled may hold the only copy of the user's excerpts
* (its 202 was lost, so the ACK-echo restore never ran). Re-stage
* them before removal; `mergeRestagedQuotes` keeps this idempotent
* with the ACK path for the same excerpts. */
if (appliedPartQuotes == null || appliedPartQuotes.length === 0) {
const chips = snapshot
.getLoadable(store.pendingSteersByConvoId(conversationId))
.getValue();
const droppedQuotes = chips.find(
(steer) => settledIds.includes(steer.steerId) && (steer.quotes?.length ?? 0) > 0,
)?.quotes;
if (droppedQuotes != null && droppedQuotes.length > 0) {
set(store.pendingQuotesByConvoId(conversationId), (prev) =>
mergeRestagedQuotes(prev, droppedQuotes),
);
}
}
set(store.appliedSteerIdsByConvoId(conversationId), (prev) =>
appendAppliedSteerIds(prev, settledIds),
);
@ -960,7 +985,8 @@ export default function useResumableSSE(
/** Replaces the chip list with the server's still-queued steers (reconnect).
* Local `failed` entries are kept so their text stays recoverable, and a
* reseeded chip keeps its client-only quotes/skill picks. */
* reseeded chip keeps its quotes/skill picks (from the local chip, or the
* server item's persisted quotes when no chip survives). */
const seedSteerChips = useRecoilCallback(
({ set }) =>
(
@ -1013,7 +1039,10 @@ export default function useResumableSSE(
generationCreatedAt: chipGenerationCreatedAt,
}),
generationProtocolVersion,
...carriedSteerContext(localChip),
// The local chip carries skill picks the server never sees; a
// fresh tab has no chip, so fall back to the server item's
// persisted quotes rather than reseeding the chip without them.
...carriedSteerContext(localChip ?? steer),
};
}),
...prev.filter((steer) => steer.status === 'failed' && !claimedIds.has(steer.steerId)),
@ -1024,13 +1053,25 @@ export default function useResumableSSE(
);
const settleAppliedSteerParts = useRecoilCallback(
({ set }) =>
({ snapshot, set }) =>
(conversationId: string, values: unknown[] | undefined) => {
const ids = collectAppliedSteerIds(values);
if (ids.length === 0) {
return;
}
const settled = new Set(ids);
/** Chips settled by quote-less applied parts hold the only copy of
* their excerpts (a pre-quotes server injected the words bare)
* re-stage them as composer chips before the removal below. */
const droppedQuotes = collectDroppedSteerQuotes(
values,
snapshot.getLoadable(store.pendingSteersByConvoId(conversationId)).getValue(),
);
if (droppedQuotes.length > 0) {
set(store.pendingQuotesByConvoId(conversationId), (prev) =>
mergeRestagedQuotes(prev, droppedQuotes),
);
}
set(store.appliedSteerIdsByConvoId(conversationId), (prev) =>
appendAppliedSteerIds(prev, ids),
);
@ -1387,7 +1428,7 @@ export default function useResumableSSE(
* the chip pending during that wait lets an intervening error/final
* convert already-applied words into a duplicate queued message. */
if (attempt === 0) {
resolveSteerChip(chipConvoId, event.steerId, event.clientSteerId);
resolveSteerChip(chipConvoId, event.steerId, event.clientSteerId, event.part?.quotes);
}
const retryNextFrame = () => {
if (attempt < PENDING_ACTION_MAX_RETRY_FRAMES) {

View file

@ -14,6 +14,8 @@ import {
dedupeSteersById,
appendAppliedSteerIds,
collectAppliedSteerIds,
collectDroppedSteerQuotes,
mergeRestagedQuotes,
applyPendingAction,
carriedSteerContext,
getBranchSiblingIndexesForTarget,
@ -354,7 +356,10 @@ export default function useResumeOnLoad(
generationCreatedAt: chipGenerationCreatedAt,
}),
generationProtocolVersion,
...carriedSteerContext(localChip),
// The local chip carries skill picks the server never sees; a
// fresh tab has no chip, so fall back to the server item's
// persisted quotes rather than reseeding the chip without them.
...carriedSteerContext(localChip ?? steer),
};
}),
...prev.filter((steer) => steer.status === 'failed' && !claimedIds.has(steer.steerId)),
@ -365,13 +370,25 @@ export default function useResumeOnLoad(
);
const settleAppliedSteerParts = useRecoilCallback(
({ set }) =>
({ snapshot, set }) =>
(activeConversationId: string, values: unknown[] | undefined) => {
const ids = collectAppliedSteerIds(values);
if (ids.length === 0) {
return;
}
const settled = new Set(ids);
/** Chips settled by quote-less applied parts hold the only copy of
* their excerpts (a pre-quotes server injected the words bare)
* re-stage them as composer chips before the removal below. */
const droppedQuotes = collectDroppedSteerQuotes(
values,
snapshot.getLoadable(store.pendingSteersByConvoId(activeConversationId)).getValue(),
);
if (droppedQuotes.length > 0) {
set(store.pendingQuotesByConvoId(activeConversationId), (prev) =>
mergeRestagedQuotes(prev, droppedQuotes),
);
}
set(store.appliedSteerIdsByConvoId(activeConversationId), (prev) =>
appendAppliedSteerIds(prev, ids),
);

View file

@ -1780,6 +1780,7 @@
"com_ui_queue_send": "Queue message for after the response",
"com_ui_queued_attachment_count": "{{0}} attachments queued with this message",
"com_ui_queued_messages": "Queued messages",
"com_ui_queued_quote_count": "{{0}} quoted excerpts included with this message",
"com_ui_quote_selections": "{{0}} selections",
"com_ui_quotes_queued": "Quotes added for your next message",
"com_ui_ran_n_agents": "Ran {{0}} agents",

View file

@ -350,10 +350,12 @@ export type PendingSteer = {
createdAt: number;
/** Attachments steered with the message (refs; already uploaded). */
files?: TMessage['files'];
/** Quote chips carried by a queued-origin steer (client-only; never sent to
* the server), restored onto the queued item if the run ends first. */
/** Quoted excerpts riding this steer (also sent on the POST the server
* merges them into the injected turn); kept on the chip so a steer that
* never injects restores onto the queued item with them intact. */
quotes?: string[];
/** Manual skill picks carried the same way as `quotes`. */
/** Manual skill picks, carried for restoration only (a skill pick
* configures a NEW turn's run, so it never rides the steer POST). */
manualSkills?: string[];
/** Asked the run to seal generation at the next safe boundary rather than
* wait for a tool step. Labelling only the server owns the behaviour and

View file

@ -8,6 +8,8 @@ import {
appendAppliedSteerIds,
resolveAbortSteerTarget,
insertQueuedOrigin,
mergeRestagedQuotes,
collectDroppedSteerQuotes,
} from '../steer';
const buildEvent = (overrides: Partial<TSteerAppliedEvent> = {}): TSteerAppliedEvent => ({
@ -218,3 +220,56 @@ describe('insertQueuedOrigin', () => {
]);
});
});
describe('mergeRestagedQuotes', () => {
it('appends only fresh excerpts and keeps referential stability when none land', () => {
const prev = ['kept'];
expect(mergeRestagedQuotes(prev, ['kept'])).toBe(prev);
expect(mergeRestagedQuotes(prev, ['kept', 'new'])).toEqual(['kept', 'new']);
});
it('never grows past the sendable cap, letting already-staged chips win', () => {
// A chip beyond MAX_QUOTE_COUNT would render but silently miss the next
// send (both ends keep only the first 10) — drop the overflow explicitly.
const staged = Array.from({ length: 9 }, (_, i) => `staged-${i}`);
expect(mergeRestagedQuotes(staged, ['restored-a', 'restored-b'])).toEqual([
...staged,
'restored-a',
]);
const full = Array.from({ length: 10 }, (_, i) => `staged-${i}`);
expect(mergeRestagedQuotes(full, ['restored-a'])).toBe(full);
});
});
describe('collectDroppedSteerQuotes', () => {
const chips = [
{ steerId: 'srv-1', clientSteerId: 'local-1', quotes: ['excerpt one'] },
{ steerId: 'srv-2', quotes: ['excerpt two'] },
{ steerId: 'srv-3' },
];
it('collects quotes for chips whose applied part carries none', () => {
const values = [
{
content: [
{ type: ContentTypes.STEER, steerId: 'srv-1' },
{ type: ContentTypes.STEER, steerId: 'srv-2', quotes: ['excerpt two'] },
{ type: ContentTypes.STEER, steerId: 'srv-3' },
],
},
];
expect(collectDroppedSteerQuotes(values, chips)).toEqual(['excerpt one']);
});
it('matches a quote-less part by the client correlation id too', () => {
const values = [{ content: [{ type: ContentTypes.STEER, clientSteerId: 'local-1' }] }];
expect(collectDroppedSteerQuotes(values, chips)).toEqual(['excerpt one']);
});
it('returns nothing when every applied part kept its quotes', () => {
const values = [
{ content: [{ type: ContentTypes.STEER, steerId: 'srv-1', quotes: ['excerpt one'] }] },
];
expect(collectDroppedSteerQuotes(values, chips)).toEqual([]);
});
});

View file

@ -56,6 +56,89 @@ export function collectAppliedSteerIds(values: unknown[] | undefined): string[]
return [...ids];
}
/** Ids of applied steer parts that carry NO quotes, same traversal as
* `collectAppliedSteerIds`. Paired with a quote-bearing local chip, such a
* part proves a pre-quotes server injected the words bare the chip's
* excerpts must be re-staged before the settle removes their only copy. */
export function collectQuotelessAppliedSteerIds(values: unknown[] | undefined): Set<string> {
if (!values) {
return new Set();
}
const ids = new Set<string>();
for (const value of values) {
if (value == null || typeof value !== 'object') {
continue;
}
const object = value as { content?: unknown };
const parts = Array.isArray(object.content) ? object.content : [value];
for (const part of parts) {
if (part == null || typeof part !== 'object') {
continue;
}
const candidate = part as {
type?: unknown;
steerId?: unknown;
clientSteerId?: unknown;
quotes?: unknown;
};
if (candidate.type !== ContentTypes.STEER) {
continue;
}
if (Array.isArray(candidate.quotes) && candidate.quotes.length > 0) {
continue;
}
if (typeof candidate.steerId === 'string') {
ids.add(candidate.steerId);
}
if (typeof candidate.clientSteerId === 'string') {
ids.add(candidate.clientSteerId);
}
}
}
return ids;
}
/** Max excerpts staged at once; mirrors the backend `QUOTE_MAX_COUNT` cap so
* every displayed chip actually reaches the model on the next send. */
export const MAX_QUOTE_COUNT = 10;
/** Dedupe-appends re-staged excerpts onto the composer's pending-quote chips,
* returning `prev` untouched when nothing new lands (Recoil referential
* stability). The dedupe also makes the multiple restore triggers ACK echo,
* applied event, reconnect settle idempotent for the same excerpts. Capped
* at `MAX_QUOTE_COUNT` with the already-staged chips winning: a restored tail
* that cannot ride the next send is dropped explicitly rather than displayed
* as a chip the submission would silently discard. */
export function mergeRestagedQuotes(prev: string[], quotes: string[]): string[] {
const room = MAX_QUOTE_COUNT - prev.length;
if (room <= 0) {
return prev;
}
const fresh = quotes.filter((quote) => !prev.includes(quote)).slice(0, room);
return fresh.length > 0 ? [...prev, ...fresh] : prev;
}
/** Excerpts to re-stage when applied steer parts settle their chips: the
* quotes carried by each chip whose applied part has none proof a
* pre-quotes server injected the words bare, leaving the chip as the only
* copy of the user's excerpts. */
export function collectDroppedSteerQuotes(
values: unknown[] | undefined,
chips: readonly Pick<TPendingSteer, 'steerId' | 'clientSteerId' | 'quotes'>[],
): string[] {
const quoteless = collectQuotelessAppliedSteerIds(values);
if (quoteless.size === 0) {
return [];
}
return chips.flatMap((steer) =>
(steer.quotes?.length ?? 0) > 0 &&
(quoteless.has(steer.steerId) ||
(steer.clientSteerId != null && quoteless.has(steer.clientSteerId)))
? (steer.quotes ?? [])
: [],
);
}
/**
* Places an injected steer part at its absolute content index on the target
* response message. The server reserved that slot (subsequent SDK events were
@ -143,9 +226,11 @@ export function appendAppliedSteerIds(prev: string[], steerIds: string[]): strin
export type SteerCarriedContext = { quotes?: string[]; manualSkills?: string[] };
/** Quotes/skill picks are client-only (a steer never sends them to the
* server); chip mints, reseeds, and queued conversions carry them from the
* local source so the context survives a steer that never injects. */
/** Quotes ride the steer POST (the server merges them into the injected
* turn) but chips, reseeds, and queued conversions still carry them locally
* so a steer that never injects restores with its excerpts intact. Skill
* picks are client-only they configure a NEW turn's run, so only the
* restore paths carry them. */
export function carriedSteerContext(source?: SteerCarriedContext): SteerCarriedContext {
const quotes = source?.quotes;
const manualSkills = source?.manualSkills;

View file

@ -1,7 +1,7 @@
import type { IMongoFile } from '@librechat/data-schemas';
import type { SteerFileFetcher } from '../request';
import type { SteerMediaClient } from '../media';
import { buildSteerMedia, stampSteerPartMedia } from '../media';
import { buildSteerMedia, collectSteerStampTargets, stampSteerPartMedia } from '../media';
jest.spyOn(console, 'log').mockImplementation();
@ -180,6 +180,23 @@ describe('buildSteerMedia', () => {
{},
);
});
it('merges quoted excerpts into the encoded text part', async () => {
const getFiles: SteerFileFetcher = jest.fn(async () => [imageDoc]);
const client = createClient({ image_urls: [imagePart] });
const result = await buildSteerMedia({
client,
user,
item: { ...steerItem([{ file_id: 'f1' }], 'what about this?'), quotes: ['the excerpt'] },
getFiles,
});
expect(result?.content).toEqual([
{ type: 'text', text: '> the excerpt\n\nwhat about this?' },
imagePart,
]);
});
});
describe('stampSteerPartMedia', () => {
@ -268,4 +285,148 @@ describe('stampSteerPartMedia', () => {
expect((message.content as unknown[])[0]).toBe(steerPart);
expect(steerPart).not.toHaveProperty('media');
});
it('stamps merged text media for a quote-bearing part without files', async () => {
const getFiles: SteerFileFetcher = jest.fn(async () => []);
const client = createClient();
const steerPart = {
type: 'steer',
steer: 'and this part?',
steerId: 's3',
quotes: ['first excerpt', 'second excerpt'],
};
const message = { messageId: 'assistant-q', role: 'assistant', content: [steerPart] };
const stamped = await stampSteerPartMedia({ client, user, payload: [message], getFiles });
expect(getFiles).not.toHaveBeenCalled();
expect(client.processAttachments).not.toHaveBeenCalled();
const merged = '> first excerpt\n\n> second excerpt\n\nand this part?';
expect((message.content as Array<Record<string, unknown>>)[0].media).toEqual([
{ type: 'text', text: merged },
]);
expect(steerPart).not.toHaveProperty('media');
expect(stamped).toEqual([
{
index: 0,
sourceMessageId: 'assistant-q',
fileIds: [],
media: [{ type: 'text', text: merged }],
steerText: 'and this part?',
},
]);
});
it('merges quotes into the encoded text part of a files-carrying steer', async () => {
const getFiles: SteerFileFetcher = jest.fn(async () => [imageDoc]);
const client = createClient({ image_urls: [imagePart] });
const steerPart = {
type: 'steer',
steer: 'see attachment',
steerId: 's4',
files: [{ file_id: 'f1' }],
quotes: ['quoted line'],
};
const message = { role: 'assistant', content: [steerPart] };
const stamped = await stampSteerPartMedia({ client, user, payload: [message], getFiles });
expect(stamped[0].media).toEqual([
{ type: 'text', text: '> quoted line\n\nsee attachment' },
imagePart,
]);
expect(stamped[0].steerText).toBe('see attachment');
});
it('still stamps merged text when a quote-bearing part loses its files', async () => {
const getFiles: SteerFileFetcher = jest.fn(async () => []);
const steerPart = {
type: 'steer',
steer: 'orphaned but quoted',
steerId: 's5',
files: [{ file_id: 'gone' }],
quotes: ['the reference'],
};
const message = { role: 'assistant', content: [steerPart] };
const stamped = await stampSteerPartMedia({
client: createClient(),
user,
payload: [message],
getFiles,
});
expect(stamped[0].fileIds).toEqual([]);
expect(stamped[0].media).toEqual([
{ type: 'text', text: '> the reference\n\norphaned but quoted' },
]);
});
it('collects stamp targets synchronously so steer-free payloads skip the await', () => {
const plain = [
{ role: 'user', content: 'hi' },
{ role: 'assistant', content: [{ type: 'text', text: 'answer' }] },
];
expect(collectSteerStampTargets(plain, true)).toHaveLength(0);
const filesOnly = [
{ role: 'assistant', content: [{ type: 'steer', steer: 's', files: [{ file_id: 'f1' }] }] },
];
expect(collectSteerStampTargets(filesOnly, true)).toHaveLength(1);
expect(collectSteerStampTargets(filesOnly, false)).toHaveLength(0);
const quoted = [{ role: 'assistant', content: [{ type: 'steer', steer: 's', quotes: ['q'] }] }];
expect(collectSteerStampTargets(quoted, false)).toHaveLength(1);
});
it('consumes pre-collected targets without re-scanning the payload', async () => {
const getFiles: SteerFileFetcher = jest.fn(async () => []);
const steerPart = { type: 'steer', steer: 'quoted turn', steerId: 's8', quotes: ['kept'] };
const message = { role: 'assistant', content: [steerPart] };
const targets = collectSteerStampTargets([message], false);
const stamped = await stampSteerPartMedia({
client: createClient(),
user,
payload: [message],
targets,
getFiles,
resendFiles: false,
});
expect(stamped[0].media).toEqual([{ type: 'text', text: '> kept\n\nquoted turn' }]);
});
it('replays quotes without encoding files when resendFiles is off', async () => {
const getFiles: SteerFileFetcher = jest.fn(async () => [imageDoc]);
const client = createClient({ image_urls: [imagePart] });
const quotedPart = {
type: 'steer',
steer: 'quoted turn',
steerId: 's6',
files: [{ file_id: 'f1' }],
quotes: ['kept excerpt'],
};
const filesOnlyPart = {
type: 'steer',
steer: 'files only',
steerId: 's7',
files: [{ file_id: 'f1' }],
};
const message = { role: 'assistant', content: [quotedPart, filesOnlyPart] };
const stamped = await stampSteerPartMedia({
client,
user,
payload: [message],
getFiles,
resendFiles: false,
});
expect(getFiles).not.toHaveBeenCalled();
expect(client.processAttachments).not.toHaveBeenCalled();
expect(stamped).toHaveLength(1);
expect(stamped[0].media).toEqual([{ type: 'text', text: '> kept excerpt\n\nquoted turn' }]);
expect((message.content as Array<Record<string, unknown>>)[1]).toBe(filesOnlyPart);
});
});

View file

@ -296,6 +296,89 @@ describe('handleSteerRequest (real in-memory job manager)', () => {
]);
});
it('normalizes quoted excerpts into the queue item like the chat route', async () => {
const streamId = 'steer-req-quotes';
await GenerationJobManager.createJob(streamId, user.id, undefined, {
initialMetadata: { steerQuotesCapable: true },
});
const result = await handleSteerRequest(user, {
conversationId: streamId,
text: 'about the selection',
quotes: [' kept excerpt ', '', 42, 'second'],
});
expect(result.status).toBe(202);
expect(result.body.quotesAccepted).toBe(true);
const queued = await GenerationJobManager.steering.peek(streamId);
expect(queued[0].quotes).toEqual(['kept excerpt', 'second']);
});
it('atomically strips quotes when a legacy HITL handover races the admission', async () => {
// A resume keeps createdAt, so the enqueue fence cannot see the handover.
// A LEGACY resumer rewrites providerExecutionId without knowing the quote
// marker, which invalidates the previous owner's assertion; the enqueue
// transaction evaluates that equality against the LIVE job — after the
// admission's own capability read already said capable — and the returned
// persisted item keeps the echo honest.
const streamId = 'steer-req-quotes-downgrade';
await GenerationJobManager.createJob(streamId, user.id, undefined, {
initialMetadata: { steerQuotesCapable: true },
});
const result = await handleSteerRequest(
user,
{ conversationId: streamId, text: 'about the selection', quotes: ['the excerpt'] },
{
checkAgentAccess: async () => {
const stored = await GenerationJobManager.getJobStore().getJob(streamId);
(stored as { providerExecutionId?: string }).providerExecutionId = 'legacy-resume-exec';
return true;
},
},
);
expect(result.status).toBe(202);
expect(result.body).not.toHaveProperty('quotesAccepted');
const queued = await GenerationJobManager.steering.peek(streamId);
expect(queued[0]).not.toHaveProperty('quotes');
});
it('drops quotes without the echo when the generation owner cannot merge them', async () => {
// The job was created by a pre-quotes replica (no capability flag): an
// upgraded admission replica must not store quotes its owning drain would
// silently ignore — the missing echo makes the client re-stage them.
const streamId = 'steer-req-quotes-incapable-owner';
await GenerationJobManager.createJob(streamId, user.id);
const result = await handleSteerRequest(user, {
conversationId: streamId,
text: 'about the selection',
quotes: ['the excerpt'],
});
expect(result.status).toBe(202);
expect(result.body).not.toHaveProperty('quotesAccepted');
const queued = await GenerationJobManager.steering.peek(streamId);
expect(queued[0]).not.toHaveProperty('quotes');
});
it('omits quotes from the queue item when nothing usable was sent', async () => {
const streamId = 'steer-req-no-quotes';
await GenerationJobManager.createJob(streamId, user.id);
const result = await handleSteerRequest(user, {
conversationId: streamId,
text: 'plain steer',
quotes: 'not-an-array',
});
expect(result.status).toBe(202);
expect(result.body).not.toHaveProperty('quotesAccepted');
const queued = await GenerationJobManager.steering.peek(streamId);
expect(queued[0]).not.toHaveProperty('quotes');
});
describe('injected getFiles (owner-scoped resolve at enqueue)', () => {
const dbDoc = {
file_id: 'f1',
@ -805,6 +888,103 @@ describe('generation protocol bridge for steering mutations', () => {
expect(publishUpdate).toHaveBeenCalledTimes(1);
});
it('keeps receipt fingerprints quote-independent so legacy replicas can replay them', async () => {
// The 3-field hash is the one shape EVERY deployed version computes: a
// lost-ACK retry of a quoted steer routed through a pre-quotes replica
// must replay the receipt, not 409 accepted words as a conflict.
const streamId = 'steer-protocol-v2-legacy-replayable';
await GenerationJobManager.createJob(streamId, user.id, undefined, {
initialMetadata: { generationProtocolVersion: 2, steerQuotesCapable: true },
});
const base = { conversationId: streamId, text: 'identical words' };
await handleSteerRequest(
user,
{ ...base, clientSteerId: 'client-quoted', quotes: ['the excerpt'] },
{ generationProtocolVersion: 2 },
);
await handleSteerRequest(
user,
{ ...base, clientSteerId: 'client-plain' },
{ generationProtocolVersion: 2 },
);
const quoted = await GenerationJobManager.steering.getReceipt(streamId, 'client-quoted');
const plain = await GenerationJobManager.steering.getReceipt(streamId, 'client-plain');
expect(quoted?.fingerprint).toBe(plain?.fingerprint);
expect(typeof quoted?.requestedQuotesFingerprint).toBe('string');
expect(plain?.requestedQuotesFingerprint).toBeUndefined();
});
it('treats quotes as part of the idempotency identity', async () => {
const streamId = 'steer-protocol-v2-quote-fingerprint';
await GenerationJobManager.createJob(streamId, user.id, undefined, {
initialMetadata: { generationProtocolVersion: 2, steerQuotesCapable: true },
});
const requestBody = {
conversationId: streamId,
clientSteerId: 'client-v2-quoted',
text: 'about this excerpt',
quotes: ['the excerpt'],
};
const accepted = await handleSteerRequest(user, requestBody, {
generationProtocolVersion: 2,
});
const replayed = await handleSteerRequest(user, requestBody, {
generationProtocolVersion: 2,
});
const conflicting = await handleSteerRequest(
user,
{ ...requestBody, quotes: ['a different excerpt'] },
{ generationProtocolVersion: 2 },
);
expect(accepted.status).toBe(202);
expect(accepted.body.quotesAccepted).toBe(true);
expect(replayed.body).toMatchObject({
steerId: accepted.body.steerId,
replayed: true,
// Echoed from the durable item so a lost-ACK retry still learns the
// excerpts were attached to the accepted words.
quotesAccepted: true,
});
expect(conflicting.status).toBe(409);
expect(conflicting.body.code).toBe('STEER_IDEMPOTENCY_CONFLICT');
});
it('replays a legacy quote-less receipt for a quoted retry of the same words', async () => {
// Cross-version lost ACK: a pre-quotes replica accepted the words and its
// receipt hashes only text/files/preempt. The retry now carries quotes —
// it must replay that receipt (the words are already durable) and OMIT the
// quotesAccepted echo so the client re-stages the dropped excerpts.
const streamId = 'steer-protocol-v2-legacy-fingerprint';
await GenerationJobManager.createJob(streamId, user.id, undefined, {
initialMetadata: { generationProtocolVersion: 2 },
});
const requestBody = {
conversationId: streamId,
clientSteerId: 'client-v2-legacy-quoted',
text: 'same accepted words',
};
const receiptEnqueue = jest.spyOn(GenerationJobManager.steering, 'enqueueWithReceipt');
const accepted = await handleSteerRequest(user, requestBody, {
generationProtocolVersion: 2,
});
const quotedRetry = await handleSteerRequest(
user,
{ ...requestBody, quotes: ['the excerpt'] },
{ generationProtocolVersion: 2 },
);
expect(accepted.status).toBe(202);
expect(quotedRetry.status).toBe(202);
expect(quotedRetry.body).toMatchObject({ steerId: accepted.body.steerId, replayed: true });
expect(quotedRetry.body).not.toHaveProperty('quotesAccepted');
expect(receiptEnqueue).toHaveBeenCalledTimes(1);
});
it('replays a v2 receipt after terminal cleanup deletes the accepting job', async () => {
const streamId = 'steer-protocol-v2-replay-after-delete';
const job = await GenerationJobManager.createJob(streamId, user.id, undefined, {

View file

@ -6,10 +6,9 @@ import type {
} from '@librechat/agents';
import type { SteerQueueItem } from '~/stream/interfaces/IJobStore';
/** Mirrors runtime.ts's local extension — the field predates the SDK pin bump. */
type SteerDrainOutput = PostToolBatchHookOutput & {
injectedMessages?: Array<{ role: string; content: string; source: string }>;
};
/** The pinned SDK's hook output declares `injectedMessages` natively; a
* narrower local re-declaration would no longer be assignable from it. */
type SteerDrainOutput = PostToolBatchHookOutput;
import { InMemoryEventTransport } from '~/stream/implementations/InMemoryEventTransport';
import { InMemoryJobStore } from '~/stream/implementations/InMemoryJobStore';
import { GenerationJobManager } from '~/stream/GenerationJobManager';
@ -204,6 +203,54 @@ describe('createSteerDrainHook', () => {
]);
});
it('merges quoted excerpts into text-only injections (media path merges its own)', async () => {
const streamId = `drain-quotes-${Date.now()}`;
const job = await GenerationJobManager.createJob(streamId, 'user-1', undefined, {
initialMetadata: { steerQuotesCapable: true },
});
await GenerationJobManager.steering.enqueue(streamId, {
...buildSteer('s1', 'what does this mean?'),
quotes: ['selected passage'],
});
const hook = createSteerDrainHook({
streamId,
jobCreatedAt: job.createdAt,
applySteer: jest.fn(),
});
const output: SteerDrainOutput = await hook(batchInput(), abortSignal);
expect(output.injectedMessages).toEqual([
{ role: 'user', content: '> selected passage\n\nwhat does this mean?', source: 'steer' },
]);
});
it('keeps quotes in the injection when media encoding degrades to text', async () => {
const streamId = `drain-quotes-degrade-${Date.now()}`;
const job = await GenerationJobManager.createJob(streamId, 'user-1', undefined, {
initialMetadata: { steerQuotesCapable: true },
});
await GenerationJobManager.steering.enqueue(streamId, {
...buildSteer('s1', 'and the doc?'),
files: [{ file_id: 'f1', type: 'image/png' }],
quotes: ['quoted context'],
});
const hook = createSteerDrainHook({
streamId,
jobCreatedAt: job.createdAt,
applySteer: jest.fn(),
buildMedia: jest.fn(async () => {
throw new Error('encode failed');
}),
});
const output: SteerDrainOutput = await hook(batchInput(), abortSignal);
expect(output.injectedMessages).toEqual([
{ role: 'user', content: '> quoted context\n\nand the doc?', source: 'steer' },
]);
});
it('persists the steer part BEFORE media encoding (abort-safe ordering)', async () => {
const streamId = `drain-apply-first-${Date.now()}`;
const job = await GenerationJobManager.createJob(streamId, 'user-1');

View file

@ -21,8 +21,8 @@ export type {
SteerFileFetcher,
SteerRequestResult,
} from './request';
export { buildSteerMedia, stampSteerPartMedia } from './media';
export type { SteerMediaClient, StampedSteerMedia } from './media';
export { buildSteerMedia, collectSteerStampTargets, stampSteerPartMedia } from './media';
export type { SteerMediaClient, SteerStampTarget, StampedSteerMedia } from './media';
export { createSteerIndexOffsetHandlers } from './offset';
export type { SteerOffsetState } from './offset';
export { toSteerFileRef } from './refs';

View file

@ -8,6 +8,7 @@ import type { SteerFileFetcher } from './request';
import type { SteerMediaResult } from './runtime';
import type { SteerRequestUser } from './refs';
import { toSteerFileRef, collectFileIds, buildOwnerFilter } from './refs';
import { getReferencedQuotes, mergeQuotedText } from '~/utils';
import { prependFileContext } from '../client';
/** The BaseClient encode surface the steer media pipeline reuses. */
@ -33,6 +34,7 @@ interface SteerPart {
type?: string;
steerId?: string;
files?: Partial<TFile>[];
quotes?: string[];
media?: Array<Record<string, unknown>>;
[key: string]: unknown;
}
@ -49,32 +51,44 @@ export interface StampedSteerMedia {
steerText: string;
}
/** The model-bound body for a steer: quoted excerpts prepended as Markdown
* blockquotes, exactly like `prependQuotes` does for regular user turns. The
* persisted part keeps `steer`/`quotes` separate; only this boundary merges. */
function mergeSteerModelText(text: string, quotes?: string[] | null): string {
const normalized = getReferencedQuotes(quotes);
return normalized != null ? mergeQuotedText(text, normalized) : text;
}
/**
* Encodes authorized file docs for one steer and assembles the multimodal
* content array, reusing the exact pipeline regular user turns go through:
* `addFileContextToMessage` + `processAttachments` (single-pass categorize +
* encode images/documents/videos/audios) on a throwaway message, then the
* SDK's `formatMessage` for part ordering (no `endpoint` arg matching the
* agents payload path, which formats without one).
* agents payload path, which formats without one). Quoted excerpts merge into
* the text part so the model receives them wherever the content array lands.
*/
async function encodeSteerContent({
client,
text,
quotes,
steerId,
fileDocs,
}: {
client: SteerMediaClient;
text: string;
quotes?: string[] | null;
steerId: string;
fileDocs: IMongoFile[];
}): Promise<SteerMediaResult> {
const modelText = mergeSteerModelText(text, quotes);
const pseudo: PseudoMessage = { messageId: `steer:${steerId}` };
await client.addFileContextToMessage(pseudo, fileDocs);
const validated = await client.processAttachments(pseudo, fileDocs);
const formatted = formatMessage({
message: {
role: 'user',
content: text,
content: modelText,
image_urls: pseudo.image_urls,
documents: pseudo.documents,
videos: pseudo.videos,
@ -86,7 +100,7 @@ async function encodeSteerContent({
}
const content = Array.isArray(formatted.content)
? formatted.content
: [{ type: ContentTypes.TEXT, text: formatted.content ?? text }];
: [{ type: ContentTypes.TEXT, text: formatted.content ?? modelText }];
const refSource = Array.isArray(validated) && validated.length > 0 ? validated : fileDocs;
const files = refSource.map(toSteerFileRef).filter((ref): ref is Partial<TFile> => ref != null);
return { content, files };
@ -128,16 +142,71 @@ export async function buildSteerMedia({
.map((id) => docsById.get(id))
.filter((doc): doc is IMongoFile => doc != null);
assertFilesAllowed?.(fileDocs);
return encodeSteerContent({ client, text: item.text, steerId: item.steerId, fileDocs });
return encodeSteerContent({
client,
text: item.text,
quotes: item.quotes,
steerId: item.steerId,
fileDocs,
});
}
export interface SteerStampTarget {
message: { id?: string; messageId?: string; content?: unknown };
part: SteerPart;
index: number;
quotes: string[] | null;
encodeFiles: boolean;
}
export type SteerStampPayload = Array<{
id?: string;
messageId?: string;
role?: string;
content?: unknown;
}>;
/** One pass over the payload for everything the stamp needs. Callers check
* `.length` for the zero-await fast path and hand the result to
* `stampSteerPartMedia`, so the history is never scanned twice. */
export function collectSteerStampTargets(
payload: SteerStampPayload,
resendFiles: boolean,
): SteerStampTarget[] {
const targets: SteerStampTarget[] = [];
for (let index = 0; index < payload.length; index++) {
const message = payload[index];
if (message?.role !== 'assistant' || !Array.isArray(message.content)) {
continue;
}
for (const part of message.content as SteerPart[]) {
if (part?.type !== ContentTypes.STEER) {
continue;
}
const quotes = getReferencedQuotes(part.quotes);
const encodeFiles = resendFiles && Array.isArray(part.files) && part.files.length > 0;
if (encodeFiles || quotes != null) {
targets.push({ message, part, index, quotes, encodeFiles });
}
}
}
return targets;
}
/**
* Re-encodes attachments for persisted steer parts of PAST turns and stamps
* the assembled content array as a transient `media` field, which the SDK's
* `formatAgentMessages` prefers over the plain text when reconstructing the
* steer's HumanMessage. Refs are re-encoded per turn encoded data is never
* persisted and parts are replaced immutably so the stamp cannot leak into
* a message save. Encodes run in parallel after doc resolution.
* Re-encodes attachments and re-merges quotes for persisted steer parts of
* PAST turns, stamping the assembled content array as a transient `media`
* field, which the SDK's `formatAgentMessages` prefers over the plain text
* when reconstructing the steer's HumanMessage. Refs are re-encoded per turn
* encoded data is never persisted and parts are replaced immutably so the
* stamp cannot leak into a message save. Encodes run in parallel after doc
* resolution.
*
* Quote-bearing parts are stamped UNCONDITIONALLY (a merged text part is the
* only way the excerpts reach the model on replay, mirroring `prependQuotes`
* for regular user turns), while file encoding remains gated on the
* conversation's `resendFiles` setting a quote-bearing part whose files are
* not resent still replays its quotes, exactly like its text.
*
* `docsById` should be the owner-scoped doc map `addPreviousAttachments`
* already fetched this turn (its single historical-files query collects
@ -149,85 +218,101 @@ export async function stampSteerPartMedia({
client,
user,
payload,
targets,
docsById,
getFiles,
resendFiles = true,
}: {
client: SteerMediaClient;
user: SteerRequestUser | undefined;
payload: Array<{ id?: string; messageId?: string; role?: string; content?: unknown }>;
payload: SteerStampPayload;
/** Pre-collected via `collectSteerStampTargets` so the caller's zero-await
* probe and this stamp share one payload scan; collected here otherwise. */
targets?: SteerStampTarget[];
docsById?: Map<string, IMongoFile>;
getFiles: SteerFileFetcher;
resendFiles?: boolean;
}): Promise<StampedSteerMedia[]> {
const stampTargets: Array<{
message: { id?: string; messageId?: string; content?: unknown };
part: SteerPart;
index: number;
}> = [];
for (let index = 0; index < payload.length; index++) {
const message = payload[index];
if (message?.role !== 'assistant' || !Array.isArray(message.content)) {
continue;
}
for (const part of message.content as SteerPart[]) {
if (part?.type === ContentTypes.STEER && Array.isArray(part.files) && part.files.length > 0) {
stampTargets.push({ message, part, index });
}
}
}
const stampTargets = targets ?? collectSteerStampTargets(payload, resendFiles);
if (stampTargets.length === 0) {
return [];
}
let resolvedDocsById = docsById;
if (resolvedDocsById == null) {
const allIds = collectFileIds(stampTargets.flatMap(({ part }) => part.files ?? []));
const fileTargets = stampTargets.filter(({ encodeFiles }) => encodeFiles);
if (resolvedDocsById == null && fileTargets.length > 0) {
const allIds = collectFileIds(fileTargets.flatMap(({ part }) => part.files ?? []));
const filter = buildOwnerFilter(allIds, user);
if (filter == null) {
return [];
if (filter != null) {
const fileDocs = await getFiles(filter, {}, {});
if (Array.isArray(fileDocs) && fileDocs.length > 0) {
resolvedDocsById = new Map(fileDocs.map((file) => [file.file_id, file]));
}
}
const fileDocs = await getFiles(filter, {}, {});
if (!Array.isArray(fileDocs) || fileDocs.length === 0) {
return [];
}
resolvedDocsById = new Map(fileDocs.map((file) => [file.file_id, file]));
}
const docs = resolvedDocsById;
const stamped: Array<StampedSteerMedia | null> = await Promise.all(
stampTargets.map(async ({ message, part, index }): Promise<StampedSteerMedia | null> => {
const partDocs = (part.files ?? [])
.map((file) => (file?.file_id != null ? docs.get(file.file_id) : undefined))
.filter((doc): doc is IMongoFile => doc != null);
if (partDocs.length === 0) {
return null;
}
try {
const { content, files } = await encodeSteerContent({
client,
text: (part[ContentTypes.STEER] as string | undefined) ?? '',
steerId: part.steerId ?? 'replay',
fileDocs: partDocs,
});
message.content = (message.content as SteerPart[]).map((candidate) =>
candidate === part ? { ...candidate, media: content } : candidate,
);
return {
index,
sourceMessageId: message.messageId ?? message.id,
fileIds: (files ?? [])
.map((file) => file.file_id)
.filter((fileId): fileId is string => typeof fileId === 'string' && fileId.length > 0),
media: content,
steerText: (part[ContentTypes.STEER] as string | undefined) ?? '',
stampTargets.map(
async ({ message, part, index, quotes, encodeFiles }): Promise<StampedSteerMedia | null> => {
const steerText = (part[ContentTypes.STEER] as string | undefined) ?? '';
const partDocs = encodeFiles
? (part.files ?? [])
.map((file) => (file?.file_id != null ? docs?.get(file.file_id) : undefined))
.filter((doc): doc is IMongoFile => doc != null)
: [];
const stampPart = (content: Array<Record<string, unknown>>, fileIds: string[]) => {
message.content = (message.content as SteerPart[]).map((candidate) =>
candidate === part ? { ...candidate, media: content } : candidate,
);
return {
index,
sourceMessageId: message.messageId ?? message.id,
fileIds,
media: content,
steerText,
};
};
} catch (error) {
logger.warn(
`[stampSteerPartMedia] Failed to re-encode steer media (steer=${part.steerId}); replaying text only`,
error,
);
return null;
}
}),
/** No authorized docs (or files not resent): a quote-bearing part
* still stamps its merged text so the excerpts replay; a files-only
* part falls back to plain-text replay exactly as before. */
const stampMergedTextOnly = () => {
if (quotes == null) {
return null;
}
return stampPart(
[{ type: ContentTypes.TEXT, text: mergeSteerModelText(steerText, quotes) }],
[],
);
};
if (partDocs.length === 0) {
return stampMergedTextOnly();
}
try {
const { content, files } = await encodeSteerContent({
client,
text: steerText,
quotes,
steerId: part.steerId ?? 'replay',
fileDocs: partDocs,
});
return stampPart(
content,
(files ?? [])
.map((file) => file.file_id)
.filter(
(fileId): fileId is string => typeof fileId === 'string' && fileId.length > 0,
),
);
} catch (error) {
logger.warn(
`[stampSteerPartMedia] Failed to re-encode steer media (steer=${part.steerId}); replaying text only`,
error,
);
return stampMergedTextOnly();
}
},
),
);
return stamped.filter((entry): entry is StampedSteerMedia => entry != null);
}

View file

@ -17,6 +17,7 @@ import {
import { toSteerFileRef, collectFileIds, buildOwnerFilter } from './refs';
import { GenerationJobManager } from '~/stream/GenerationJobManager';
import { isSteeringSupported } from './runtime';
import { getReferencedQuotes } from '~/utils';
/** Attachment cap per steer, mirroring the composer's practical limits. */
export const STEER_MAX_FILES = 10;
@ -48,6 +49,10 @@ export interface SteerRequestBody {
text?: unknown;
clientSteerId?: unknown;
files?: unknown;
/** Quoted excerpts steered with the message ("Add to chat" selections);
* normalized like the chat route's quotes and merged into the model-bound
* turn at the injection boundary. */
quotes?: unknown;
/** Ask the generating replica to seal the live model stream at the next
* provider-safe boundary instead of waiting for a tool step. NEVER a
* rejection reason: on an SDK without the capability the steer still
@ -230,6 +235,11 @@ function hasTenantMismatch(
return metadata?.tenantId != null && metadata.tenantId !== user.tenantId;
}
/** DELIBERATELY quote-independent: this exact 3-field hash is what EVERY
* deployed replica version computes, so a lost-ACK retry can replay its
* receipt no matter which replica wrote it or reads it. Quote identity is
* enforced separately via `SteerReceipt.requestedQuotesFingerprint`, which
* only quote-aware readers consult. */
function steerFingerprint(
text: string,
files: Partial<TFile>[] | undefined,
@ -240,6 +250,28 @@ function steerFingerprint(
.digest('base64url');
}
/** Normalized-quote identity stored beside (never inside) the fingerprint. */
function quotesFingerprint(quotes: string[]): string {
return createHash('sha256').update(JSON.stringify(quotes)).digest('base64url');
}
/** Whether a receipt's recorded quote identity accepts this request's quotes.
* An ABSENT record means the receipt was written by a pre-quotes replica (or
* for a quote-less request) quotes were never part of its contract, so any
* retry of the same words replays (the item carries no quotes; the missing
* `quotesAccepted` echo keeps the client's copy on its chip). A present
* record must match exactly: reusing a clientSteerId with different quotes
* is the same conflict a content-hash mismatch signals. */
function receiptQuotesCompatible(
recorded: string | undefined,
requested: string[] | null,
): boolean {
if (recorded == null) {
return true;
}
return requested != null && quotesFingerprint(requested) === recorded;
}
function receiptResponse(conversationId: string, receipt: SteerReceipt): SteerRequestResult {
return {
status: 202,
@ -252,6 +284,13 @@ function receiptResponse(conversationId: string, receipt: SteerReceipt): SteerRe
settled: receipt.state !== 'queued' && receipt.state !== 'claimed',
leftover: receipt.state === 'leftover',
replayed: true,
/** From the DURABLE item, mirroring the fresh 202: a receipt written by
* a pre-quotes replica replays without this marker, telling the client
* its excerpts never attached to the accepted words. */
...(receipt.item.quotes != null &&
receipt.item.quotes.length > 0 && {
quotesAccepted: true,
}),
...(receipt.item.preemptRevision != null && {
preemptRevision: receipt.item.preemptRevision,
}),
@ -401,6 +440,10 @@ async function handleSteerRequestInternal(
return { status: 400, body: { code: filesError } };
}
/** Same normalization as the chat route (trim, drop empties, cap count and
* excerpt length) so a steer's quotes obey the caps a normal send does. */
const quotes = getReferencedQuotes(body.quotes);
/** streamId === conversationId for resumable agent jobs */
const streamId = conversationId;
const wantsPreempt = body.preempt === true;
@ -430,7 +473,10 @@ async function handleSteerRequestInternal(
) {
return { status: 409, body: { code: 'RUN_REPLACED' } };
}
if (receipt.fingerprint !== fingerprint) {
if (
receipt.fingerprint !== fingerprint ||
!receiptQuotesCompatible(receipt.requestedQuotesFingerprint, quotes)
) {
return { status: 409, body: { code: 'STEER_IDEMPOTENCY_CONFLICT' } };
}
if (
@ -560,6 +606,19 @@ async function handleSteerRequestInternal(
if (isAborted(deps.signal)) {
return { status: 499, body: { code: 'STEER_ABORTED' } };
}
/** The OWNER's execution-bound capability: an upgraded admission replica
* must not store quotes (and claim them accepted) for a generation whose
* owning drain would silently drop them at injection. This read is only
* the FAST PATH the enqueue transaction re-evaluates the same
* marker-equals-execution predicate atomically against the live job and
* strips `item.quotes` itself, so a HITL handover landing after this read
* (same `createdAt`, invisible to the enqueue fence) cannot smuggle
* quotes past a legacy owner. The returned persisted item reflects any
* strip, keeping the `quotesAccepted` echo honest; on a missing echo the
* client re-stages the excerpts. */
const ownerAcceptsQuotes =
owner.metadata?.steerQuotesExecutionId != null &&
owner.metadata.steerQuotesExecutionId === owner.metadata.providerExecutionId;
const item: SteerQueueItem = {
steerId: randomUUID(),
...(protocol.value === 2 && typeof clientSteerId === 'string' && { clientSteerId }),
@ -567,6 +626,7 @@ async function handleSteerRequestInternal(
userId: user.id ?? '',
createdAt: Date.now(),
...(queuedFiles && { files: queuedFiles }),
...(quotes != null && ownerAcceptsQuotes && { quotes }),
};
/**
* Fenced to the generation the capability decision was made against. The
@ -588,6 +648,7 @@ async function handleSteerRequestInternal(
{
clientSteerId,
fingerprint,
...(quotes != null && { requestedQuotesFingerprint: quotesFingerprint(quotes) }),
userId: user.id ?? '',
...(user.tenantId && { tenantId: user.tenantId }),
...(job.metadata?.agent_id && { agentId: job.metadata.agent_id }),
@ -600,7 +661,11 @@ async function handleSteerRequestInternal(
if (typeof result === 'number') {
depth = result;
} else {
if (!('fingerprint' in result) || result.fingerprint !== fingerprint) {
if (
!('fingerprint' in result) ||
result.fingerprint !== fingerprint ||
!receiptQuotesCompatible(result.requestedQuotesFingerprint, quotes)
) {
return { status: 409, body: { code: 'STEER_IDEMPOTENCY_CONFLICT' } };
}
if (result.userId !== (user.id ?? '') || hasTenantMismatch(result, user)) {
@ -724,6 +789,14 @@ async function handleSteerRequestInternal(
position: depth,
conversationId,
preempt: preemptArmed,
/** Echoed from the DURABLE item so the client can tell whether its
* quoted excerpts will actually inject. A pre-quotes replica never
* sets this, and the client re-stages the excerpts on that absence
* a 202 must not silently drop model-bound context. */
...(persistedItem.quotes != null &&
persistedItem.quotes.length > 0 && {
quotesAccepted: true,
}),
...(protocol.value === 2 && preemptRevision != null && { preemptRevision }),
},
};

View file

@ -9,6 +9,7 @@ import type {
} from '@librechat/agents';
import type { SteerQueueItem } from '~/stream/interfaces/IJobStore';
import { GenerationJobManager } from '~/stream/GenerationJobManager';
import { getReferencedQuotes, mergeQuotedText } from '~/utils';
type SteerDrainOutput = HookOutputByEvent['PostToolBatch'];
@ -168,9 +169,14 @@ async function drainAndBuildInjections(opts: SteerDrainHookOptions): Promise<Inj
);
}
}
/** The media path already merged quotes into its text part; the plain
* path (no files, or a degraded encode) merges here so the excerpts
* reach the model exactly like `prependQuotes` on a normal turn. */
const quotes = getReferencedQuotes(item.quotes);
const textContent = quotes != null ? mergeQuotedText(item.text, quotes) : item.text;
injectedMessages.push({
role: 'user' as const,
content: (media?.content ?? item.text) as InjectedMessage['content'],
content: (media?.content ?? textContent) as InjectedMessage['content'],
source: 'steer' as const,
});
}

View file

@ -63,6 +63,34 @@ describe('submitted content adapters', () => {
});
});
it('inspects persisted steer-part quotes as quote fragments', () => {
// Steer parts persist their excerpts under `content[i].quotes`, mirroring
// the top-level `message.quotes` — import and share preflights must see
// them or blocked data could ride in on a quoted steer.
const message = {
role: 'assistant',
content: [
{
type: 'steer',
steer: 'about the selection',
steerId: 's1',
quotes: ['quoted secret excerpt'],
},
],
};
expect(fieldValues(extractStoredMessageContent(message))).toEqual(
expect.arrayContaining([
{
source: 'message',
field: 'quote',
text: 'quoted secret excerpt',
path: '/content/0/quotes/0',
},
]),
);
});
it('classifies persisted and imported summary content parts as message summaries', () => {
const message = {
content: [{ type: 'summary', text: 'persisted summary text' }],

View file

@ -187,6 +187,7 @@ export interface StoredMessagePartInput {
readonly original?: string;
readonly updated?: string;
readonly steer?: string;
readonly quotes?: readonly (string | { readonly text?: string } | null | undefined)[];
readonly error?: string;
readonly image_url?: string | { readonly url?: string };
readonly video_url?: { readonly url?: string };
@ -368,6 +369,7 @@ const STORED_MESSAGE_HANDLED_PART_PATH_SUFFIXES = new Set([
'/original',
'/updated',
'/steer',
'/quotes',
'/error',
'/image_url',
'/video_url',
@ -1564,7 +1566,30 @@ function extractStoredMessageContentWithBudget(
part?.files,
STORED_MESSAGE_ATTACHMENT_ARRAY_SCOPES,
);
/** Steer parts persist their quoted excerpts under `quotes`, exactly
* like the top-level `message.quotes`: model-bound user text that
* import and share preflights must inspect as quote fragments. */
const partQuotes = captureBoundedArray<string | { readonly text?: string }>(
part?.quotes,
STORED_MESSAGE_QUOTE_ARRAY_SCOPES,
);
const toolCall = part?.tool_call;
withReservedTraversalWork(
hasArrayValues(nestedContent) + hasArrayValues(partFiles) + (toolCall != null ? 1 : 0),
() =>
visitBoundedArray<string | { readonly text?: string }>(
partQuotes,
STORED_MESSAGE_QUOTE_ARRAY_SCOPES,
(quote, quoteIndex) => {
pushString(fragments, typeof quote === 'string' ? quote : quote?.text, {
id: `stored-message.content.${index}.quote.${quoteIndex}`,
path: `/content/${index}/quotes/${quoteIndex}`,
source: 'message',
field: 'quote',
});
},
),
);
withReservedTraversalWork(hasArrayValues(partFiles) + (toolCall != null ? 1 : 0), () =>
visitBoundedArray<{
readonly text?: string | { readonly value?: string };

View file

@ -348,6 +348,18 @@ export class ApprovalLifecycle {
await this.expire(streamId, expectedActionId ?? job.pendingAction.actionId, job.createdAt);
return false;
}
/** Translate the resuming owner's transient quote-capability assertion
* into its execution-bound marker (see `steerQuotesExecutionId`). A
* legacy resumer never reaches this code its execution rewrite alone
* invalidates the previous owner's marker. */
const { steerQuotesCapable, ...ownerPatch } = resumePatch ?? {};
const boundPatch = {
...ownerPatch,
...(steerQuotesCapable === true &&
typeof ownerPatch.providerExecutionId === 'string' && {
steerQuotesExecutionId: ownerPatch.providerExecutionId,
}),
};
const resumed = await this.store.transitionStatus(streamId, {
from: 'requires_action',
to: 'running',
@ -357,7 +369,7 @@ export class ApprovalLifecycle {
/** Ownership can move across replicas on resume. Owner-specific fields
* must change in this SAME CAS: once status is `running`, steering
* routes are live and may atomically inspect them. */
patch: { lastActiveAt: Date.now(), ...resumePatch },
patch: { lastActiveAt: Date.now(), ...boundPatch },
expectActionId: expectedActionId,
expectCreatedAt: job.createdAt,
});

View file

@ -2035,8 +2035,15 @@ class GenerationJobManagerClass {
const tenantId = getTenantId();
const safeTenantId = tenantId && tenantId !== SYSTEM_TENANT_ID ? tenantId : undefined;
const creationAttemptId = randomUUID();
const sanitizedMetadata = sanitizeJobMetadata(options.initialMetadata ?? {});
/** Translate the transient capability assertion into its execution-bound
* marker: valid only while `providerExecutionId` still names this owner,
* so a legacy replica winning a later HITL resume (which rewrites the
* execution id without knowing this field) self-invalidates it. */
const { steerQuotesCapable, ...storedMetadata } = sanitizedMetadata;
const initialMetadata = {
...sanitizeJobMetadata(options.initialMetadata ?? {}),
...storedMetadata,
...(steerQuotesCapable === true && { steerQuotesExecutionId: creationAttemptId }),
providerExecutionId: creationAttemptId,
providerDrained: true,
};
@ -2566,6 +2573,9 @@ class GenerationJobManagerClass {
// Surface the owning replica's seal capability so the steer route can
// honour it instead of probing its own (possibly older) SDK.
preemptCapable: jobData.preemptCapable,
// Same owner-recorded pattern for quote handling, execution-bound so a
// legacy resume's execution rewrite invalidates a stale assertion.
steerQuotesExecutionId: jobData.steerQuotesExecutionId,
providerExecutionId: jobData.providerExecutionId,
providerDrained: jobData.providerDrained,
steersClosed: jobData.steersClosed,

View file

@ -1,4 +1,5 @@
import type { TFile, TPendingSteer } from 'librechat-data-provider';
import { getReferencedQuotes } from '~/utils';
/** Immutable user-visible payload a parked steer recovery is allowed to submit. */
export interface RecoveredSteerPayload {
@ -6,6 +7,11 @@ export interface RecoveredSteerPayload {
/** Sorted, unique file ids. Display metadata is deliberately excluded: the
* normal send path re-resolves files by owner and only identity is binding. */
fileIds: string[];
/** Normalized quoted excerpts, order-significant. Model-bound exactly like
* the text, so the proof must bind them too: a stale client presenting the
* same recoverySteerId with altered or missing quotes must not consume the
* parked source. Empty when the source carried none. */
quotes: string[];
}
/** A recovery-shaped request did not reproduce the parked source exactly. */
@ -47,12 +53,16 @@ export function canonicalRecoveryFileIds(files: unknown): string[] | null {
export function buildRecoveredSteerPayload(
text: unknown,
files: unknown,
quotes?: unknown,
): RecoveredSteerPayload | null {
if (typeof text !== 'string') {
return null;
}
const fileIds = canonicalRecoveryFileIds(files);
return fileIds == null ? null : { text, fileIds };
if (fileIds == null) {
return null;
}
return { text, fileIds, quotes: getReferencedQuotes(quotes) ?? [] };
}
export function isRecoveredSteerPayload(value: unknown): value is RecoveredSteerPayload {
@ -65,19 +75,24 @@ export function isRecoveredSteerPayload(value: unknown): value is RecoveredSteer
Array.isArray(payload.fileIds) &&
payload.fileIds.every((id) => typeof id === 'string' && id.length > 0) &&
payload.fileIds.length === new Set(payload.fileIds).size &&
payload.fileIds.every((id, index) => index === 0 || payload.fileIds![index - 1] < id)
payload.fileIds.every((id, index) => index === 0 || payload.fileIds![index - 1] < id) &&
Array.isArray(payload.quotes) &&
payload.quotes.every((quote) => typeof quote === 'string' && quote.length > 0)
);
}
export function recoveredSteerPayloadMatches(
item: Pick<TPendingSteer, 'text' | 'files'>,
item: Pick<TPendingSteer, 'text' | 'files' | 'quotes'>,
expected: RecoveredSteerPayload,
): boolean {
const fileIds = canonicalRecoveryFileIds(item.files);
const itemQuotes = getReferencedQuotes(item.quotes) ?? [];
return (
item.text === expected.text &&
fileIds != null &&
fileIds.length === expected.fileIds.length &&
fileIds.every((id, index) => id === expected.fileIds[index])
fileIds.every((id, index) => id === expected.fileIds[index]) &&
itemQuotes.length === expected.quotes.length &&
itemQuotes.every((quote, index) => quote === expected.quotes[index])
);
}

View file

@ -22,6 +22,7 @@ export function toPendingSteer(item: SteerQueueItem): TPendingSteer {
text: item.text,
createdAt: item.createdAt,
...(item.files && item.files.length > 0 && { files: item.files }),
...(item.quotes && item.quotes.length > 0 && { quotes: item.quotes }),
...(item.preempt === true && { preempt: true }),
...(item.preemptRevision != null && { preemptRevision: item.preemptRevision }),
};

View file

@ -348,6 +348,7 @@ describe('RedisJobStore', () => {
promptTokens: 0,
discoveredTools: [],
preemptCapable: true,
steerQuotesExecutionId: 'exec-1',
generationProtocolVersion: 2,
resolvedAskUserQuestions: [
{
@ -370,6 +371,7 @@ describe('RedisJobStore', () => {
* degrading to ordinary steering in every Redis deployment.
*/
expect(job.preemptCapable).toBe(true);
expect(job.steerQuotesExecutionId).toBe('exec-1');
expect(job.generationProtocolVersion).toBe(2);
expect(job.checkpointNamespace).toBe(String(job.createdAt));
expect(job.resolvedAskUserQuestions).toEqual([

View file

@ -3118,7 +3118,7 @@ describe('RedisJobStore Integration Tests', () => {
undefined,
undefined,
undefined,
{ text: 'kept', fileIds: [] },
{ text: 'kept', fileIds: [], quotes: [] },
);
expect(await store.claimParkedSteers(streamId, 'steer-user')).toBeUndefined();
expect(await ioredisClient.exists(`stream:{${streamId}}:parked`)).toBe(1);
@ -3145,7 +3145,7 @@ describe('RedisJobStore Integration Tests', () => {
undefined,
undefined,
undefined,
{ text: 'kept', fileIds: [] },
{ text: 'kept', fileIds: [], quotes: [] },
);
expect(
await store.consumeParkedSteer(
@ -3163,8 +3163,12 @@ describe('RedisJobStore Integration Tests', () => {
});
test.each([
['changed text', { text: 'forged words', fileIds: ['file-a', 'file-b'] }],
['changed files', { text: 'original words', fileIds: ['file-a', 'file-c'] }],
['changed text', { text: 'forged words', fileIds: ['file-a', 'file-b'], quotes: [] }],
['changed files', { text: 'original words', fileIds: ['file-a', 'file-c'], quotes: [] }],
[
'changed quotes',
{ text: 'original words', fileIds: ['file-a', 'file-b'], quotes: ['forged excerpt'] },
],
])('atomically refuses parked recovery with %s', async (_label, proof) => {
if (!ioredisClient) {
return;
@ -4177,6 +4181,7 @@ describe('RedisJobStore Integration Tests', () => {
{
text: item.text,
fileIds: (item.files ?? []).flatMap((file) => file.file_id ?? []).sort(),
quotes: item.quotes ?? [],
},
);
@ -4267,6 +4272,7 @@ describe('RedisJobStore Integration Tests', () => {
{
text: item.text,
fileIds: (item.files ?? []).flatMap((file) => file.file_id ?? []).sort(),
quotes: item.quotes ?? [],
},
);
await expect(

View file

@ -110,7 +110,7 @@ describe('generation protocol rollout storage', () => {
undefined,
undefined,
undefined,
{ text: leased.text, fileIds: [] },
{ text: leased.text, fileIds: [], quotes: [] },
);
const downgraded = await store.claimParkedSteersDetailed(streamId, 'user-1', undefined, 1);
@ -189,7 +189,7 @@ describe('generation protocol rollout storage', () => {
undefined,
undefined,
undefined,
{ text: 'legacy words', fileIds: [] },
{ text: 'legacy words', fileIds: [], quotes: [] },
),
).rejects.toMatchObject({ code: 'RECOVERY_PAYLOAD_MISMATCH' });
});

View file

@ -135,7 +135,7 @@ describe('Redis generation protocol rollout bridge', () => {
undefined,
undefined,
undefined,
{ text: leased.text, fileIds: [] },
{ text: leased.text, fileIds: [], quotes: [] },
);
const downgraded = await store.claimParkedSteersDetailed(streamId, 'user-1', undefined, 1);
@ -191,7 +191,7 @@ describe('Redis generation protocol rollout bridge', () => {
undefined,
undefined,
undefined,
{ text: 'legacy words', fileIds: [] },
{ text: 'legacy words', fileIds: [], quotes: [] },
),
).rejects.toMatchObject({ code: 'RECOVERY_PAYLOAD_MISMATCH' });
});

View file

@ -158,6 +158,7 @@ describe('InMemoryJobStore steer receipt integrity', () => {
{
text: item.text,
fileIds: (item.files ?? []).flatMap((file) => file.file_id ?? []).sort(),
quotes: item.quotes ?? [],
},
);
@ -209,6 +210,7 @@ describe('InMemoryJobStore steer receipt integrity', () => {
{
text: item.text,
fileIds: (item.files ?? []).flatMap((file) => file.file_id ?? []).sort(),
quotes: item.quotes ?? [],
},
);

View file

@ -51,6 +51,17 @@ describe('SteeringLifecycle via GenerationJobManager.steering (in-memory)', () =
};
}
describe('toPendingSteer', () => {
test('keeps quotes in the client-safe projection while dropping userId', () => {
const projected = toPendingSteer({
...buildSteer('with context'),
quotes: ['the excerpt'],
});
expect(projected.quotes).toEqual(['the excerpt']);
expect(projected).not.toHaveProperty('userId');
});
});
describe('enqueue', () => {
test('appends to a running job and returns the queue depth', async () => {
const streamId = 'steer-enqueue';
@ -286,6 +297,103 @@ describe('SteeringLifecycle via GenerationJobManager.steering (in-memory)', () =
});
});
describe('execution-bound quote capability', () => {
function pauseAction(streamId: string) {
const payload = buildToolApprovalPayload([
{ name: 'shell', arguments: { command: 'ls' }, tool_call_id: 'call_qc' },
]);
return buildPendingAction(payload, {
streamId,
conversationId: streamId,
runId: 'run-qc',
responseMessageId: 'msg-qc',
});
}
test('a capable resume re-binds the marker to its own execution', async () => {
const streamId = 'steer-quote-capability-resume';
const job = await manager.createJob(streamId, 'user-1', undefined, {
initialMetadata: { steerQuotesCapable: true },
});
expect(await manager.approvals.pause(streamId, pauseAction(streamId))).toBe(true);
expect(
await manager.approvals.resolve(
streamId,
undefined,
{ steerQuotesCapable: true, providerExecutionId: 'resumed-exec', providerDrained: true },
job.createdAt,
),
).toBe(true);
const resumed = await manager.getJob(streamId);
expect(resumed?.metadata.steerQuotesExecutionId).toBe('resumed-exec');
const depth = await manager.steering.enqueue(streamId, {
...buildSteer('quoted after resume'),
quotes: ['kept excerpt'],
});
expect(depth).toBe(1);
const [queued] = await manager.steering.peek(streamId);
expect(queued.quotes).toEqual(['kept excerpt']);
});
test('a legacy resume (no assertion) invalidates the previous marker atomically', async () => {
const streamId = 'steer-quote-capability-legacy-resume';
const job = await manager.createJob(streamId, 'user-1', undefined, {
initialMetadata: { steerQuotesCapable: true },
});
expect(await manager.approvals.pause(streamId, pauseAction(streamId))).toBe(true);
// A pre-quotes replica's patch rewrites the execution id but cannot
// know the marker field — exactly the omit-not-clear shape.
expect(
await manager.approvals.resolve(
streamId,
undefined,
{ providerExecutionId: 'legacy-exec', providerDrained: true },
job.createdAt,
),
).toBe(true);
await manager.steering.enqueue(streamId, {
...buildSteer('quoted after legacy resume'),
quotes: ['dropped excerpt'],
});
const [queued] = await manager.steering.peek(streamId);
expect(queued).not.toHaveProperty('quotes');
});
});
describe('recovered-steer payload proof', () => {
const { buildRecoveredSteerPayload, recoveredSteerPayloadMatches } =
jest.requireActual<typeof import('~/stream/SteerRecovery')>('~/stream/SteerRecovery');
test('binds normalized quotes into the proof (empty when none)', () => {
expect(buildRecoveredSteerPayload('words', undefined)).toEqual({
text: 'words',
fileIds: [],
quotes: [],
});
expect(buildRecoveredSteerPayload('words', undefined, [' kept ', ''])).toEqual({
text: 'words',
fileIds: [],
quotes: ['kept'],
});
});
test('a recovery matches only when the quotes match, order-significant', () => {
const item = { text: 'words', quotes: ['first', 'second'] };
const proofFor = (quotes?: unknown) => buildRecoveredSteerPayload('words', undefined, quotes);
expect(recoveredSteerPayloadMatches(item, proofFor(['first', 'second'])!)).toBe(true);
expect(recoveredSteerPayloadMatches(item, proofFor(['second', 'first'])!)).toBe(false);
expect(recoveredSteerPayloadMatches(item, proofFor(['first'])!)).toBe(false);
// A stale client omitting the quotes must not consume the parked source.
expect(recoveredSteerPayloadMatches(item, proofFor(undefined)!)).toBe(false);
// Quote-less sources keep matching quote-less recoveries (pre-quotes parity).
expect(recoveredSteerPayloadMatches({ text: 'words' }, proofFor(undefined)!)).toBe(true);
});
});
describe('park / claim (no-subscriber recovery)', () => {
const owner = { userId: 'user-1' };
@ -430,7 +538,7 @@ describe('SteeringLifecycle via GenerationJobManager.steering (in-memory)', () =
const failedRecovery = await manager.createJob(streamId, 'user-1', undefined, {
recoveredSteerId: 'p3',
recoveredSteerPayload: { text: 'stale', fileIds: [] },
recoveredSteerPayload: { text: 'stale', fileIds: [], quotes: [] },
});
expect(await manager.steering.claim(streamId, owner)).toEqual([]);
@ -443,7 +551,7 @@ describe('SteeringLifecycle via GenerationJobManager.steering (in-memory)', () =
const persistedRecovery = await manager.createJob(streamId, 'user-1', undefined, {
recoveredSteerId: 'p3',
recoveredSteerPayload: { text: 'stale', fileIds: [] },
recoveredSteerPayload: { text: 'stale', fileIds: [], quotes: [] },
});
expect(
await manager.steering.consumeRecovered(streamId, 'p3', owner, persistedRecovery.createdAt),
@ -452,8 +560,8 @@ describe('SteeringLifecycle via GenerationJobManager.steering (in-memory)', () =
});
test.each([
['changed text', { text: 'forged words', fileIds: ['file-a', 'file-b'] }],
['changed files', { text: 'original words', fileIds: ['file-a', 'file-c'] }],
['changed text', { text: 'forged words', fileIds: ['file-a', 'file-b'], quotes: [] }],
['changed files', { text: 'original words', fileIds: ['file-a', 'file-c'], quotes: [] }],
])(
'refuses recovery with %s without leasing or consuming the source',
async (_label, proof) => {

View file

@ -1585,6 +1585,12 @@ export class InMemoryJobStore implements IJobStoreV2 {
...(job.preemptCapable === true && { preempt: true }),
}),
};
if (
persisted.quotes != null &&
(job.steerQuotesExecutionId == null || job.steerQuotesExecutionId !== job.providerExecutionId)
) {
delete persisted.quotes;
}
queue.push(persisted);
return { item: { ...persisted }, position: queue.length };
}
@ -1703,6 +1709,12 @@ export class InMemoryJobStore implements IJobStoreV2 {
...(job.preemptCapable === true && { preempt: true }),
}),
};
if (
persisted.quotes != null &&
(job.steerQuotesExecutionId == null || job.steerQuotesExecutionId !== job.providerExecutionId)
) {
delete persisted.quotes;
}
queue.push(persisted);
const receipt: SteerReceipt = {
...receiptInput,

View file

@ -203,6 +203,7 @@ const JOB_CAS_LUA =
'local clientItem = { steerId = item.steerId, text = item.text, createdAt = item.createdAt } ' +
'if item.clientSteerId then clientItem.clientSteerId = item.clientSteerId end ' +
'if item.files then clientItem.files = item.files end ' +
'if item.quotes then clientItem.quotes = item.quotes end ' +
'if item.preempt then clientItem.preempt = item.preempt end ' +
'if item.preemptRevision then clientItem.preemptRevision = item.preemptRevision end ' +
'projected[#projected + 1] = clientItem ' +
@ -460,10 +461,20 @@ const JOB_CREATE_LUA =
'local expectedSeen = {} for i = 1, #decoded.fileIds do local fileId = decoded.fileIds[i] ' +
'if type(fileId) ~= "string" or fileId == "" or expectedSeen[fileId] then ' +
'return { "", "", "0", "recovery_payload_mismatch" } end expectedSeen[fileId] = true end ' +
'if decoded.quotes ~= nil then if not isDenseArray(decoded.quotes) then ' +
'return { "", "", "0", "recovery_payload_mismatch" } end ' +
'for i = 1, #decoded.quotes do if type(decoded.quotes[i]) ~= "string" or decoded.quotes[i] == "" then ' +
'return { "", "", "0", "recovery_payload_mismatch" } end end end ' +
'expectedRecovery = decoded elseif ARGV[9] ~= "" then ' +
'return { "", "", "0", "recovery_payload_mismatch" } end ' +
'local function recoveryMatches(item, expected) ' +
'if not expected or type(item.text) ~= "string" or item.text ~= expected.text then return false end ' +
// Quotes are model-bound like the text: order-significant identity, with a
// missing array on either side reading as empty (pre-quotes compatibility).
'local expectedQuotes = expected.quotes or {} local itemQuotes = item.quotes ' +
'if itemQuotes ~= nil and not isDenseArray(itemQuotes) then return false end ' +
'itemQuotes = itemQuotes or {} if #itemQuotes ~= #expectedQuotes then return false end ' +
'for i = 1, #itemQuotes do if itemQuotes[i] ~= expectedQuotes[i] then return false end end ' +
'local actualSeen = {} local actualCount = 0 local files = item.files ' +
'if files then if not isDenseArray(files) then return false end ' +
'for i = 1, #files do local file = files[i] ' +
@ -505,7 +516,8 @@ const JOB_CREATE_LUA =
'if ok and item.steerId and not seen[item.steerId] then seen[item.steerId] = true ' +
'local projected = { steerId = item.steerId, text = item.text, createdAt = item.createdAt } ' +
'if item.clientSteerId then projected.clientSteerId = item.clientSteerId end ' +
'if item.files then projected.files = item.files end if item.preempt then projected.preempt = item.preempt end ' +
'if item.files then projected.files = item.files end if item.quotes then projected.quotes = item.quotes end ' +
'if item.preempt then projected.preempt = item.preempt end ' +
'if item.preemptRevision then projected.preemptRevision = item.preemptRevision end ' +
'merged[#merged + 1] = projected receiptUpdates[#receiptUpdates + 1] = item end end end end ' +
'local recoveryOwnerMatches = parkedUserId == ARGV[6] and ' +
@ -682,7 +694,8 @@ const STALE_JOB_DELETE_LUA =
'if item.steerId and not seen[item.steerId] then seen[item.steerId] = true fullItems[#fullItems + 1] = item ' +
'local clientItem = { steerId = item.steerId, text = item.text, createdAt = item.createdAt } ' +
'if item.clientSteerId then clientItem.clientSteerId = item.clientSteerId end ' +
'if item.files then clientItem.files = item.files end if item.preempt then clientItem.preempt = item.preempt end ' +
'if item.files then clientItem.files = item.files end if item.quotes then clientItem.quotes = item.quotes end ' +
'if item.preempt then clientItem.preempt = item.preempt end ' +
'if item.preemptRevision then clientItem.preemptRevision = item.preemptRevision end ' +
'projected[#projected + 1] = clientItem end ' +
'if generationProtocol == 2 and item.clientSteerId then local raw = redis.call("HGET", KEYS[8], item.clientSteerId) ' +
@ -932,6 +945,7 @@ const STEER_ENQUEUE_VERSIONED_LUA =
'if redis.call("HGET", KEYS[1], "steersClosed") == "1" then return -1 end ' +
'if redis.call("LLEN", KEYS[2]) >= tonumber(ARGV[3]) then return -2 end ' +
'local item = cjson.decode(ARGV[1]) ' +
'if item.quotes then local qexec = redis.call("HGET", KEYS[1], "steerQuotesExecutionId") if not qexec or qexec == "" or qexec ~= redis.call("HGET", KEYS[1], "providerExecutionId") then item.quotes = nil end end ' +
'if ARGV[5] == "1" then item.preemptRevision = 1 ' +
'if redis.call("HGET", KEYS[1], "preemptCapable") == "1" then item.preempt = true end end ' +
'local itemJson = cjson.encode(item) ' +
@ -969,7 +983,10 @@ const STEER_ENQUEUE_RECEIPT_LUA =
'if redis.call("HGET", KEYS[1], "status") ~= "running" then return -1 end ' +
'if redis.call("HGET", KEYS[1], "steersClosed") == "1" then return -1 end ' +
'if redis.call("LLEN", KEYS[2]) >= tonumber(ARGV[3]) then return -2 end ' +
'local legacyItem = cjson.decode(ARGV[1]) if ARGV[7] == "1" then legacyItem.preemptRevision = 1 ' +
'local legacyItem = cjson.decode(ARGV[1]) ' +
'if legacyItem.quotes then local qexec = redis.call("HGET", KEYS[1], "steerQuotesExecutionId") ' +
'if not qexec or qexec == "" or qexec ~= redis.call("HGET", KEYS[1], "providerExecutionId") then legacyItem.quotes = nil end end ' +
'if ARGV[7] == "1" then legacyItem.preemptRevision = 1 ' +
'if redis.call("HGET", KEYS[1], "preemptCapable") == "1" then legacyItem.preempt = true end end ' +
'redis.call("RPUSH", KEYS[2], cjson.encode(legacyItem)) ' +
'redis.call("EXPIRE", KEYS[2], tonumber(ARGV[2])) ' +
@ -980,6 +997,7 @@ const STEER_ENQUEUE_RECEIPT_LUA =
'if redis.call("LLEN", KEYS[2]) >= tonumber(ARGV[3]) then return -2 end ' +
'if redis.call("ZCARD", KEYS[4]) >= tonumber(ARGV[9]) then return -3 end ' +
'local item = cjson.decode(ARGV[1]) ' +
'if item.quotes then local qexec = redis.call("HGET", KEYS[1], "steerQuotesExecutionId") if not qexec or qexec == "" or qexec ~= redis.call("HGET", KEYS[1], "providerExecutionId") then item.quotes = nil end end ' +
'if ARGV[7] == "1" then ' +
'item.preemptRevision = 1 ' +
'if redis.call("HGET", KEYS[1], "preemptCapable") == "1" then item.preempt = true end ' +
@ -1475,6 +1493,7 @@ const STEER_CLOSE_DRAIN_LUA =
'local projected = { steerId = item.steerId, text = item.text, createdAt = item.createdAt } ' +
'if item.clientSteerId then projected.clientSteerId = item.clientSteerId end ' +
'if item.files then projected.files = item.files end ' +
'if item.quotes then projected.quotes = item.quotes end ' +
'if item.preempt then projected.preempt = item.preempt end ' +
'if item.preemptRevision then projected.preemptRevision = item.preemptRevision end ' +
'currentProjected[#currentProjected + 1] = projected end ' +
@ -4602,6 +4621,10 @@ export class RedisJobStore implements IJobStoreV2 {
* `preemptArmed: false` and silently degrade interrupt-steer to
* tool-boundary steering in EVERY Redis deployment. */
preemptCapable: data.preemptCapable != null ? data.preemptCapable === '1' : undefined,
/** Same explicit-mapper trap as `preemptCapable`: without this line every
* Redis read reports the owner quote-incapable, so admission would drop
* all steer quotes (and their echo) in EVERY Redis deployment. */
steerQuotesExecutionId: data.steerQuotesExecutionId || undefined,
providerAbortReady:
data.providerAbortReady != null ? data.providerAbortReady === '1' : undefined,
providerExecutionId: data.providerExecutionId || undefined,

View file

@ -167,6 +167,24 @@ export interface SerializableJobData {
* preempt shipped, which reads as incapable: the honest outcome.
*/
preemptCapable?: boolean;
/**
* Transient owner assertion that this replica's drain merges
* `SteerQueueItem.quotes` into the injected turn. Never stored as-is:
* createJob and `ApprovalLifecycle.resolve` translate it into
* `steerQuotesExecutionId` bound to the asserting owner's execution.
*/
steerQuotesCapable?: boolean;
/**
* The `providerExecutionId` of the owner that asserted quote capability.
* Valid only while it equals the LIVE `providerExecutionId`: a legacy
* replica winning a HITL resume rewrites the execution id but cannot know
* this field, so its stale assertion self-invalidates which a bare
* boolean could not do (an old resume patch omits rather than clears it).
* The fenced enqueue evaluates the equality atomically and strips
* `item.quotes` on mismatch, keeping the persisted item and the
* `quotesAccepted` echo honest; the client re-stages dropped excerpts.
*/
steerQuotesExecutionId?: string;
/** Explicitly false until the provider-owning replica has installed its
* generation-fenced abort subscription. Missing is conservative legacy
@ -383,6 +401,8 @@ export type JobMetadataPatch = Partial<
| 'discoveredTools'
| 'activityPhaseSnapshot'
| 'preemptCapable'
| 'steerQuotesCapable'
| 'steerQuotesExecutionId'
| 'providerExecutionId'
| 'providerDrained'
| 'generationProtocolVersion'
@ -427,6 +447,10 @@ export interface SteerQueueItem {
* drain re-fetches each file by id scoped to the run's user and encodes
* fresh, so nothing here is trusted beyond identifying the file. */
files?: Partial<TFile>[];
/** Quoted excerpts steered with the message, normalized at admission
* (`getReferencedQuotes`). Kept separate from `text` so the persisted
* steer part stays clean; merged into the model-bound turn at injection. */
quotes?: string[];
/** The steer asked to seal the live model stream at the next provider-safe
* boundary instead of waiting for a tool step. Durable so a parked,
* claimed, or replayed chip keeps its "interrupting" label. */
@ -442,7 +466,16 @@ export interface SteerQueueItem {
* the same instruction twice after drain, terminal cleanup, or replacement. */
export interface SteerReceipt {
clientSteerId: string;
/** Quote-INDEPENDENT content hash (text/files/preempt) the one shape every
* replica version computes, so lost-ACK retries replay across a rolling
* deploy in both directions. */
fingerprint: string;
/** Identity of the REQUESTED quotes (pre any owner-capability strip),
* recorded beside the fingerprint so quote-aware readers enforce quote
* identity without making the fingerprint unreadable to legacy admission.
* Absent on receipts written by pre-quotes replicas or for quote-less
* requests. */
requestedQuotesFingerprint?: string;
userId: string;
tenantId?: string;
agentId?: string;

View file

@ -63,6 +63,9 @@ export function sanitizeJobMetadata(metadata: Partial<GenerationJobMetadata>): J
if (metadata.preemptCapable !== undefined) {
patch.preemptCapable = metadata.preemptCapable;
}
if (metadata.steerQuotesCapable !== undefined) {
patch.steerQuotesCapable = metadata.steerQuotesCapable;
}
if (metadata.generationProtocolVersion === 1 || metadata.generationProtocolVersion === 2) {
patch.generationProtocolVersion = metadata.generationProtocolVersion;
}

View file

@ -65,6 +65,10 @@ export interface GenerationJobMetadata {
activityPhaseSnapshot?: ActivityPhaseSnapshot;
/** See `SerializableJobData.preemptCapable`. */
preemptCapable?: boolean;
/** See `SerializableJobData.steerQuotesCapable`. */
steerQuotesCapable?: boolean;
/** See `SerializableJobData.steerQuotesExecutionId`. */
steerQuotesExecutionId?: string;
/** Exact provider segment whose completion gates destructive user cleanup. */
providerExecutionId?: string;
/** False only while that exact provider segment can still mutate user data. */

View file

@ -687,6 +687,10 @@ export type SteerContentPart = {
/** Attachments steered with the message; re-encoded per turn on replay
* like any other user-message media (refs only, never encoded data). */
files?: Partial<TFile>[];
/** Quoted excerpts steered with the message, persisted separately from the
* typed text (mirroring `TMessage.quotes`) so the UI renders them as
* reference blocks; merged into the model-bound user turn on every replay. */
quotes?: string[];
};
export type TMessageContentParts =

View file

@ -201,6 +201,9 @@ export type TPendingSteer = {
text: string;
createdAt?: number;
files?: Partial<TFile>[];
/** Quoted excerpts steered with the message ("Add to chat" selections);
* merged into the model-bound text at the injection boundary. */
quotes?: string[];
/** The steer asked to interrupt generation at the next safe boundary
* kept on parked/replayed chips so the "interrupting" label survives. */
preempt?: boolean;
@ -222,6 +225,9 @@ export type TSteerAppliedEvent = {
clientSteerId?: string;
createdAt?: number;
files?: Partial<TFile>[];
/** Quoted excerpts steered with the message (mirrors `SteerContentPart`,
* which cannot be imported here without a module cycle). */
quotes?: string[];
};
responseMessageId?: string;
conversationId?: string;