🧭 feat: Mid-Run Steering and Queued Messages for Agent Runs (#14220)

* 🧭 feat: Mid-Run Steering and Queued Messages for Agent Runs Steering: submit a message while a run is generating; the server queues it in the job store (cross-instance) and a run-scoped PostToolBatch hook injects it into graph state at the next tool-batch boundary, records an inline 'steer' content part on the response (replayed as a user message on later turns), and streams on_steer_applied to the client. Queuing: messages composed during a run auto-send as normal follow-up turns after clean completion (one per final event, FIFO); user aborts leave them as chips unless armed by interrupt-and-send. Requires hook injectedMessages support in @librechat/agents (danny-avila/agents#299); hard-gated via a capability probe so older SDKs 501 the steer route instead of draining and dropping messages. * 🧵 fix: Harden Steering Against Finalization Races and Route Guard Gaps Addresses local Codex review findings on the steering feature: - Close-and-drain the steer queue atomically at finalization (final event, abort) so a steer POST racing teardown is rejected instead of 202-ACKed and then silently cleared; the closed flag lives on the job hash and is reset when a replacement job reuses the stream id. - Clear inherited steer queues on createJob — a job replacement must not drain the replaced run's messages. - Keep steers queued across a HITL pause instead of draining them into ephemeral client state: resumeState re-seeds chips on reload and the resumed run injects them at its first tool boundary (steers key TTL now extends to the approval window; on_steers_pending event removed). - Queue the NO_ACTIVE_RUN steer fallback while the final SSE is still settling — a direct send would be dropped by ask()'s in-flight guard. - Reconcile the 202 ACK against on_steer_applied events that beat it over the SSE, so a chip can't be re-minted after its removal event passed. - Allow the per-send Steer override when the default action is queue. - Apply the configured message rate limiters and the PII filter to POST /chat/steer — a steer is model-bound user text. * ✅ ci: Assert Steering Capability Probe Against the Installed SDK CI installs the published @librechat/agents pin (pre-injectedMessages), where isSteeringSupported() is legitimately false — the probe test now asserts it mirrors the installed SDK's capability flag instead of hardcoding the capability-bearing build's value. Verified against both the published 3.2.61 dist and the agents#299 build. * 🛟 fix: Preserve Steer Text Across Run-End, Error, and Abort Races Codex round 2 (4 P2s): - Applied-steer-id set survives run end (capped at 100) and converted ids join it, so a 202 ACK that lands after final/abort drops its chip instead of re-minting a stranded pending one. - Failed runs no longer strand acknowledged chips: both error paths convert local pending chips to queued follow-ups (chip text is client-local), and the server closes the steer queue before emitting the error so a racing steer POST gets 404 fallback instead of a 202 whose payload dies with the job. - sendQueuedNow keys on steer availability, not the default action — send-now on a queued chip is an explicit override for queue-preferring users. - Stop path consumes pendingSteers from the abort HTTP response as a fallback for the SSE final event it may close before processing; conversion is deduped so double delivery is a no-op (shared useSteerConvert hook). * 📎 feat: Carry Attachments Through During-Run Queued Messages Steering stays text-only (SDK injection, inline STEER part, and replay are all text), so a during-run submit with media now queues the whole message as one unit instead of silently stranding the files: - QueuedMessage gains `files`; composer attachments are consumed into the queued item at queue time (steerFromComposer / queueFromComposer / interruptAndSend), fixing the latent hazard where lingering composer files glued onto whatever `ask` vacuumed up next. - Enter-steer with attachments degrades to queue with an explanatory toast; the per-send menu routes through the same composer-aware wrappers. - The drain and sendQueuedNow pass the item's files as `overrideFiles`; media items never steer (send as a normal turn when idle, re-front otherwise). ask() no longer clears composer state for caller-supplied overrideFiles — only regenerate keeps that behavior. - During-run submits hold while uploads are in flight, mirroring the send button's filesLoading gate; queued chips show a paperclip count. * 🎛️ feat: Rework During-Run Chips into Action Rows Full-width rows above the composer (reference-UI parity): each queued message shows a primary Steer/Send-now action, delete, and a "…" menu with Edit message (restores text + attachments into the composer) and a Turn on queueing/steering toggle that flips the Enter default. Steer rows share the layout with status text; failed steers keep retry / edit / queue-convert. The per-send menu gains the same default toggle. Queued file refs now retain filename + bytes so edit-restore rebuilds real composer entries (draft-recovery shape). * 🖇️ feat: Steer With Attachments (Multimodal Mid-Run Injection) Steering now carries media end-to-end instead of degrading to queue: - The steer POST accepts sanitized attachment refs (cap 10; only file_id is trusted — the drain re-fetches owner-scoped and re-derives everything else). SteerQueueItem/TPendingSteer/SteerContentPart carry `files` refs; encoded data is never persisted or queued. - New api/server/services/Files/steering.js decouples attachment building from the request path: encodeSteerContent reuses the exact per-turn pipeline (addFileContextToMessage + processAttachments' single-pass categorize/encode, SDK formatMessage assembly, prependFileContext for extracted text) with zero new encoding code. buildSteerMedia feeds the drain hook's new buildMedia seam (any failure degrades that steer to text-only — words always land); stampSteerPartMedia re-encodes past steer parts per turn with ONE batched owner-scoped fetch and stamps a transient `media` array, replaced immutably so it can never leak into a save. Replay honors resendFiles like regular message media. - The SDK's formatAgentMessages (the formatter agents actually use) gained the steer replay branch on the PR branch; the local formatMessages.js branch now mirrors the media preference. - Client: steerFromComposer consumes composer files into the POST, chips/seeding/conversions carry files everywhere (retry, queue convert, abort/error recovery), queued media items steer for real, and SteerBubble renders the steered attachments inline. * 🧵 fix: Harden Steer Recovery Races and Drain Isolation Codex round 3 (7 fixes): - A 202 ACK landing after the run ended converts straight to a queued follow-up (server queue is gone; no event will ever resolve a pending chip for a finished run). Covers stream errors with in-flight POSTs. - A Stop that lands pre-completion can arrive as a final with unfinished:true and no aborted flag — runEnd now treats it as aborted so queued messages are not auto-sent against the user's Stop. - Leftover-steer conversion merges chronologically by createdAt instead of appending, preserving the order the user composed. - Auto-drained queued messages pass explicit (possibly empty) overrideFiles/overrideQuotes/overrideManualSkills: a drain can no longer vacuum up files, quotes, or skill picks staged in the composer for the user's NEXT message (ask() treats overrideFiles != null as authoritative). - Failed-steer Retry and resume-on-load chip restoration keep the steer's attachments. - The job-replacement guard moved INSIDE the store's atomic drain/close-and-drain (Lua createdAt compare; in-memory equivalent): a stale run's hook or finalization can neither consume, close, nor steal a replacement job's steer queue, and the drain hook drops its separate check-then-drain round trip. * 🧰 refactor: Typed Steer Controller, Single-Query Media Pass, Round-4 Fixes Codex round 4 + efficiency tightening in one pass: - Moved the steer guard ladder (validation, file sanitization via a shared toSteerFileRef picker, ownership/tenant checks, status-guarded enqueue) into packages/api as handleSteerRequest; api/steer.js is now a thin wrapper. Ladder covered against the REAL in-memory job manager in request.spec.ts; the api spec pins only the wrapper contract. - Folded the steer replay stamp into the turn's ONE historical-files query: collectHistoricalFileRefs also gathers steer-part refs, the owner-scoped doc map rides client state, and stampSteerPartMedia consumes it (no second round trip) while encoding parts in parallel. - Stamped steer media now counts against the run budget (existing multimodal counter over the non-text parts, folded into indexTokenCountMap/promptTokens after the stamp). - Steer route runs the PII filter BEFORE moderateText, matching chat.js so blocked sensitive text never reaches the external moderation API. - Interrupt & send survives the abort-response-beats-SSE-final race: stopGenerating writes the run-end signal itself when the one-shot interrupt flag is armed and no signal landed (double-fire safe). - Resume reconciles chips against the server's still-queued list even when EMPTY, clearing chips for steers applied while disconnected. - The local formatter's steer flush preserves non-text assistant parts (array-content AIMessage) instead of folding to text. * 🔒 fix: Replay-Aware Capability Gate and Round-5 Race Closures - isSteeringSupported now requires BOTH halves of the SDK contract: injection (HOOK_INJECTED_MESSAGES_CAPABLE) AND replay (ContentTypes.STEER, shipped in the same SDK commit as the formatAgentMessages steer branch). An SDK that can inject but not replay 501s the steer route — no release window can create steer parts that would leak into provider-facing assistant content. - The local formatter mirrors the SDK's anchor reset: a post-steer tool_call mints a fresh AIMessage instead of attaching to the pre-steer anchor (invalid provider ordering). - Queued-chip send-now and the NO_ACTIVE_RUN fallback pass explicit (possibly empty) overrideFiles so an idle send can't vacuum composer files staged for a different draft. - Redis createJob deletes the stale steer list BEFORE the replacement hash is written as running — a steer 202-accepted against the new job can never be wiped by the reset. - Resumed-turn finalization mirrors the normal path's terminal drain: createdAt-guarded close-and-drain, leftovers ride the resumed final event as pendingSteers instead of being cleared by completeJob. - buildSteerMedia restores composer order over the $in result so multi-attachment steers reach the model in the order the user saw. * ⚛️ fix: Atomic Job Replacement and Boundary-Clean Steering Module Codex round 6 (5 fixed, 1 standing deferral): - createJob resets the steer queue and writes the job hash in ONE same-slot Lua script (JOB_CREATE_LUA): a steer POST can no longer interleave between them on cluster, so a steer accepted against one run can never be drained into another. Redis-validated. - The steering media pipeline moved to packages/api (agents/steering/media.ts) with injected getFiles and a structural client interface — /api keeps zero steering logic; specs ported to the DI seam. - handleSteerRequest checks the job BEFORE the capability gate: a steer racing completion on an unsupported SDK gets 404 (send-now) instead of a 501 queue with no run-end signal left to drain it. - useQueueDrain binds to the active conversation: navigating away between the final SSE and the drain effect leaves the signal unconsumed instead of submitting A's follow-up into B; the drain fires on return. - abortJob closes and drains the steer queue BEFORE the content snapshot, so a drain-hook apply that lands pre-drain is captured inline rather than lost between the snapshot and the terminal drain. * 🚦 fix: Parked Run-End Signals, Interrupt Priority, Settled-Run Fallbacks Codex round 7 (5 fixes): - Run-end signals for a non-active conversation are PARKED per conversation instead of squatting the shared index slot: a later run finishing on the same pane can no longer overwrite them, and the parked drain fires when the user returns. - "Interrupt & send" front-inserts carry a priority flag that outranks createdAt when abort leftovers merge back chronologically — the urgent redirect drains first, not the oldest steer. - STEER_UNSUPPORTED/RUN_PAUSED/QUEUE_FULL rejections landing after the run settled mirror the NO_ACTIVE_RUN fallback and send immediately (queueing would strand the text with no run-end signal left); on the pinned SDK this is the common Enter-near-run-end path. - A failed abort (e.g. 404 when the run completed first) still signals the interrupt drain, so the queued interrupt message can't strand and the armed flag can't leak onto a later run. - Steered-image fallback alt text is localized (com_ui_attached_image). * 📌 chore: Adopt Published @librechat/agents Types Post-Bump dev's pin bump to ^3.2.62 (the release carrying injection + steer replay) landed via merge; the steering runtime now uses the SDK's real InjectedMessage/hook-output types instead of the local structural mirrors that bridged the pre-publish window. The two-half capability probe stays as the defensive gate for mismatched deployments — and the capability spec now exercises its TRUE path against the published package in CI. * 🛅 feat: Park-and-Claim Steer Recovery + Host-View Content Reads Codex round 8 (6 fixed incl. both P1s, 1 push-back): - The long-deferred no-subscriber gap is closed: every terminal drain (final, aborted-final, error, abortJob, resumed finalize) PARKS acknowledged leftovers on the job hash (unrecoveredSteers), and the status route claims them exactly once for inactive jobs — a client that closed/reloaded past the transient final event restores its steers as queued chips within the post-terminal TTL. A replacement run clears the parked copy (a live client started it). - Same-instance content reads are steer-complete: RedisJobStore now caches the HOST content array (WeakRef) via setContentParts and prefers it over the SDK graph cache, whose view never contains host-authored steer parts; the graph fallback splice-INSERTS steer chunks at their recorded host-view indices (the graph array is unshifted, so assignment would overwrite SDK parts). - Replay token accounting now counts prepended file-context text: full stamped content minus the steer body (already counted), so large steered documents hit the budget instead of bypassing pruning. - The queue drain restores an item when ask() refuses without sending (history not yet in cache after navigating back) — text is never silently dropped. - The armed interrupt flag travels WITH a parked run-end signal, so another run on the same pane can neither consume nor clear it. - parseTextParts extracts steer text (search indexing / audio). * 🎛️ refactor: Single Send Slot + In-Thread Steer Messages - Merge the during-run send affordance into the send/stop button slot: with composer text the send button replaces Stop (Enter = default action), hover reveals Steer/Queue/Interrupt rows with shortcuts; drop the separate DuringRunActionsMenu chevron - Add during-run keyboard chords: Cmd/Ctrl+Enter = non-default action, Alt+Enter = interrupt & send (plain-Enter submitters only) - Render steers as standard user messages in the thread: SteerPart (icon + author header + user text presentation) replaces the SteerBubble, and submitted steers appear immediately at the projected injection point via the PendingSteers slot on the streaming message - Keep composer rows only for recoverable states: failed steers (retry/edit/queue) and queued follow-ups * 🩹 fix: Keep the Replacement Submission Alive Across Abort Settlement The aborted run's final SSE event fires before the abort HTTP response resolves, so an armed interrupt & send drains and starts the NEXT submission while the abort POST is still in flight. The response handler's unconditional clearAllSubmissions() then reset the new submission, aborting its stream attach before the subscribe — the follow-up ran and persisted server-side but the live placeholder finalized empty (content appeared only after reload). useAbortCleanup captures the submission before the abort round-trip and both settlement paths (success and 404-catch) clear only when the captured submission is still current; a replacement stays untouched. Plain Stop behavior is unchanged. * 🧭 test: Playwright E2E for Mid-Run Steering and Queuing - Add e2e/specs/mock/steering.spec.ts: steer mid-run (202 + immediate in-thread pending part + real MCP tool boundary + words survive run end), Cmd/Ctrl+Enter queue with auto-send after clean completion, and Alt+Enter interrupt & send with the follow-up streaming into the live view - Add the E2E_STEER_TOOL_REPLY fake-model marker: slow preamble, a real remember_fact MCP tool call (PostToolBatch boundary), then a final turn - Test 1 pins the run-end degradation contract while the SDK's top-level agentId stamping bug blocks live injection; its header documents the assertions to flip once the fixed SDK is pinned * 🧷 fix: Job-Independent Steer Recovery + Expiry and Resume-Gap Parking Codex round 10: the park-and-claim recovery had lifecycle holes. - Move parked steers off the job hash onto their own bounded-TTL store key (JOB_CREATE_LUA resets it; deleteJob leaves it alone): the default completeJob path deletes the job record immediately, and the Redis read path never deserialized the old hash field — recovery previously worked only with STREAM_KEEP_COMPLETED_JOBS on the in-memory store - Carry the owner identity inside the parked payload and authorize the claim against it, so the status route recovers steers on its jobless branch too (the common reload-after-terminal case); a non-owner claim returns nothing and re-parks the payload - Park queued steers on approval expiry: snapshot the frozen queue before the requires_action→aborted CAS (whose terminal cleanup drops the steers key) and park only when the CAS wins - Mirror the terminal drain/park block in resume.js's failure path, which previously let completeJob's backstop clear 202-accepted steers - Close the Redis snapshot→subscribe resume gap: re-peek the queue after attaching and re-surface missed on_steer_applied events from the durable content view (synthesizeAppliedSteerEvents), updating resumeState.pendingSteers to the live queue * 📌 chore: Require @librechat/agents 3.2.63 + Applied-Steer E2E Contract - Bump the @librechat/agents pin to ^3.2.63 in api/ and packages/api/: it scopes the hook agentId marker to subagent child graphs, so the steering drain hook fires at top-level tool-batch boundaries and mid-run injection is active (danny-avila/agents PR 307) - Flip e2e steering test 1 from the documented degradation contract to the applied-steer contract: the optimistic in-thread part transitions to the persisted part at the tool boundary and survives inside the response after run end, with no queued follow-up turn * 🎗️ feat: Steered Messages Join the Message-Nav Ribs Steers are user messages, so they get their own clickable rib on the navigation rail, interleaved at their in-thread position inside the response that absorbed them (one DOM query in document order). SteerPart anchors itself as #steer-<id> with a steer-render marker — both the optimistic pending entry and the persisted part — and the rib carries the user role label with a preview drawn from the steer's text body, skipping the author header. * ❎ feat: Cancel a Queued Steer Before Injection + True User-Message Alignment - Add POST /chat/steer/cancel: removes ONE still-queued steer by id via an atomic list rebuild (Redis Lua preserves order and TTL), authorized against the job owner; removed:false is advisory — the cancel lost its race to the drain or the run end, never an error - Surface an × on the in-thread pending steer (server-acknowledged entries only): optimistic removal, restored if the POST fails since the server would still inject the words - Outdent SteerPart past the response's icon column so steers sit flush with top-level message rows, reading as regular user messages * 🧯 fix: Round-11 Recovery Hardening + Provider-Free Pending Slot - Reconcile the resume steer gap by steerId SETS, not queue length — a steer added in the gap (or an equal-length drain+enqueue swap) now refreshes resumeState.pendingSteers and still synthesizes the missed on_steer_applied events - Make completeJob's terminal backstop park: direct error-path callers without the controllers' close-and-park no longer silently clear 202-accepted steers (createdAt-guarded closeAndDrain + owner park before the terminal write) - Persist the steer part BEFORE media encoding in the drain hook: an abort inside the encode window can no longer lose a file-steer (the part refs come from the enqueue-sanitized item; replay re-encodes per turn unchanged) - Move the parked-claim owner check INSIDE the atomic store claim (substring gate in the Lua / in-memory equivalent): a non-owner probe can no longer transiently delete the recovery payload; the app-side parse stays authoritative - Park queued steers in BOTH stores' own requires_action expiry cleanup, which bypassed the manager-level sweep - Sweep expired parked steers from the in-memory store's periodic cleanup; restore a queued chip when send-now's submit is refused; upsert steer ACKs so an SSE reconnect reseed cannot duplicate chips - Mount the cancel mutation per steer item so the pending slot needs no QueryClient on ordinary streaming renders (fixes the CI failure in ContentParts.integration.test) - Skipped delivery-gated parking (finding 8): transport receiver counts cannot prove browser delivery, and gating the only durable copy on them trades cosmetic chip resurrection for real text loss; the window is already bounded by claim-on-read, createJob reset, and the TTL * 🩺 fix: Annotate PARKED_STEERS_TTL_MS for isolatedDeclarations tsdown's d.ts generation requires explicit types on exported consts with computed initializers; tsc --noEmit does not run that check, so the round-11 export slipped past local verification and broke Build packages (and every downstream CI job that consumes the built dist). * 🛟 fix: Round-12 Terminal-Path Recovery + Durable Steer Events - Park queued steers before the stale-running reap deletes a crashed or hung job in BOTH stores — the one terminal path with no controller finalization; requires_action expiry parking refactored onto the same snapshot/park helpers - Enqueue instead of dropping when a steer fallback send is refused: both the NO_ACTIVE_RUN branch and the settled-run rejection branch now observe sendNow's false return - Recover on the SSE reconnect-404 terminal path: convert local pending steers to queued, claim parked steers via /chat/status, and write a non-completed run-end signal so interrupt flags release without auto-sending an unknown outcome - Fall back to a positive parked-recovery TTL when completedTtl is 0 (SET EX 0 is invalid and silently killed recovery) - Make on_steer_applied durable before publish: emitChunk gains a durable option that awaits the chunk-log append (best-effort) ahead of the transport publish; the default delta path stays fire-and-forget * 🔐 fix: Round-13 Steer Authorization + Trusted File Refs - Resolve client-supplied steer file refs against the DB owner-scoped at enqueue and queue only DB-derived shapes (same filter as the injection fetch, shared via refs.ts); any unresolved id fails loud with 400 — spoofed type/filepath metadata can no longer be persisted into assistant content or rendered in chat/share views - Enforce agent authorization on /chat/steer against the ORIGINATING run's job identity: the chat path's role gate (AGENTS:USE, with the same non-agents-endpoint skip) plus the per-agent ACL check with the capability bypass — revoked access mid-run can no longer inject; cancel stays ownership-only (nothing model-bound) - Mark steered uploads used after a successful enqueue (owner-scoped, best-effort) so the upload-window TTL cannot reap a file the persisted steer part references - Consume the parked recovery copy after live delivery: converting final/abort/error pendingSteers fires one owner-gated claim-on-read, so dismissed chips can no longer resurrect on a later reload * 🎙️ fix: Round-14 Composer-Context Fidelity + TTS and Queue-State Gaps - Keep steer text out of generic assistant text extraction: parseTextParts excludes STEER parts by default with an includeSteer opt-in for the full-record surfaces (Meili indexing, aborted-response persistence) — TTS callers no longer speak the user's own mid-run words - Mark queued uploads used at enqueue time via a minimal owner-scoped POST /files/usage (fail-closed without a user; upload limiters do not apply to a metadata touch), fired once wherever composer files enter the queued state — the upload-window TTL can no longer reap a file waiting out a long run or approval pause - Carry quote chips and manual skill picks on queued items: captured and consumed from the composer at queue/interrupt time exactly like files, threaded through the drain and send-now overrides, and restored by the queued row's Edit message - Key an early-aborted FIRST turn's run-end signal to NEW_CONVO (resolveRunEndTarget) so queued follow-ups stay visible on the restored new-chat composer instead of parking under an optimistic stream id the user never sees again * 🧿 fix: Round-15 Gap Coverage + Consolidated Sweep (Share Leak, Abort Ids, Chip Hygiene) - Run the resume steer-gap check for every still-active job: an empty snapshot no longer skips the re-peek, and synthesis now keys on the FRESH content view so an applied-in-gap steer that was never snapshotted still re-surfaces (over-emission is benign — applied-id dedupe, index-stable parts) - Thread queued context through steer degradation: sendQueuedNow passes the item's quotes/skills into submitSteer, and every fallback (requeue or settled send) restores them instead of dropping to text+files - Stop shared links from leaking steer attachment refs: the share snapshot now walks content — files-excluded shares strip steer-part files entirely; files-included shares sanitize and share-route them like top-level files (copy-on-write, non-steer content by reference) - Seed pending-steer chips unconditionally on load/return so a steer applied while away cannot linger as a stale chip beside its part - Use the abort response's resolved job id: chips/drain-signal land where the user actually is (NEW_CONVO for a new-held first turn, consistent with resolveRunEndTarget) while the parked-copy claim hits the resolved id instead of a no-op /chat/status/new - Open steered documents like normal message files (FilePreviewDialog) - Cap the applied-steer id set on the live path via a shared helper; kept surviving run end deliberately (late-ACK race depends on it) and fixed the atom comment that claimed otherwise * 💡 fix: Un-light Steer Ribs When Their Node Is Replaced Two stacked gaps kept a steer rib lit after scrolling away: the pending→applied swap replaces the DOM node under the same id, which produces no IntersectionObserver exit and — because the entry list dedupes on (id, preview) — no entries change either, so the observer kept watching a detached node; and the rail's mutation filter only reacted to .message-render nodes, so steer-node swaps and removals never triggered a refresh at all. - reconcileObservedElements re-points the observer at replaced nodes from the mutation-driven refresh regardless of entries identity, dropping stale visibility until the fresh node reports (the observer fires its initial intersection immediately, so a truly visible part re-lights within a frame) - The mutation filter now recognizes steer-render nodes alongside message rows * 🪪 fix: Round-16 Recovery Owner Fields + Context Stickiness + Share Labels - Park resumed-run leftovers with the manager facade's metadata owner fields: a bare job.userId is undefined on that shape, which made every parked payload from a resumed HITL run unclaimable - Keep a queued item's quotes/skills sticky through a successful steer ACK: the pending chip carries them (client-only), reseeds preserve them across reconnects, and every terminal conversion — local or server-list, merged by steerId — restores them onto the queued item - Convert resumeState.pendingSteers on the inactive status branch (deduped against unrecoveredSteers) so steers observed in the expired-pause-before-sweeper window convert instead of vanishing until a later reload - Label shared steer parts share-safely via the existing ShareContext: a viewer's own name no longer appears on the sharer's steered messages * ✂️ fix: Carry Steer Context Through the Failed-Chip Edit Action Retry and convert-to-queue already preserve a failed steer's carried quotes/skills; Edit message dropped them on the way back to the composer. It now restores them through the same context path.
2026-08-04 14:57:42 +00:00 · 2026-07-14 10:11:10 -04:00 · 2026-07-14 10:11:10 -04:00 · 9bb351ad9c
commit 9bb351ad9c
parent b0d46b0518
99 changed files with 10957 additions and 139 deletions
--- a/api/server/controllers/agents/tests/steer.spec.js
+++ b/api/server/controllers/agents/tests/steer.spec.js
@ -0,0 +1,194 @@
+const express = require('express');
+const request = require('supertest');
+
+const mockHandleSteerRequest = jest.fn();
+const mockCheckAccess = jest.fn();
+const mockCheckPermission = jest.fn();
+const mockHasCapability = jest.fn();
+const mockGetAgent = jest.fn();
+const mockLogger = { warn: jest.fn(), error: jest.fn(), debug: jest.fn(), info: jest.fn() };
+
+jest.mock('@librechat/data-schemas', () => ({
+  ...jest.requireActual('@librechat/data-schemas'),
+  logger: mockLogger,
+}));
+
+jest.mock('@librechat/api', () => ({
+  ...jest.requireActual('@librechat/api'),
+  handleSteerRequest: (...args) => mockHandleSteerRequest(...args),
+  checkAccess: (...args) => mockCheckAccess(...args),
+}));
+
+jest.mock('~/server/services/PermissionService', () => ({
+  checkPermission: (...args) => mockCheckPermission(...args),
+}));
+
+jest.mock('~/server/middleware/roles/capabilities', () => ({
+  hasCapability: (...args) => mockHasCapability(...args),
+}));
+
+jest.mock('~/models', () => ({
+  getRoleByName: jest.fn(),
+  getAgent: (...args) => mockGetAgent(...args),
+  getFiles: jest.fn(),
+  updateFilesUsage: jest.fn(),
+}));
+
+const { Permissions, PermissionTypes, PermissionBits } = require('librechat-data-provider');
+const SteerController = require('~/server/controllers/agents/steer');
+
+/**
+ * The guard ladder itself (validation, file sanitization, ownership, enqueue
+ * codes) is typed logic in `@librechat/api` and is covered against the REAL
+ * in-memory job manager by `packages/api/src/agents/steering/__tests__/request.spec.ts`.
+ * This spec only pins the thin wrapper contract: pass-through of user/body,
+ * verbatim status/body serialization, and the 500 failure envelope.
+ */
+function buildApp(user = { id: 'user-1', tenantId: 'tenant-1' }) {
+  const app = express();
+  app.use(express.json());
+  app.use((req, _res, next) => {
+    req.user = user;
+    next();
+  });
+  app.post('/chat/steer', SteerController);
+  return app;
+}
+
+describe('SteerController (wrapper)', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('serializes the handler result verbatim', async () => {
+    mockHandleSteerRequest.mockResolvedValue({
+      status: 202,
+      body: { status: 'queued', steerId: 's1', position: 1, conversationId: 'c1' },
+    });
+
+    const res = await request(buildApp())
+      .post('/chat/steer')
+      .send({ conversationId: 'c1', text: 'hello', files: [{ file_id: 'f1' }] });
+
+    expect(res.status).toBe(202);
+    expect(res.body).toEqual({
+      status: 'queued',
+      steerId: 's1',
+      position: 1,
+      conversationId: 'c1',
+    });
+    expect(mockHandleSteerRequest).toHaveBeenCalledWith(
+      { id: 'user-1', tenantId: 'tenant-1' },
+      { conversationId: 'c1', text: 'hello', files: [{ file_id: 'f1' }] },
+      {
+        getFiles: expect.any(Function),
+        updateFilesUsage: expect.any(Function),
+        checkAgentAccess: expect.any(Function),
+      },
+    );
+  });
+
+  it('passes rejection statuses through untouched', async () => {
+    mockHandleSteerRequest.mockResolvedValue({ status: 409, body: { code: 'RUN_PAUSED' } });
+
+    const res = await request(buildApp()).post('/chat/steer').send({ conversationId: 'c1' });
+
+    expect(res.status).toBe(409);
+    expect(res.body.code).toBe('RUN_PAUSED');
+  });
+
+  it('500s with STEER_FAILED when the handler throws', async () => {
+    mockHandleSteerRequest.mockRejectedValue(new Error('store down'));
+
+    const res = await request(buildApp())
+      .post('/chat/steer')
+      .send({ conversationId: 'c1', text: 'x' });
+
+    expect(res.status).toBe(500);
+    expect(res.body.code).toBe('STEER_FAILED');
+    expect(mockLogger.error).toHaveBeenCalled();
+  });
+});
+
+describe('createAgentAccessCheck (chat-route parity via job identity)', () => {
+  /** Posts a steer to capture the wired deps, then exercises the callback. */
+  async function captureAccessCheck(user) {
+    mockHandleSteerRequest.mockResolvedValue({ status: 202, body: {} });
+    await request(buildApp(user)).post('/chat/steer').send({ conversationId: 'c1', text: 'x' });
+    return mockHandleSteerRequest.mock.calls[0][2].checkAgentAccess;
+  }
+
+  const roleUser = { id: 'user-1', tenantId: 'tenant-1', role: 'USER' };
+
+  beforeEach(() => {
+    jest.clearAllMocks();
+    mockCheckAccess.mockResolvedValue(true);
+    mockHasCapability.mockResolvedValue(false);
+    mockGetAgent.mockResolvedValue({ _id: 'oid-1', id: 'agent_abc' });
+    mockCheckPermission.mockResolvedValue(true);
+  });
+
+  it('denies an agents run when the AGENTS:USE role gate fails, skipping resource calls', async () => {
+    mockCheckAccess.mockResolvedValue(false);
+    const check = await captureAccessCheck(roleUser);
+
+    await expect(check({ agentId: 'agent_abc', endpoint: 'agents' })).resolves.toBe(false);
+    expect(mockCheckAccess).toHaveBeenCalledWith(
+      expect.objectContaining({
+        permissionType: PermissionTypes.AGENTS,
+        permissions: [Permissions.USE],
+      }),
+    );
+    expect(mockGetAgent).not.toHaveBeenCalled();
+    expect(mockCheckPermission).not.toHaveBeenCalled();
+  });
+
+  it('runs the VIEW resource check against the resolved agent', async () => {
+    const check = await captureAccessCheck(roleUser);
+
+    await expect(check({ agentId: 'agent_abc', endpoint: 'agents' })).resolves.toBe(true);
+    expect(mockGetAgent).toHaveBeenCalledWith({ id: 'agent_abc' });
+    expect(mockCheckPermission).toHaveBeenCalledWith(
+      expect.objectContaining({
+        userId: 'user-1',
+        resourceId: 'oid-1',
+        requiredPermission: PermissionBits.VIEW,
+      }),
+    );
+  });
+
+  it('denies when the agent is gone or the ACL check fails', async () => {
+    const check = await captureAccessCheck(roleUser);
+
+    mockGetAgent.mockResolvedValueOnce(null);
+    await expect(check({ agentId: 'agent_abc', endpoint: 'agents' })).resolves.toBe(false);
+
+    mockCheckPermission.mockResolvedValueOnce(false);
+    await expect(check({ agentId: 'agent_abc', endpoint: 'agents' })).resolves.toBe(false);
+  });
+
+  it('honors the capability bypass without touching the agent or ACL', async () => {
+    mockHasCapability.mockResolvedValue(true);
+    const check = await captureAccessCheck(roleUser);
+
+    await expect(check({ agentId: 'agent_abc', endpoint: 'agents' })).resolves.toBe(true);
+    expect(mockGetAgent).not.toHaveBeenCalled();
+    expect(mockCheckPermission).not.toHaveBeenCalled();
+  });
+
+  it('allows ephemeral runs with no role gate (skipAgentCheck parity for non-agents endpoints)', async () => {
+    const check = await captureAccessCheck(roleUser);
+
+    await expect(check({ agentId: undefined, endpoint: 'openAI' })).resolves.toBe(true);
+    expect(mockCheckAccess).not.toHaveBeenCalled();
+    expect(mockCheckPermission).not.toHaveBeenCalled();
+  });
+
+  it('applies both gates when metadata has a real agent but no endpoint yet', async () => {
+    const check = await captureAccessCheck(roleUser);
+
+    await expect(check({ agentId: 'agent_abc', endpoint: undefined })).resolves.toBe(true);
+    expect(mockCheckAccess).toHaveBeenCalled();
+    expect(mockCheckPermission).toHaveBeenCalled();
+  });
+});
--- a/api/server/controllers/agents/client.js
+++ b/api/server/controllers/agents/client.js
@ -45,6 +45,11 @@ const {
  agentRequestsAskUserQuestion,
  attachAskUserQuestionArgs,
  createContentIndexOffsetHandlers,
+  createSteerIndexOffsetHandlers,
+  createSteerDrainHook,
+  isSteeringSupported,
+  buildSteerMedia,
+  stampSteerPartMedia,
  getRequestMemories,
  getMemoryAgentId,
  createMemoryProcessor,
@ -79,6 +84,7 @@ const {
 } = require('@librechat/agents');
 const {
  Constants,
+  SteerEvents,
  UsageEvents,
  Permissions,
  VisionModes,
@ -189,6 +195,11 @@ class AgentClient extends BaseClient {
    this.indexTokenCountMap = {};
    /** @type {Array<Record<string, unknown>> | null} */
    this.memoryPayload = null;
+    /** Mutable content-index shift shared with the steer offset handlers.
+     *  Incremented each time a steer part is spliced into `contentParts`, so
+     *  SDK-emitted indices that arrive after an injection land past it.
+     *  @type {import('@librechat/api').SteerOffsetState} */
+    this.steerOffsetState = { offset: 0 };
    /** @type {(messages: BaseMessage[]) => Promise<void>} */
    this.processMemory;
  }
@ -239,6 +250,78 @@ class AgentClient extends BaseClient {
    buffer.clear();
  }

+  /**
+   * Apply one drained steer to host state: append the steer content part at
+   * the live content index, bump the shared index offset so subsequent SDK
+   * step indices land past it, and emit `on_steer_applied` so the live client
+   * replaces its pending chip with the inline part (the emitted chunk also
+   * reaches the Redis chunk log for reconnect reconstruction).
+   *
+   * Runs BEFORE the drain hook's media encode so an abort during the encode
+   * cannot lose the steer. File refs persist from the queue item (sanitized at
+   * enqueue); replay/token accounting re-fetch owner-scoped and re-encode per
+   * turn (stampSteerPartMedia), so unauthorized ids drop out there.
+   *
+   * @param {string} streamId
+   * @param {import('@librechat/api').SteerQueueItem} item
+   */
+  async applySteerPart(streamId, item) {
+    const index = this.contentParts.length;
+    const part = {
+      type: ContentTypes.STEER,
+      [ContentTypes.STEER]: item.text,
+      steerId: item.steerId,
+      createdAt: item.createdAt,
+      ...(item.files?.length && { files: item.files }),
+    };
+    this.contentParts.push(part);
+    this.steerOffsetState.offset += 1;
+    // durable: the chunk-log XADD is this event's recovery record — it must
+    // commit before the publish or a cross-replica reconnect that missed the
+    // pub/sub delivery reconstructs content without the steer part.
+    await GenerationJobManager.emitChunk(
+      streamId,
+      {
+        event: SteerEvents.ON_STEER_APPLIED,
+        data: {
+          steerId: item.steerId,
+          index,
+          part,
+          responseMessageId: this.responseMessageId,
+          conversationId: this.conversationId,
+        },
+      },
+      { durable: true },
+    );
+  }
+
+  /**
+   * The `steering` fragment for `createRun`: the run-scoped PostToolBatch
+   * drain hook, or `undefined` when there is no resumable job surface or the
+   * installed SDK cannot inject hook messages (draining would drop them).
+   *
+   * @param {string | undefined} streamId
+   */
+  buildSteerWiring(streamId) {
+    if (!streamId || !isSteeringSupported()) {
+      return undefined;
+    }
+    return {
+      hook: createSteerDrainHook({
+        streamId,
+        jobCreatedAt: this.jobCreatedAt,
+        applySteer: (item) => this.applySteerPart(streamId, item),
+        buildMedia: (item) =>
+          buildSteerMedia({
+            client: this,
+            user: this.options.req?.user,
+            item,
+            getFiles: db.getFiles,
+          }),
+      }),
+    };
+  }
+
  setOptions(_options) {}

  /**
@ -513,6 +596,42 @@ class AgentClient extends BaseClient {
    }

    payload = formattedMessages;
+    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. */
+      const stamped = await stampSteerPartMedia({
+        client: this,
+        user: this.options.req?.user,
+        payload,
+        // addPreviousAttachments already fetched steer-part refs in its single
+        // per-turn historical-files query — no second round trip.
+        docsById: this.authorizedHistoricalFiles,
+        getFiles: db.getFiles,
+      });
+      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. */
+        const fullTokens = countFormattedMessageTokens({ role: 'user', content: media }, encoding);
+        const bodyTokens = steerText
+          ? countFormattedMessageTokens(
+              { role: 'user', content: [{ type: ContentTypes.TEXT, text: steerText }] },
+              encoding,
+            )
+          : 0;
+        const mediaTokens = Math.max(0, (fullTokens ?? 0) - (bodyTokens ?? 0));
+        if (Number.isFinite(mediaTokens) && mediaTokens > 0) {
+          indexTokenCountMap[index] = (indexTokenCountMap[index] ?? 0) + mediaTokens;
+          promptTokenTotal += mediaTokens;
+        }
+      }
+    }
    this.memoryPayload = hasFileContext ? memoryPayload : null;
    messages = orderedMessages;
    promptTokens = promptTokenTotal;
@ -1226,6 +1345,9 @@ class AgentClient extends BaseClient {
      (part, index) =>
        index >= this.contentParts.length - 1 ||
        part.type === ContentTypes.TOOL_CALL ||
+        // Steer parts are user speech, not intermediate agent output — dropping
+        // one would erase the user's words from the persisted turn.
+        part.type === ContentTypes.STEER ||
        part.tool_call_ids,
    );
  }
@ -1366,6 +1488,13 @@ class AgentClient extends BaseClient {
      event: ApprovalEvents.ON_PENDING_ACTION,
      data: toClientPendingAction(pendingAction),
    });
+    // Steers queued before this pause stay IN the store for the whole approval
+    // window: `resumeState.pendingSteers` re-seeds the client's chips on
+    // reload, and the resumed run drains them at its first tool boundary.
+    // Draining here would leave the only copy in ephemeral client state — a
+    // reload during the pause would silently lose the user's message. New
+    // steers are rejected while paused (enqueue is status-guarded), and the
+    // requires_action TTL extension keeps the queue key alive.
    logger.debug(
      `[AgentClient] Paused ${streamId} for ${interrupt.payload.type} (action ${pendingAction.actionId})`,
    );
@ -1608,6 +1737,7 @@ class AgentClient extends BaseClient {
          );
        }

+        const streamId = this.options.req?._resumableStreamId;
        run = await createRun({
          agents,
          messages,
@ -1616,13 +1746,20 @@ class AgentClient extends BaseClient {
          // opts into the tool-approval wiring. Non-resumable callers (OpenAI-compat, Responses)
          // leave this off so an approval-gated tool can't pause where there's no resume path.
          hitlCapable: true,
+          // Mid-run steering: drain queued user messages at each tool-batch
+          // boundary and inject them into graph state. The offset wrapper
+          // shifts SDK content indices past any spliced steer parts.
+          steering: this.buildSteerWiring(streamId),
          indexTokenCountMap,
          initialSummary,
          initialSessions,
          calibrationRatio,
          runId: this.responseMessageId,
          signal: abortController.signal,
-          customHandlers: this.options.eventHandlers,
+          customHandlers: createSteerIndexOffsetHandlers(
+            this.options.eventHandlers,
+            this.steerOffsetState,
+          ),
          requestBody: config.configurable.requestBody,
          user: createSafeUser(this.options.req?.user),
          tenantId: this.options.req?.user?.tenantId,
@ -1652,7 +1789,6 @@ class AgentClient extends BaseClient {
          this._resolveRun = null;
        }

-        const streamId = this.options.req?._resumableStreamId;
        if (streamId && run.Graph) {
          GenerationJobManager.setGraph(streamId, run.Graph);
        }
@ -1931,6 +2067,7 @@ class AgentClient extends BaseClient {
      // graph otherwise has no `Graph.sessions` entries (especially cross-replica).
      const initialSessions = buildInitialToolSessions({ skillSessions, agents });

+      const streamId = this.options.req?._resumableStreamId;
      run = await createRun({
        agents,
        // State (messages, tool calls) is rehydrated from the checkpoint by
@ -1939,6 +2076,9 @@ class AgentClient extends BaseClient {
        // The resumed run can pause AGAIN (another tool, a follow-up question), and this
        // controller owns that lifecycle, so it must keep the HITL wiring on the rebuilt run.
        hitlCapable: true,
+        // Steering stays live across a pause/resume cycle: steers queued while
+        // the resumed segment runs drain at its tool-batch boundaries.
+        steering: this.buildSteerWiring(streamId),
        // Replay deferred tools discovered before the pause. With `messages: []` the
        // discovery scan finds nothing, so a deferred tool the paused call targets
        // would be absent from the rebuilt toolMap; these names (captured at pause)
@ -1950,10 +2090,15 @@ class AgentClient extends BaseClient {
        // The rebuilt graph numbers content indices from 0, but the aggregator was
        // just seeded with the pre-pause parts at those same indices — shift every
        // resumed step index past the seed, or the new output merges into (or, on a
-        // type mismatch, is silently dropped against) the pre-pause content.
-        customHandlers: createContentIndexOffsetHandlers(
-          this.options.eventHandlers,
-          Array.isArray(seedContent) ? seedContent : [],
+        // type mismatch, is silently dropped against) the pre-pause content. The
+        // steer wrapper composes on top: resumed indices shift by seed + any
+        // steer parts spliced in while the resumed segment streams.
+        customHandlers: createSteerIndexOffsetHandlers(
+          createContentIndexOffsetHandlers(
+            this.options.eventHandlers,
+            Array.isArray(seedContent) ? seedContent : [],
+          ),
+          this.steerOffsetState,
        ),
        requestBody: config.configurable.requestBody,
        user: createSafeUser(this.options.req?.user),
@ -1977,7 +2122,6 @@ class AgentClient extends BaseClient {
        this._resolveRun = null;
      }

-      const streamId = this.options.req?._resumableStreamId;
      // Do NOT cache the rebuilt graph on resume: it was created with `messages: []`, so
      // RedisJobStore.getContentParts() (which prefers a cached graph over reconstructing
      // from the chunk log) would return only the resumed segment and drop the pre-pause
--- a/api/server/controllers/agents/request.js
+++ b/api/server/controllers/agents/request.js
@ -2,6 +2,7 @@ const { logger } = require('@librechat/data-schemas');
 const { Constants, ViolationTypes, isEphemeralAgentId } = require('librechat-data-provider');
 const {
  sendEvent,
+  toPendingSteer,
  getViolationInfo,
  buildMessageFiles,
  getReferencedQuotes,
@ -756,6 +757,32 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
          await titleEventPromise;
        }

+        // Steers that never reached an injection boundary (queued after the last
+        // tool batch, or the run had none). The close-and-drain atomically stops
+        // new enqueues first — a steer POST racing this finalization gets 404
+        // (client sends it as a normal message) instead of a 202 whose payload
+        // completeJob would then silently clear. Reported on the final event so
+        // the client converts them to queued follow-up messages.
+        let pendingSteers;
+        try {
+          const leftoverSteers = await GenerationJobManager.steering.closeAndDrain(
+            streamId,
+            jobCreatedAt,
+          );
+          if (leftoverSteers.length > 0) {
+            pendingSteers = leftoverSteers.map(toPendingSteer);
+            // Parked BEFORE the final event: a client with no live subscriber
+            // recovers these via /chat/status (claim-on-read) within the
+            // recovery TTL — the SSE copy alone is transient.
+            await GenerationJobManager.steering.park(streamId, pendingSteers, {
+              userId,
+              tenantId: req.user?.tenantId,
+            });
+          }
+        } catch (err) {
+          logger.warn(`[ResumableAgentController] Failed to drain leftover steers`, err);
+        }
+
        if (!wasAbortedBeforeComplete) {
          const finalEvent = {
            final: true,
@ -763,6 +790,7 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
            title: conversation.title,
            requestMessage: sanitizeMessageForTransmit(userMessage),
            responseMessage: { ...response },
+            ...(pendingSteers && { pendingSteers }),
          };

          logger.debug(`[ResumableAgentController] Emitting FINAL event`, {
@ -783,6 +811,7 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
            title: conversation.title,
            requestMessage: sanitizeMessageForTransmit(userMessage),
            responseMessage: { ...response, unfinished: true },
+            ...(pendingSteers && { pendingSteers }),
          };

          logger.debug(`[ResumableAgentController] Emitting ABORTED FINAL event`, {
@ -849,6 +878,31 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
          // abortJob already handled emitDone and completeJob
        } else {
          logger.error(`[ResumableAgentController] Generation error for ${streamId}:`, error);
+          // Close the steer queue BEFORE the error event reaches clients: a
+          // steer POST racing this failure gets 404 (client queues or sends it)
+          // instead of a 202 whose payload would vanish with the job. Text
+          // recovery is client-side — acknowledged chips convert to queued.
+          try {
+            const erroredLeftovers = await GenerationJobManager.steering.closeAndDrain(
+              streamId,
+              jobCreatedAt,
+            );
+            if (erroredLeftovers.length > 0) {
+              // The error event is a bare string — park the acknowledged
+              // steers so a reloaded/disconnected client can still recover
+              // them via /chat/status instead of losing them with the queue.
+              await GenerationJobManager.steering.park(
+                streamId,
+                erroredLeftovers.map(toPendingSteer),
+                { userId, tenantId: req.user?.tenantId },
+              );
+            }
+          } catch (drainErr) {
+            logger.warn(
+              `[ResumableAgentController] Failed to close steer queue on error`,
+              drainErr,
+            );
+          }
          await GenerationJobManager.emitError(streamId, error.message || 'Generation failed');
          GenerationJobManager.completeJob(streamId, error.message);
        }
--- a/api/server/controllers/agents/resume.js
+++ b/api/server/controllers/agents/resume.js
@ -16,6 +16,7 @@ const {
  filterMalformedContentParts,
  decrementPendingRequest,
  checkAndIncrementPendingRequest,
+  toPendingSteer,
 } = require('@librechat/api');
 const { disposeClient } = require('~/server/cleanup');
 const {
@ -343,6 +344,32 @@ async function finalizeResumedTurn({ req, client, job, streamId, conversationId,
    return;
  }

+  // Steers that never reached an injection boundary during the resumed
+  // segment — mirror the normal request path's terminal drain: the atomic
+  // close (createdAt-guarded) rejects a steer POST racing this finalization,
+  // and the leftovers ride the final event as queued follow-ups instead of
+  // being 202-ACKed and then silently cleared by completeJob.
+  let pendingSteers;
+  try {
+    const leftoverSteers = await GenerationJobManager.steering.closeAndDrain(
+      streamId,
+      job.createdAt,
+    );
+    if (leftoverSteers.length > 0) {
+      pendingSteers = leftoverSteers.map(toPendingSteer);
+      // Same no-subscriber recovery as the normal final path (claim-on-read
+      // via /chat/status within the recovery TTL). NOTE: `job` is the manager
+      // facade — owner fields live under `metadata` (a bare `job.userId` is
+      // undefined and would make the parked payload unclaimable).
+      await GenerationJobManager.steering.park(streamId, pendingSteers, {
+        userId: job.metadata?.userId,
+        tenantId: job.metadata?.tenantId,
+      });
+    }
+  } catch (drainErr) {
+    logger.warn('[ResumeAgentController] Failed to drain leftover steers', drainErr);
+  }
+
  const finalEvent = {
    final: true,
    conversation,
@ -361,6 +388,7 @@ async function finalizeResumedTurn({ req, client, job, streamId, conversationId,
        })
      : null,
    responseMessage: { ...responseMessage },
+    ...(pendingSteers && { pendingSteers }),
  };

  await GenerationJobManager.emitDone(streamId, finalEvent);
@ -681,6 +709,25 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
        `[ResumeAgentController] Skipping failed-resume finalization — job ${streamId} was replaced`,
      );
    } else {
+      // A steer 202-accepted during the failed resume segment would otherwise
+      // be silently cleared by completeJob's backstop — mirror the normal
+      // request error path: close the queue BEFORE the error event (racing
+      // steer POSTs get 404) and park the leftovers for /chat/status recovery.
+      try {
+        const leftoverSteers = await GenerationJobManager.steering.closeAndDrain(
+          streamId,
+          job.createdAt,
+        );
+        if (leftoverSteers.length > 0) {
+          // Facade shape: owner fields are under `metadata` (see finalize).
+          await GenerationJobManager.steering.park(streamId, leftoverSteers.map(toPendingSteer), {
+            userId: job.metadata?.userId,
+            tenantId: job.metadata?.tenantId,
+          });
+        }
+      } catch (drainErr) {
+        logger.warn('[ResumeAgentController] Failed to drain steers on resume failure', drainErr);
+      }
      try {
        await GenerationJobManager.emitError(streamId, err?.message ?? 'Resume failed');
      } catch (emitErr) {
--- a/api/server/controllers/agents/steer.js
+++ b/api/server/controllers/agents/steer.js
@ -0,0 +1,111 @@
+const { checkAccess, handleSteerRequest, handleSteerCancel } = require('@librechat/api');
+const { logger, ResourceCapabilityMap } = require('@librechat/data-schemas');
+const {
+  Permissions,
+  ResourceType,
+  PermissionBits,
+  PermissionTypes,
+  isAgentsEndpoint,
+  isEphemeralAgentId,
+} = require('librechat-data-provider');
+const { checkPermission } = require('~/server/services/PermissionService');
+const { hasCapability } = require('~/server/middleware/roles/capabilities');
+const db = require('~/models');
+
+/**
+ * Steer-time agent authorization, mirroring the chat route's middlewares
+ * (`checkAgentAccess` + `canAccessAgentFromBody`) against the ORIGINATING
+ * run's identity from job metadata instead of the request body:
+ * - role gate: AGENTS:USE via `checkAccess`, applied exactly when chat.js
+ *   would run it (`skipAgentCheck` skips non-agents endpoints);
+ * - resource gate: `canAccessResource`'s capability bypass + `checkPermission`
+ *   VIEW on the resolved agent, skipped for ephemeral/no-agent runs.
+ *
+ * @param {import('express').Request} req
+ * @returns {(run: import('@librechat/api').SteerRunContext) => Promise<boolean>}
+ */
+const createAgentAccessCheck =
+  (req) =>
+  async ({ agentId, endpoint }) => {
+    const hasRealAgent = agentId != null && !isEphemeralAgentId(agentId);
+    const roleGateApplies = endpoint == null ? hasRealAgent : isAgentsEndpoint(endpoint);
+    if (roleGateApplies) {
+      const roleAllowed = await checkAccess({
+        req,
+        user: req.user,
+        permissionType: PermissionTypes.AGENTS,
+        permissions: [Permissions.USE],
+        getRoleByName: db.getRoleByName,
+      });
+      if (!roleAllowed) {
+        return false;
+      }
+    }
+    if (!hasRealAgent) {
+      return true;
+    }
+    let bypass = false;
+    try {
+      bypass = await hasCapability(req.user, ResourceCapabilityMap[ResourceType.AGENT]);
+    } catch {
+      bypass = false;
+    }
+    if (bypass) {
+      return true;
+    }
+    const agent = await db.getAgent({ id: agentId });
+    if (!agent) {
+      return false;
+    }
+    return checkPermission({
+      userId: req.user.id,
+      role: req.user.role,
+      resourceType: ResourceType.AGENT,
+      resourceId: agent._id,
+      requiredPermission: PermissionBits.VIEW,
+    });
+  };
+
+/**
+ * POST /api/agents/chat/steer
+ *
+ * Thin wrapper: the full guard ladder (validation, file sanitization,
+ * capability gate, ownership/tenant checks, agent access, owner-scoped file
+ * resolve, status-guarded enqueue) lives in `@librechat/api`
+ * (`handleSteerRequest`), which returns the HTTP status + JSON body to
+ * serialize verbatim. DB access and permission services are injected here.
+ */
+const SteerController = async (req, res) => {
+  try {
+    const { status, body } = await handleSteerRequest(req.user ?? {}, req.body ?? {}, {
+      getFiles: db.getFiles,
+      updateFilesUsage: db.updateFilesUsage,
+      checkAgentAccess: createAgentAccessCheck(req),
+    });
+    return res.status(status).json(body);
+  } catch (error) {
+    logger.error('[SteerController] Failed to queue steer', error);
+    return res.status(500).json({ code: 'STEER_FAILED' });
+  }
+};
+
+/**
+ * POST /api/agents/chat/steer/cancel
+ *
+ * Removes a still-queued steer before injection. `removed: false` is not an
+ * error — the cancel lost its race (already injected, or the run ended) and
+ * the client defers to the events it will receive. No agent-access check:
+ * a cancel injects nothing model-bound, so ownership checks suffice.
+ */
+const SteerCancelController = async (req, res) => {
+  try {
+    const { status, body } = await handleSteerCancel(req.user ?? {}, req.body ?? {});
+    return res.status(status).json(body);
+  } catch (error) {
+    logger.error('[SteerCancelController] Failed to cancel steer', error);
+    return res.status(500).json({ code: 'STEER_CANCEL_FAILED' });
+  }
+};
+
+module.exports = SteerController;
+module.exports.SteerCancelController = SteerCancelController;