diff --git a/api/app/clients/BaseClient.js b/api/app/clients/BaseClient.js index b7c5a8ae59..6d541e160f 100644 --- a/api/app/clients/BaseClient.js +++ b/api/app/clients/BaseClient.js @@ -41,6 +41,16 @@ const collectHistoricalFileRefs = (message) => { if (Array.isArray(message.attachments)) { refs.push(...message.attachments); } + /** Steer parts carry their own attachment refs inside assistant content; + * collecting them here folds the steer replay stamp's lookup into this + * single per-turn query (see `stampSteerPartMedia`). */ + if (Array.isArray(message.content)) { + for (const part of message.content) { + if (part?.type === ContentTypes.STEER && Array.isArray(part.files)) { + refs.push(...part.files); + } + } + } return refs; }; @@ -1457,6 +1467,9 @@ class BaseClient { } } } + /** Owner-scoped docs for THIS turn, including steer-part refs — the steer + * replay stamp consumes this instead of issuing a second query. */ + this.authorizedHistoricalFiles = authorizedFilesById; /** * diff --git a/api/app/clients/prompts/formatAgentMessages.spec.js b/api/app/clients/prompts/formatAgentMessages.spec.js index d8e9262ba9..5a4b6937ba 100644 --- a/api/app/clients/prompts/formatAgentMessages.spec.js +++ b/api/app/clients/prompts/formatAgentMessages.spec.js @@ -511,4 +511,113 @@ describe('formatAgentMessages', () => { expect(assistant.additional_kwargs?.signatures).toBeUndefined(); }); }); + + describe('steer content parts', () => { + it('replays a steer between tool steps as a standalone HumanMessage', () => { + const payload = [ + { + role: 'assistant', + content: [ + { + type: ContentTypes.TEXT, + [ContentTypes.TEXT]: 'Checking the weather.', + tool_call_ids: ['t1'], + }, + { + type: ContentTypes.TOOL_CALL, + tool_call: { id: 't1', name: 'search', args: '{}', output: 'sunny' }, + }, + { + type: ContentTypes.STEER, + [ContentTypes.STEER]: 'also check tomorrow', + steerId: 's1', + }, + { + type: ContentTypes.TEXT, + [ContentTypes.TEXT]: 'Checking tomorrow too.', + tool_call_ids: ['t2'], + }, + { + type: ContentTypes.TOOL_CALL, + tool_call: { id: 't2', name: 'search', args: '{}', output: 'rain' }, + }, + ], + }, + ]; + + const result = formatAgentMessages(payload); + expect(result.map((m) => m.constructor)).toEqual([ + AIMessage, + ToolMessage, + HumanMessage, + AIMessage, + ToolMessage, + ]); + expect(result[2].content).toBe('also check tomorrow'); + expect(result[2].additional_kwargs).toEqual({ source: 'steer' }); + expect(result[3].tool_calls).toHaveLength(1); + }); + + it('flushes accumulated assistant text before the steer', () => { + const payload = [ + { + role: 'assistant', + content: [ + { type: ContentTypes.TEXT, [ContentTypes.TEXT]: 'Some prose so far.' }, + { type: ContentTypes.STEER, [ContentTypes.STEER]: 'change direction' }, + { type: ContentTypes.TEXT, [ContentTypes.TEXT]: 'New direction prose.' }, + ], + }, + ]; + + const result = formatAgentMessages(payload); + expect(result.map((m) => m.constructor)).toEqual([AIMessage, HumanMessage, AIMessage]); + expect(result[0].content).toBe('Some prose so far.'); + expect(result[1].content).toBe('change direction'); + expect(result[2].content).toEqual([ + { type: ContentTypes.TEXT, [ContentTypes.TEXT]: 'New direction prose.' }, + ]); + }); + + it('handles a steer as the final content part', () => { + const payload = [ + { + role: 'assistant', + content: [ + { type: ContentTypes.TEXT, [ContentTypes.TEXT]: 'Answer text.' }, + { type: ContentTypes.STEER, [ContentTypes.STEER]: 'trailing steer' }, + ], + }, + ]; + + const result = formatAgentMessages(payload); + expect(result.map((m) => m.constructor)).toEqual([AIMessage, HumanMessage]); + expect(result[1].content).toBe('trailing steer'); + }); + + it('prefers stamped media content for multimodal steers', () => { + const media = [ + { type: 'text', text: 'see the chart' }, + { type: 'image_url', image_url: { url: 'data:image/png;base64,abc', detail: 'auto' } }, + ]; + const payload = [ + { + role: 'assistant', + content: [ + { + type: ContentTypes.STEER, + [ContentTypes.STEER]: 'see the chart', + files: [{ file_id: 'f1' }], + media, + }, + ], + }, + ]; + + const result = formatAgentMessages(payload); + expect(result).toHaveLength(1); + expect(result[0]).toBeInstanceOf(HumanMessage); + expect(result[0].content).toEqual(media); + }); + }); }); diff --git a/api/app/clients/prompts/formatMessages.js b/api/app/clients/prompts/formatMessages.js index 8435ff5280..795c62c14f 100644 --- a/api/app/clients/prompts/formatMessages.js +++ b/api/app/clients/prompts/formatMessages.js @@ -229,6 +229,43 @@ const formatAgentMessages = (payload) => { } else if (part.type === ContentTypes.THINK) { hasReasoning = true; continue; + } else if (part.type === ContentTypes.STEER) { + /* + A mid-run steer: user speech persisted inline in the assistant message. + Flush any accumulated assistant text first so ordering is preserved, then + replay the steer as a standalone user message. `lastAIMessage` is NOT + reset — the aggregator emits a fresh text-with-tool_call_ids part for any + post-steer tool step, and preceding tool_call parts already pushed their + ToolMessages, so the HumanMessage lands after them (valid provider order). + */ + if (currentContent.length > 0) { + if (currentContent.some((curr) => curr.type !== ContentTypes.TEXT)) { + /** Non-text parts (images, files) must survive the flush intact — + * folding to text here would drop them from replayed history. */ + messages.push(new AIMessage({ content: currentContent })); + } else { + const content = currentContent + .reduce((acc, curr) => `${acc}${curr[ContentTypes.TEXT] ?? ''}\n`, '') + .trim(); + if (content.length > 0) { + messages.push(new AIMessage({ content })); + } + } + currentContent = []; + } + messages.push( + new HumanMessage({ + content: + Array.isArray(part.media) && part.media.length > 0 + ? part.media + : (part[ContentTypes.STEER] ?? ''), + additional_kwargs: { source: 'steer' }, + }), + ); + /** A post-steer tool_call must mint a FRESH assistant anchor — + * attaching to the pre-steer one would emit its ToolMessage after + * the HumanMessage while the call sat before it (invalid order). */ + lastAIMessage = null; } else if (part.type === ContentTypes.ERROR || part.type === ContentTypes.AGENT_UPDATE) { continue; } else { diff --git a/api/package.json b/api/package.json index 312945cca4..3dedf01b8e 100644 --- a/api/package.json +++ b/api/package.json @@ -46,7 +46,7 @@ "@azure/storage-blob": "^12.30.0", "@google/genai": "^2.8.0", "@keyv/redis": "^4.3.3", - "@librechat/agents": "^3.2.62", + "@librechat/agents": "^3.2.63", "@librechat/api": "*", "@librechat/data-schemas": "*", "@microsoft/microsoft-graph-client": "^3.0.7", diff --git a/api/server/controllers/agents/__tests__/steer.spec.js b/api/server/controllers/agents/__tests__/steer.spec.js new file mode 100644 index 0000000000..a0108b3cbc --- /dev/null +++ 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(); + }); +}); diff --git a/api/server/controllers/agents/client.js b/api/server/controllers/agents/client.js index 4a2768f809..81364f5559 100644 --- 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> | 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} */ 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 diff --git a/api/server/controllers/agents/request.js b/api/server/controllers/agents/request.js index 81213a23d9..5b63c06ef8 100644 --- 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); } diff --git a/api/server/controllers/agents/resume.js b/api/server/controllers/agents/resume.js index 4c01063571..7f0fc2d106 100644 --- 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) { diff --git a/api/server/controllers/agents/steer.js b/api/server/controllers/agents/steer.js new file mode 100644 index 0000000000..baa86de010 --- /dev/null +++ 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} + */ +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; diff --git a/api/server/routes/agents/__tests__/abort.spec.js b/api/server/routes/agents/__tests__/abort.spec.js index c9cc6b5b0f..538e01b243 100644 --- a/api/server/routes/agents/__tests__/abort.spec.js +++ b/api/server/routes/agents/__tests__/abort.spec.js @@ -47,6 +47,7 @@ jest.mock('~/server/middleware', () => ({ req.user = { id: 'test-user-123' }; next(); }, + moderateText: (req, res, next) => next(), messageIpLimiter: (req, res, next) => next(), configMiddleware: (req, res, next) => next(), messageUserLimiter: (req, res, next) => next(), diff --git a/api/server/routes/agents/__tests__/streamTenant.spec.js b/api/server/routes/agents/__tests__/streamTenant.spec.js index 708a071228..55d96d8cc3 100644 --- a/api/server/routes/agents/__tests__/streamTenant.spec.js +++ b/api/server/routes/agents/__tests__/streamTenant.spec.js @@ -39,6 +39,7 @@ jest.mock('~/server/middleware', () => ({ req.user = { id: mockUserId, tenantId: mockTenantId }; next(); }, + moderateText: (req, res, next) => next(), messageIpLimiter: (req, res, next) => next(), configMiddleware: (req, res, next) => next(), messageUserLimiter: (req, res, next) => next(), diff --git a/api/server/routes/agents/index.js b/api/server/routes/agents/index.js index 81612197e8..759d9c0177 100644 --- a/api/server/routes/agents/index.js +++ b/api/server/routes/agents/index.js @@ -9,17 +9,20 @@ const { isHITLEnabled, deleteAgentCheckpoint, attachAskUserQuestionArgs, + createMessageFilterPii, } = require('@librechat/api'); const { createSseStreamTelemetry } = require('@librechat/api/telemetry'); const { logger } = require('@librechat/data-schemas'); const { uaParser, checkBan, + moderateText, requireJwtAuth, messageIpLimiter, configMiddleware, messageUserLimiter, } = require('~/server/middleware'); +const SteerController = require('~/server/controllers/agents/steer'); const { saveMessage } = require('~/models'); const responses = require('./responses'); const openai = require('./openai'); @@ -193,7 +196,18 @@ router.get('/chat/status/:conversationId', async (req, res) => { const job = await GenerationJobManager.getJob(conversationId); if (!job) { - return res.json({ active: false }); + // The default completeJob path deletes the job record immediately, so the + // jobless branch IS the common reload-after-terminal case — parked steers + // live under their own bounded-TTL key and the claim authorizes against + // the payload's stored owner (no job record is left to check). + const claimed = await GenerationJobManager.steering.claim(conversationId, { + userId: req.user.id, + tenantId: req.user.tenantId, + }); + return res.json({ + active: false, + ...(claimed.length > 0 && { unrecoveredSteers: claimed }), + }); } if (job.metadata.userId !== req.user.id) { @@ -215,8 +229,23 @@ router.get('/chat/status/:conversationId', async (req, res) => { const pendingLive = job.status === 'requires_action' && !isPendingActionStale({ pendingAction }); const isActive = job.status === 'running' || pendingLive; + /** Acknowledged steers the terminal drains parked because no subscriber was + * live to receive the final/abort event — claim-on-read (cleared once + * returned) so the reloading client restores them as queued follow-ups. */ + let unrecoveredSteers; + if (!isActive) { + const claimed = await GenerationJobManager.steering.claim(conversationId, { + userId: req.user.id, + tenantId: req.user.tenantId, + }); + if (claimed.length > 0) { + unrecoveredSteers = claimed; + } + } + res.json({ active: isActive, + ...(unrecoveredSteers && { unrecoveredSteers }), streamId: conversationId, status: job.status, aggregatedContent: resumeState?.aggregatedContent ?? [], @@ -379,13 +408,58 @@ router.post('/chat/abort', configMiddleware, async (req, res) => { } } - return res.json({ success: true, aborted: jobStreamId }); + return res.json({ + success: true, + aborted: jobStreamId, + // Steers that never reached an injection boundary — restored client-side + // as queued chips so the user's words aren't dropped with the abort. + ...(abortResult.pendingSteers?.length > 0 && { pendingSteers: abortResult.pendingSteers }), + }); } logger.warn(`[AgentStream] Job not found for streamId: ${jobStreamId}`); return res.status(404).json({ error: 'Job not found', streamId: jobStreamId }); }); +/** + * @route POST /chat/steer + * @desc Queue a mid-run user message for injection at the next tool boundary + * @access Private + * @description Mounted before chatRouter to bypass buildEndpointOption middleware, + * but a steer is model-bound user text, so it carries the same guards as a normal + * message IN THE SAME ORDER as chat.js: the configured IP/user rate limiters, + * the PII filter FIRST (blocked sensitive text must never reach the external + * moderation endpoint), then `moderateText`. + */ +const steerLimiters = []; +if (isEnabled(LIMIT_MESSAGE_IP)) { + steerLimiters.push(messageIpLimiter); +} +if (isEnabled(LIMIT_MESSAGE_USER)) { + steerLimiters.push(messageUserLimiter); +} +router.post( + '/chat/steer', + configMiddleware, + ...steerLimiters, + createMessageFilterPii({ getConfig: (req) => req.config?.messageFilter?.pii }), + moderateText, + SteerController, +); + +/** + * @route POST /chat/steer/cancel + * @desc Remove a still-queued steer before injection (no model-bound content, + * so no PII/moderation pass — just the shared rate limiters) + * @access Private + */ +router.post( + '/chat/steer/cancel', + configMiddleware, + ...steerLimiters, + SteerController.SteerCancelController, +); + router.use('/', v1); const chatRouter = express.Router(); diff --git a/api/server/routes/files/files.js b/api/server/routes/files/files.js index bfd0faaf0d..fafa02e66d 100644 --- a/api/server/routes/files/files.js +++ b/api/server/routes/files/files.js @@ -4,6 +4,7 @@ const { logger, SystemCapabilities } = require('@librechat/data-schemas'); const { logAxiosError, refreshS3FileUrls, + handleFilesUsageRequest, resolveUploadErrorMessage, verifyAgentUploadPermission, } = require('@librechat/api'); @@ -138,6 +139,26 @@ router.get('/config', async (req, res) => { } }); +/** + * POST /files/usage + * + * Owner-scoped TTL touch for uploads held in a client-side queue (mid-run + * queued messages), so the upload-window TTL cannot reap them before drain. + * Thin wrapper: validation, cap, and best-effort semantics live in + * `@librechat/api` (`handleFilesUsageRequest`). + */ +router.post('/usage', async (req, res) => { + try { + const { status, body } = await handleFilesUsageRequest(req.user ?? {}, req.body ?? {}, { + updateFilesUsage: db.updateFilesUsage, + }); + return res.status(status).json(body); + } catch (error) { + logger.error('[/files/usage] Failed to mark files used', error); + return res.status(500).json({ code: 'FILES_USAGE_FAILED' }); + } +}); + router.delete('/', async (req, res) => { try { const { files: _files } = req.body; diff --git a/api/server/routes/files/files.test.js b/api/server/routes/files/files.test.js index c2930f3294..6cd60d43d6 100644 --- a/api/server/routes/files/files.test.js +++ b/api/server/routes/files/files.test.js @@ -936,4 +936,77 @@ describe('File Routes - Delete with Agent Access', () => { ); }); }); + + describe('POST /files/usage', () => { + it('marks owned files used and clears the upload TTL', async () => { + const ownFileId = uuidv4(); + await createFile({ + user: otherUserId, + file_id: ownFileId, + filename: 'queued.png', + filepath: '/uploads/queued.png', + bytes: 10, + type: 'image/png', + }); + await File.updateOne({ file_id: ownFileId }, { $set: { expiresAt: new Date() } }); + + const response = await request(app) + .post('/files/usage') + .send({ file_ids: [ownFileId] }); + + expect(response.status).toBe(200); + expect(response.body).toEqual({ marked: 1 }); + const marked = await File.findOne({ file_id: ownFileId }).lean(); + expect(marked.usage).toBe(1); + expect(marked.expiresAt).toBeUndefined(); + }); + + it("is owner-scoped: another user's file stays untouched (best-effort 200)", async () => { + await File.updateOne({ file_id: fileId }, { $set: { expiresAt: new Date() } }); + + const response = await request(app) + .post('/files/usage') + .send({ file_ids: [fileId] }); + + expect(response.status).toBe(200); + expect(response.body).toEqual({ marked: 0 }); + const untouched = await File.findOne({ file_id: fileId }).lean(); + expect(untouched.usage).toBe(0); + expect(untouched.expiresAt).toBeDefined(); + }); + + it('rejects a list over the cap', async () => { + const file_ids = Array.from({ length: 11 }, () => uuidv4()); + const response = await request(app).post('/files/usage').send({ file_ids }); + expect(response.status).toBe(400); + expect(response.body.code).toBe('TOO_MANY_FILES'); + }); + + it('rejects invalid bodies', async () => { + expect((await request(app).post('/files/usage').send({})).status).toBe(400); + expect((await request(app).post('/files/usage').send({ file_ids: 'f1' })).status).toBe(400); + expect( + ( + await request(app) + .post('/files/usage') + .send({ file_ids: [1] }) + ).status, + ).toBe(400); + }); + + it('rejects unauthenticated requests', async () => { + const bareApp = express(); + bareApp.use(express.json()); + bareApp.use((req, res, next) => { + req.app.locals = {}; + next(); + }); + bareApp.use('/files', router); + + const response = await request(bareApp) + .post('/files/usage') + .send({ file_ids: [fileId] }); + expect(response.status).toBe(401); + }); + }); }); diff --git a/api/server/routes/files/index.js b/api/server/routes/files/index.js index ec00750a60..f7e2428c4c 100644 --- a/api/server/routes/files/index.js +++ b/api/server/routes/files/index.js @@ -31,9 +31,10 @@ const initialize = async () => { const { fileUploadIpLimiter, fileUploadUserLimiter } = createFileLimiters(); - /** Apply rate limiters to all POST routes (excluding /speech which is handled above) */ + /** Apply rate limiters to all POST routes (excluding /speech which is handled + * above, and /usage — a metadata touch that must not consume upload quota) */ router.use((req, res, next) => { - if (req.method === 'POST' && !req.path.startsWith('/speech')) { + if (req.method === 'POST' && !req.path.startsWith('/speech') && req.path !== '/usage') { return fileUploadIpLimiter(req, res, (err) => { if (err) { return next(err); diff --git a/client/src/components/Chat/ChatView.tsx b/client/src/components/Chat/ChatView.tsx index 5ab4efb1e9..c894e251d5 100644 --- a/client/src/components/Chat/ChatView.tsx +++ b/client/src/components/Chat/ChatView.tsx @@ -11,6 +11,7 @@ import { useResumeOnLoad, useAdaptiveSSE, useChatHelpers, + useQueueDrain, useLocalize, } from '~/hooks'; import { ChatContext, AddedChatContext, ChatFormProvider, useFileMapContext } from '~/Providers'; @@ -73,6 +74,9 @@ function ChatView({ index = 0, project }: { index?: number; project?: TChatProje // Wait for messages to load before resuming to avoid race condition useResumeOnLoad(conversationId, chatHelpers.getMessages, index, !isLoading); + // Auto-send queued follow-up messages once a run finishes cleanly. + useQueueDrain(index, conversationId, chatHelpers.ask); + let content: JSX.Element | null | undefined; const isLandingPage = (!messagesTree || messagesTree.length === 0) && diff --git a/client/src/components/Chat/Input/ChatForm.tsx b/client/src/components/Chat/Input/ChatForm.tsx index 894e234592..a37fe87084 100644 --- a/client/src/components/Chat/Input/ChatForm.tsx +++ b/client/src/components/Chat/Input/ChatForm.tsx @@ -1,10 +1,11 @@ import { memo, useRef, useMemo, useEffect, useState, useCallback } from 'react'; import { useWatch } from 'react-hook-form'; import { TextareaAutosize } from '@librechat/client'; -import { useRecoilState, useRecoilValue } from 'recoil'; +import { useRecoilState, useRecoilValue, useRecoilCallback } from 'recoil'; import { Constants, isAssistantsEndpoint, isAgentsEndpoint } from 'librechat-data-provider'; -import type { TConversation } from 'librechat-data-provider'; +import type { TMessage, TConversation } from 'librechat-data-provider'; import type { ExtendedFile, FileSetter, ConvoGenerator } from '~/common'; +import type { QueuedMessageContext } from '~/hooks/Chat/useSteering'; import { useTextarea, useAutoSave, @@ -25,10 +26,13 @@ import PendingManualSkillsChips from './PendingManualSkillsChips'; import useAskAnswerMode from '~/hooks/Input/useAskAnswerMode'; import AskUserQuestionPopover from './AskUserQuestionPopover'; import { cn, getModelSpec, removeFocusRings } from '~/utils'; +import DuringRunSendButton from './DuringRunSendButton'; import { useGetStartupConfig } from '~/data-provider'; import { mainTextareaId, BadgeItem } from '~/common'; +import PendingSteerChips from './PendingSteerChips'; import PendingQuoteChips from './PendingQuoteChips'; import AttachFileChat from './Files/AttachFileChat'; +import useSteering from '~/hooks/Chat/useSteering'; import FileFormChat from './Files/FileFormChat'; import TextareaHeader from './TextareaHeader'; import PromptsCommand from './PromptsCommand'; @@ -57,6 +61,7 @@ interface ChatFormProps { setFilesLoading: React.Dispatch>; newConversation: ConvoGenerator; handleStopGenerating: (e: React.MouseEvent) => void; + stopGenerating: () => void; } const ChatForm = memo(function ChatForm({ @@ -70,6 +75,7 @@ const ChatForm = memo(function ChatForm({ setFilesLoading, newConversation, handleStopGenerating, + stopGenerating, }: ChatFormProps) { const submitButtonRef = useRef(null); const textAreaRef = useRef(null); @@ -181,6 +187,104 @@ const ChatForm = memo(function ChatForm({ const { submitMessage, submitPrompt } = useSubmitMessage(); + /** Queued/steered sends carry their FULL submission context: explicit + * (possibly empty) overrides stop `ask` from vacuuming quotes or skill + * picks the user has staged in the composer for their NEXT message. */ + const sendNow = useCallback( + (text: string, overrideFiles?: TMessage['files'], context?: QueuedMessageContext) => + submitMessage({ + text, + overrideFiles, + overrideQuotes: context?.quotes ?? [], + overrideManualSkills: context?.manualSkills ?? [], + }), + [submitMessage], + ); + /** Chip "Edit message" restore: quote chips + skill picks merge back into + * their compose-time atoms (the chips above the textarea re-render them). */ + const restoreComposerContext = useRecoilCallback( + ({ set }) => + (context?: QueuedMessageContext) => { + const { quotes, manualSkills } = context ?? {}; + if (quotes != null && quotes.length > 0) { + set(store.pendingQuotesByConvoId(conversationId), (prev) => [ + ...new Set([...prev, ...quotes]), + ]); + } + if (manualSkills != null && manualSkills.length > 0) { + set(store.pendingManualSkillsByConvoId(conversationId), (prev) => [ + ...new Set([...prev, ...manualSkills]), + ]); + } + }, + [conversationId], + ); + /** Chip "Edit message": the text replaces the composer draft and the chip's + * attachments merge back into the composer file map (already uploaded, so + * they restore as completed entries — same shape as draft recovery). */ + const editToComposer = useCallback( + (text: string, chipFiles?: TMessage['files'], context?: QueuedMessageContext) => { + methods.setValue('text', text, { shouldDirty: true }); + if (chipFiles != null && chipFiles.length > 0) { + setFiles((prev) => { + const next = new Map(prev); + for (const file of chipFiles) { + if (!file.file_id) { + continue; + } + next.set(file.file_id, { + file_id: file.file_id, + filename: file.filename, + filepath: file.filepath, + type: file.type ?? '', + height: file.height, + width: file.width, + size: file.bytes ?? 0, + progress: 1, + attached: true, + }); + } + return next; + }); + } + restoreComposerContext(context); + textAreaRef.current?.focus(); + }, + [methods, setFiles, restoreComposerContext], + ); + const steering = useSteering({ + index, + conversationId, + conversation, + isSubmitting, + answerModeActive: answerMode.active, + files, + setFiles, + filesLoading, + sendNow, + stopGenerating, + }); + + /** ⌘/Ctrl+Enter = the non-default during-run action, ⌥/Alt+Enter = + * interrupt & send — the counterpart of Enter's `submitDuringRun`. */ + const handleDuringRunModifier = useCallback( + (kind: 'other' | 'interrupt') => { + const text = methods.getValues('text'); + let consumed = false; + if (kind === 'interrupt') { + consumed = steering.interruptAndSend(text); + } else if (steering.effectiveAction === 'steer') { + consumed = steering.queueFromComposer(text); + } else { + consumed = steering.steerFromComposer(text); + } + if (consumed) { + methods.reset(); + } + }, + [methods, steering], + ); + const handleKeyUp = useHandleKeyUp({ index, textAreaRef, @@ -200,6 +304,9 @@ const ChatForm = memo(function ChatForm({ placeholder: answerMode.active ? (answerMode.otherLabel ?? localize('com_ui_something_else')) : placeholder, + // Enter stays live during a run when it can steer/queue instead of send. + allowSubmitWhileGenerating: steering.duringRunActive, + onDuringRunModifier: steering.duringRunActive ? handleDuringRunModifier : undefined, }); useQueryParams({ textAreaRef }); @@ -244,6 +351,23 @@ const ChatForm = memo(function ChatForm({ const isMoreThanThreeRows = visualRowCount > 3; + /** One button slot while a run is generating: with composer text the send + * button takes over (Enter steers/queues; hover reveals all actions); + * clearing the text restores Stop. */ + const duringRunSlot = + steering.duringRunActive && (textValue?.trim() ?? '') !== '' ? ( + methods.getValues('text')} + onConsumed={() => methods.reset()} + disabled={filesLoading} + /> + ) : ( + + ); + const baseClasses = useMemo( () => cn( @@ -263,6 +387,14 @@ const ChatForm = memo(function ChatForm({ if (answerMode.active && answerMode.submitText(data.text)) { return; } + // During a run, a submit steers or queues per the effective action + // instead of starting a new turn (which would be dropped anyway). + if (steering.duringRunActive) { + if (steering.submitDuringRun(data.text)) { + methods.reset(); + } + return; + } return submitMessage(data); })} className={cn( @@ -318,6 +450,13 @@ const ChatForm = memo(function ChatForm({ {quotesEnabled && } + {steering.enabled && ( + + )} {/* WIP */} )}
- {isSubmitting && showStopButton && !answerMode.active ? ( - - ) : ( - endpoint && ( - - ) - )} + {isSubmitting && showStopButton && !answerMode.active + ? duringRunSlot + : endpoint && ( + + )}
{TextToSpeech && automaticPlayback && } @@ -472,6 +609,7 @@ function ChatFormWrapper({ index = 0, placeholder }: { index?: number; placehold setFilesLoading, newConversation, handleStopGenerating, + stopGenerating, } = useChatContext(); /** @@ -512,6 +650,12 @@ function ChatFormWrapper({ index = 0, placeholder }: { index?: number; placehold [], ); + const stopRef = useRef(stopGenerating); + stopRef.current = stopGenerating; + const stableStop = useCallback(() => { + void stopRef.current(); + }, []); + return ( ); } diff --git a/client/src/components/Chat/Input/DuringRunSendButton.tsx b/client/src/components/Chat/Input/DuringRunSendButton.tsx new file mode 100644 index 0000000000..01f6d57fa1 --- /dev/null +++ b/client/src/components/Chat/Input/DuringRunSendButton.tsx @@ -0,0 +1,147 @@ +import React, { forwardRef } from 'react'; +import * as Ariakit from '@ariakit/react'; +import { useWatch } from 'react-hook-form'; +import { SendIcon } from '@librechat/client'; +import { Zap, Clock, OctagonPause } from 'lucide-react'; +import type { Control } from 'react-hook-form'; +import type { SteeringControls } from '~/hooks/Chat/useSteering'; +import { isMacPlatform } from '~/utils/shortcuts'; +import { useLocalize } from '~/hooks'; +import { cn } from '~/utils'; + +const ROW_CLASS = + 'flex w-full cursor-pointer items-center gap-2 rounded-lg px-2.5 py-1.5 text-sm text-text-primary hover:bg-surface-tertiary focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-border-xheavy aria-disabled:cursor-not-allowed aria-disabled:opacity-50'; + +function Kbd({ children }: { children: React.ReactNode }) { + return ( + + {children} + + ); +} + +type ActionRow = { + key: string; + label: string; + kbd: string; + icon: React.ReactNode; + disabled?: boolean; + onClick: () => void; +}; + +type DuringRunSendButtonProps = { + control: Control<{ text: string }>; + steering: SteeringControls; + getText: () => string; + onConsumed: () => void; + /** External hold (e.g. uploads in flight), mirroring the normal send button. */ + disabled?: boolean; +}; + +/** + * The send button while a run is generating: it takes over the send/stop slot + * (and `submitButtonRef`, so Enter's synthetic click routes here) whenever the + * composer holds text — submitting steers or queues per the effective action. + * Hovering it reveals the full action list with its shortcuts: steer, queue + * (⌘/Ctrl+Enter routes to the non-default action), and interrupt & send + * (⌥/Alt+Enter). Clearing the composer restores the Stop button. + */ +const DuringRunSendButton = React.memo( + forwardRef((props: DuringRunSendButtonProps, ref: React.ForwardedRef) => { + const localize = useLocalize(); + const { steering } = props; + const data = useWatch({ control: props.control }); + const content = data?.text?.trim(); + const primary = steering.effectiveAction; + const modEnter = isMacPlatform ? '⌘⏎' : 'Ctrl ⏎'; + const altEnter = isMacPlatform ? '⌥⏎' : 'Alt ⏎'; + + const runAction = (action: (text: string) => boolean | void) => { + const text = props.getText().trim(); + if (text.length === 0) { + return; + } + if (action(text) !== false) { + props.onConsumed(); + } + }; + + const steerRow: ActionRow = { + key: 'steer', + label: localize('com_ui_steer'), + kbd: primary === 'steer' ? '⏎' : modEnter, + icon: