diff --git a/CONTEXT.md b/CONTEXT.md index 30f79289b6..3880719db0 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -13,6 +13,7 @@ - **Subagent completion wakeup**: a durable internal `continue` trigger pre-registered before detached child execution so a process crash cannot lose the wakeup. Delivery defers until the child's terminal transcript is persisted, targets the initiating agent and exact parent response branch, carries task metadata rather than child output, waits for the parent generation to settle, and starts the parent turn that collects the result through the existing task store. - **Agent continuation preparation**: the single source-dispatch seam that resolves a durable `continue` delivery immediately before admission. Bound Event Actor work selects its binding adapter; internal completion work selects an adapter by stable source identity. Preparation may resolve authoritative input and branch state or settle already-consumed work, but it does not own source result truth, delivery ordering, or generation execution. - **Warm terminal steer continuation**: a queued steer accepted before a generation's terminal boundary may continue the same SDK `Run` without creating a replacement generation. After parallel Stop hooks fold, the serialized StopFinalize phase tells the job store whether another continuation is already planned or terminal progress is forbidden. The store atomically chooses among claiming the current protocol-v2 FIFO batch, keeping empty admission open for an already-planned segment, and sealing admission so every racing or later message becomes an ordinary follow-up. Claimed steer receipts remain the crash-recovery authority; protocol-v1 generations always seal because they cannot recover an ambiguous terminal claim. Tool-batch, preemption, and terminal boundaries share one durable apply-and-inject adapter, while the SDK owns the bounded Stop-continuation loop. +- **Agent queued turn**: a server-owned ordinary follow-up accepted while an Agent generation is active. Its Mongo row is the sole FIFO, payload, and lifecycle authority; the trigger delivery is only a replayable wakeup and Agent execution enrollment is only an execution adapter. The lifecycle reserves the deterministic delivery identity before publication, admits only after the captured branch has a clean durable predecessor outcome, and commits the source-owned generation receipt only after provider invocation has been enrolled. An accepted or deduplicated loopback response requires that exact receipt; a process death before it remains explicit admission-indeterminate evidence instead of silently consuming the text. Admission reconciliation is leased, backoff-scheduled work: exact generation evidence may repair legacy ambiguity, while current source-owned ambiguity never infers provider outcome from transient job state. Aborted or failed predecessors and exhausted admission attempts remain visible terminal rows. Transport-ambiguous enqueue outcomes stay non-resendable until exact request-identity reconciliation succeeds; an elapsed client observation window may hide the warning locally but never converts ambiguity into permission to submit again. Conversation deletion cancels the rows, retires their deliveries, and removes their payloads before deleting the conversation wave. - **Subagent activity stream**: an observational, task-scoped live projection of bounded child progress for the currently open private panel. It may cross API replicas through Redis, never carries hidden reasoning text, and never controls or settles execution. The durable child thread remains canonical and its existing polling view is the fallback for missed or unavailable live events. - **Agent event handling outcome**: the durable, generation-fenced result of a previously accepted event delivery. `started` proves generation admission; terminal states distinguish verified tool application, clean completion without action, failure, and cancellation. Transport success remains separate so an accepted event cannot masquerade as completed work. - **Agent event expected action**: an optional source-declared tool name and bounded argument subset evaluated against host-observed completed run steps. It is evidence policy, not authorization and not a model-authored success claim. diff --git a/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js b/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js index 693aa6cb56..09d41af482 100644 --- a/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js +++ b/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js @@ -76,6 +76,8 @@ const mockExemptFromConcurrencyLimiter = jest.fn(); const mockRecordScheduleOutcome = jest.fn(); const mockIsScheduleLive = jest.fn(); const mockDeleteAgentCheckpoint = jest.fn(); +const mockSettleAgentQueuedTurnExecutionAdmission = jest.fn(); +const mockVerifyAgentQueuedTurnExecutionAdmission = jest.fn(); const mockExecuteAgentEventActor = jest.fn(); const mockResumeAgentEventActor = jest.fn(); const mockCreateAgentEventActorTurn = jest.fn((input, dependencies) => { @@ -379,6 +381,13 @@ jest.mock('~/server/services/Schedules', () => ({ isScheduleLive: (...args) => mockIsScheduleLive(...args), })); +jest.mock('~/server/services/Agents/triggers', () => ({ + settleAgentQueuedTurnExecutionAdmission: (...args) => + mockSettleAgentQueuedTurnExecutionAdmission(...args), + verifyAgentQueuedTurnExecutionAdmission: (...args) => + mockVerifyAgentQueuedTurnExecutionAdmission(...args), +})); + const AgentController = require('../request'); const { ErrorTypes } = require('librechat-data-provider'); const { disposeClient: mockDisposeClient } = require('~/server/cleanup'); @@ -485,6 +494,8 @@ describe('ResumableAgentController resume metadata', () => { mockSaveMessage.mockResolvedValue({}); mockSaveConvo.mockResolvedValue({}); mockDeleteAgentCheckpoint.mockResolvedValue(undefined); + mockSettleAgentQueuedTurnExecutionAdmission.mockResolvedValue(true); + mockVerifyAgentQueuedTurnExecutionAdmission.mockResolvedValue(true); mockGetAgentEventActorSnapshot.mockResolvedValue({ state: null, reconciliations: [] }); mockCommitAgentEventActorState.mockResolvedValue({ status: 'stale' }); mockBeginAgentEventActorLegacyTurn.mockResolvedValue(true); @@ -1135,6 +1146,131 @@ describe('ResumableAgentController resume metadata', () => { ); }); + it('keeps a queued-turn source nonterminal when provider initialization fails', async () => { + mockGetMessages.mockResolvedValue([{ _id: 'persisted-parent' }]); + const admissionSource = { + source: 'agent-queued-turn', + sourceId: 'queued-turn-1', + claimId: 'queued-delivery-1', + claimBy: 'queued-worker-1', + }; + const req = { + _isAgentTrigger: true, + user: { id: 'user-123', tenantId: 'tenant-1' }, + body: { + text: 'Run the queued turn.', + messageId: 'queued-user-message', + parentMessageId: 'persisted-response_', + conversationId: 'conversation-123', + clientRequestId: 'queued-delivery-1', + agentContinuationAdmission: admissionSource, + endpointOption: { endpoint: 'agents', modelOptions: { model: 'gpt-4.1' } }, + }, + config: {}, + }; + const res = createResumableResponse(); + const initializeClient = jest.fn().mockRejectedValue(new Error('stop before provider')); + + await AgentController(req, res, jest.fn(), initializeClient, null); + + expect(initializeClient).toHaveBeenCalledTimes(1); + expect(mockSettleAgentQueuedTurnExecutionAdmission).not.toHaveBeenCalled(); + expect(res.status).toHaveBeenCalledWith(500); + }); + + it('commits a queued-turn receipt after provider invocation and before accepting HTTP', async () => { + mockGetMessages.mockResolvedValue([{ _id: 'persisted-parent' }]); + const admissionSource = { + source: 'agent-queued-turn', + sourceId: 'queued-turn-1', + claimId: 'queued-delivery-1', + claimBy: 'queued-worker-1', + }; + const req = { + _isAgentTrigger: true, + user: { id: 'user-123', tenantId: 'tenant-1' }, + body: { + text: 'Run the queued turn.', + messageId: 'queued-user-message', + parentMessageId: 'persisted-response_', + conversationId: 'conversation-123', + clientRequestId: 'queued-delivery-1', + agentContinuationAdmission: admissionSource, + endpointOption: { endpoint: 'agents', modelOptions: { model: 'gpt-4.1' } }, + }, + config: {}, + }; + const sendMessage = jest.fn(() => new Promise(() => {})); + const initializeClient = jest.fn().mockResolvedValue({ client: { options: {}, sendMessage } }); + const res = createResumableResponse(); + + await AgentController(req, res, jest.fn(), initializeClient, null); + await nextTick(); + + expect(mockSettleAgentQueuedTurnExecutionAdmission).toHaveBeenCalledWith(admissionSource, { + userId: 'user-123', + tenantId: 'tenant-1', + conversationId: 'conversation-123', + clientRequestId: 'queued-delivery-1', + generationId: 'conversation-123', + generationCreatedAt: 1000, + }); + expect(initializeClient.mock.invocationCallOrder[0]).toBeLessThan( + sendMessage.mock.invocationCallOrder[0], + ); + expect(sendMessage.mock.invocationCallOrder[0]).toBeLessThan( + mockSettleAgentQueuedTurnExecutionAdmission.mock.invocationCallOrder[0], + ); + expect(mockSettleAgentQueuedTurnExecutionAdmission.mock.invocationCallOrder[0]).toBeLessThan( + res.json.mock.invocationCallOrder[0], + ); + }); + + it('fails closed after provider invocation when the queued-turn receipt is unavailable', async () => { + mockGetMessages.mockResolvedValue([{ _id: 'persisted-parent' }]); + mockSettleAgentQueuedTurnExecutionAdmission.mockRejectedValue(new Error('mongo unavailable')); + const req = { + _isAgentTrigger: true, + user: { id: 'user-123' }, + body: { + text: 'Run the queued turn.', + messageId: 'queued-user-message', + parentMessageId: 'persisted-response_', + conversationId: 'conversation-123', + clientRequestId: 'queued-delivery-1', + agentContinuationAdmission: { + source: 'agent-queued-turn', + sourceId: 'queued-turn-1', + claimId: 'queued-delivery-1', + claimBy: 'queued-worker-1', + }, + endpointOption: { endpoint: 'agents', modelOptions: { model: 'gpt-4.1' } }, + }, + config: {}, + }; + const sendMessage = jest.fn(() => new Promise(() => {})); + const initializeClient = jest.fn().mockResolvedValue({ client: { options: {}, sendMessage } }); + const res = createResumableResponse(); + + await AgentController(req, res, jest.fn(), initializeClient, null); + await nextTick(); + + expect(mockGenerationJobManager.beginProviderExecution).toHaveBeenCalledTimes(1); + expect(sendMessage).toHaveBeenCalledTimes(1); + expect(res.status).toHaveBeenCalledWith(500); + expect(mockGenerationJobManager.completeJob).toHaveBeenCalledWith( + 'conversation-123', + 'mongo unavailable', + 1000, + expect.objectContaining({ beforeErrorPublication: expect.any(Function) }), + ); + expect(mockGenerationJobManager.markProviderExecutionDrained).toHaveBeenCalledWith( + 'conversation-123', + 1000, + 'provider-segment-1', + ); + }); + it('prefetches conversation state before admission and joins it with job metadata', async () => { let resolveConversation; let signalMetadataStarted; @@ -2042,6 +2178,61 @@ describe('ResumableAgentController resume metadata', () => { expect(mockGenerationJobManager.createJob).not.toHaveBeenCalled(); }); + it('fails closed when a deduplicated queued generation lacks its source receipt', async () => { + const reacquired = wonGenerationClaim({ + streamId: 'conversation-123', + conversationId: 'conversation-123', + claimToken: 'reacquired-token', + }); + mockGenerationJobManager.claimGeneration.mockResolvedValue(reacquired); + mockGenerationJobManager.resumeClaimedGeneration.mockResolvedValue({ + ...reacquired.existing, + startedAt: 42, + }); + mockVerifyAgentQueuedTurnExecutionAdmission.mockRejectedValue( + new Error('source receipt missing'), + ); + const admissionSource = { + source: 'agent-queued-turn', + sourceId: 'queued-turn-1', + claimId: 'queued-delivery-1', + claimBy: 'queued-worker-1', + }; + const req = { + _isAgentTrigger: true, + user: { id: 'user-123', tenantId: 'tenant-1' }, + body: { + text: 'Retry a queued generation.', + messageId: 'user-msg', + parentMessageId: 'assistant-1', + clientRequestId: 'queued-delivery-1', + conversationId: 'conversation-123', + agentContinuationAdmission: admissionSource, + endpointOption: { endpoint: 'agents', modelOptions: { model: 'gpt-4.1' } }, + }, + config: {}, + }; + const res = createResumableResponse(); + + await AgentController(req, res, jest.fn(), jest.fn(), null); + + expect(mockVerifyAgentQueuedTurnExecutionAdmission).toHaveBeenCalledWith(admissionSource, { + userId: 'user-123', + tenantId: 'tenant-1', + conversationId: 'conversation-123', + clientRequestId: 'queued-delivery-1', + generationId: 'conversation-123', + generationCreatedAt: 42, + }); + expect(res.status).toHaveBeenCalledWith(503); + expect(res.json).toHaveBeenCalledWith( + expect.objectContaining({ + code: 'SERVER_NOT_READY', + }), + ); + expect(mockGenerationJobManager.createJob).not.toHaveBeenCalled(); + }); + it('caps a reacquired v2 lease to the immutable v1 live-job protocol', async () => { const reacquired = wonGenerationClaim({ streamId: 'conversation-123', @@ -2090,7 +2281,7 @@ describe('ResumableAgentController resume metadata', () => { conversationId: 'orig-stream', claimedAt: Date.now() - 60000, claimToken: 'existing-token', - startedAt: Date.now() - 59000, + startedAt: 1000, }, }); mockGenerationJobManager.getJob.mockResolvedValue(undefined); @@ -2112,6 +2303,7 @@ describe('ResumableAgentController resume metadata', () => { expect(res.json).toHaveBeenCalledWith({ streamId: 'orig-stream', conversationId: 'orig-stream', + generationCreatedAt: 1000, status: 'resumed', generationProtocolVersion: 1, }); @@ -2127,7 +2319,7 @@ describe('ResumableAgentController resume metadata', () => { conversationId: 'orig-stream', claimedAt: Date.now() - 60_000, claimToken: 'existing-token', - startedAt: Date.now() - 59_000, + startedAt: 2000, generationProtocolVersion: 2, }, }); @@ -2150,6 +2342,7 @@ describe('ResumableAgentController resume metadata', () => { expect(res.json).toHaveBeenCalledWith({ conversationId: 'orig-stream', + generationCreatedAt: 2000, status: 'settled', generationProtocolVersion: 2, }); diff --git a/api/server/controllers/agents/queuedTurns.js b/api/server/controllers/agents/queuedTurns.js new file mode 100644 index 0000000000..3e66c9d3dd --- /dev/null +++ b/api/server/controllers/agents/queuedTurns.js @@ -0,0 +1,80 @@ +const { + handleAgentQueuedTurnEnqueue, + handleAgentQueuedTurnList, + handleAgentQueuedTurnCancel, +} = require('@librechat/api'); +const { logger } = require('@librechat/data-schemas'); +const db = require('~/models'); +const { createAgentAccessCheck } = require('./steer'); +const dependencies = (req) => { + /** Keep trigger-service composition lazy so importing the route for unrelated + * middleware probes does not initialize or require the worker graph. */ + const { + scheduleAgentQueuedTurn, + cancelAgentQueuedTurn, + } = require('~/server/services/Agents/triggers'); + return { + methods: db, + lifecycle: { + schedule: scheduleAgentQueuedTurn, + cancel: cancelAgentQueuedTurn, + }, + getFiles: db.getFiles, + updateFilesUsage: db.updateFilesUsage, + checkAgentAccess: createAgentAccessCheck(req), + isPrincipalActive: db.isAgentTriggerPrincipalActive, + }; +}; + +const send = (res, result) => res.status(result.status).json(result.body); + +const AgentQueuedTurnEnqueueController = async (req, res) => { + try { + return send( + res, + await handleAgentQueuedTurnEnqueue(req.user ?? {}, req.body ?? {}, dependencies(req)), + ); + } catch (error) { + logger.error('[AgentQueuedTurns] Failed to enqueue turn', error); + return res.status(500).json({ code: 'QUEUED_TURN_FAILED' }); + } +}; + +const AgentQueuedTurnListController = async (req, res) => { + try { + return send( + res, + await handleAgentQueuedTurnList( + req.user ?? {}, + req.query?.conversationId, + dependencies(req), + req.query?.clientRequestIds, + ), + ); + } catch (error) { + logger.error('[AgentQueuedTurns] Failed to list turns', error); + return res.status(500).json({ code: 'QUEUED_TURN_LIST_FAILED' }); + } +}; + +const AgentQueuedTurnCancelController = async (req, res) => { + try { + return send( + res, + await handleAgentQueuedTurnCancel( + req.user ?? {}, + req.params?.queuedTurnId, + dependencies(req), + ), + ); + } catch (error) { + logger.error('[AgentQueuedTurns] Failed to cancel turn', error); + return res.status(500).json({ code: 'QUEUED_TURN_CANCEL_FAILED' }); + } +}; + +module.exports = { + AgentQueuedTurnEnqueueController, + AgentQueuedTurnListController, + AgentQueuedTurnCancelController, +}; diff --git a/api/server/controllers/agents/request.js b/api/server/controllers/agents/request.js index 1941594534..363012e674 100644 --- a/api/server/controllers/agents/request.js +++ b/api/server/controllers/agents/request.js @@ -523,20 +523,30 @@ function sendSettledGeneration( conversationId, startupTelemetry, generationProtocolVersion, + generationCreatedAt, ) { startupTelemetry?.end('deduplicated'); if (generationProtocolVersion < GENERATION_PROTOCOL_V2) { return sendGenerationJson( res, 200, - { streamId, conversationId, status: 'resumed' }, + { + streamId, + conversationId, + ...(generationCreatedAt != null && { generationCreatedAt }), + status: 'resumed', + }, generationProtocolVersion, ); } return sendGenerationJson( res, 200, - { conversationId, status: 'settled' }, + { + conversationId, + ...(generationCreatedAt != null && { generationCreatedAt }), + status: 'settled', + }, generationProtocolVersion, ); } @@ -753,6 +763,55 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit req._isAgentTrigger === true && !isNewConvo && (parentMessageId !== Constants.NO_PARENT || isBoundEventContinuation); + const queuedTurnAdmissionSource = isTriggerContinuation + ? req.body?.agentContinuationAdmission + : undefined; + const hasQueuedTurnAdmissionSource = queuedTurnAdmissionSource != null; + const verifyQueuedTurnAdmission = async (generationId, generationCreatedAt) => { + if (!hasQueuedTurnAdmissionSource) { + return true; + } + if ( + typeof clientRequestId !== 'string' || + !Number.isSafeInteger(generationCreatedAt) || + generationCreatedAt < 0 + ) { + return false; + } + try { + const { + verifyAgentQueuedTurnExecutionAdmission, + } = require('~/server/services/Agents/triggers'); + const confirmed = await verifyAgentQueuedTurnExecutionAdmission(queuedTurnAdmissionSource, { + userId, + ...(tenantId != null && { tenantId }), + conversationId, + clientRequestId, + generationId, + generationCreatedAt, + }); + return confirmed === true; + } catch (error) { + logger.warn( + '[ResumableAgentController] Deduplicated queued-turn admission is not confirmed', + error, + ); + return false; + } + }; + const rejectUnconfirmedQueuedTurnAdmission = () => { + res.set('Retry-After', '1'); + startupTelemetry?.end('deduplicated'); + return sendGenerationJson( + res, + 503, + { + code: 'SERVER_NOT_READY', + error: 'Queued turn execution is still being confirmed. Please retry shortly.', + }, + generationProtocolVersion, + ); + }; if ( await isUnpersistedPreliminaryParent({ @@ -942,6 +1001,14 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit ? GENERATION_PROTOCOL_V2 : 1, ); + if ( + !(await verifyQueuedTurnAdmission( + existingLiveGeneration.streamId, + existingLiveGeneration.startedAt, + )) + ) { + return rejectUnconfirmedQueuedTurnAdmission(); + } startupTelemetry?.end('deduplicated'); return sendGenerationJson( res, @@ -1006,12 +1073,16 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit // generation (usually fast completion + cleanup), never an abandoned // pre-create lease that may be taken over and billed again. There is no // attachable stream; the settled response refetches persisted history. + if (!(await verifyQueuedTurnAdmission(existingStreamId, claim.existing.startedAt))) { + return rejectUnconfirmedQueuedTurnAdmission(); + } return sendSettledGeneration( res, existingStreamId, claim.existing.conversationId, startupTelemetry, generationProtocolVersion, + claim.existing.startedAt, ); } if (!liveJob && isLegacyTokenlessClaim && claimAgeMs >= IDEMPOTENCY_STARTUP_GRACE_MS) { @@ -1020,6 +1091,9 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit * attach/refetch path: this covers fast completion without starting a * second billed generation, while an abandoned pre-create claim ages * out under the old server's bounded TTL. */ + if (!(await verifyQueuedTurnAdmission(existingStreamId, claim.existing.startedAt))) { + return rejectUnconfirmedQueuedTurnAdmission(); + } return sendSettledGeneration( res, existingStreamId, @@ -1108,6 +1182,9 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit (startedAt != null && liveJob.createdAt !== startedAt) || (liveClientRequestId != null && liveClientRequestId !== clientRequestId); if (replacedGeneration) { + if (!(await verifyQueuedTurnAdmission(existingStreamId, startedAt))) { + return rejectUnconfirmedQueuedTurnAdmission(); + } // streamId === conversationId, so a later turn reuses the same route. // Never pair this stale POST's optimistic submission with that newer // job's SSE snapshot. If the replacement is still active, distinguish @@ -1141,6 +1218,7 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit claim.existing.conversationId, startupTelemetry, generationProtocolVersion, + claim.existing.startedAt, ); } if (liveClientRequestId == null && !isLegacyTokenlessClaim) { @@ -1164,6 +1242,9 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit clientRequestId, streamId: existingStreamId, }); + if (!(await verifyQueuedTurnAdmission(existingStreamId, liveJob.createdAt))) { + return rejectUnconfirmedQueuedTurnAdmission(); + } startupTelemetry?.end('deduplicated'); return sendGenerationJson( res, @@ -1518,7 +1599,6 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit status: 409, }); } - acceptAgentStartupTelemetry(req, streamId); startupTelemetry?.mark('metadata_persisted'); req._resumableStreamId = streamId; @@ -1543,14 +1623,29 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit recoveredSteerCommitted = true; }; - // Send JSON response IMMEDIATELY so client can connect to SSE stream - // This is critical: tool loading (MCP OAuth) may emit events that the client needs to receive - sendGenerationJson( - res, - 200, - { streamId, conversationId, generationCreatedAt: jobCreatedAt, status: 'started' }, - generationProtocolVersion, - ); + // Ordinary clients receive the stream id immediately so they can attach + // before tool loading emits events. Source-owned loopback work delays only + // until its provider invocation and Mongo receipt exist. + let generationStartResponseSent = false; + const sendGenerationStarted = () => { + if (generationStartResponseSent || res.headersSent) { + return; + } + generationStartResponseSent = true; + sendGenerationJson( + res, + 200, + { streamId, conversationId, generationCreatedAt: jobCreatedAt, status: 'started' }, + generationProtocolVersion, + ); + }; + /** Ordinary clients need the stream id before tool discovery. A queued + * source instead keeps its local loopback response open until the provider + * invocation exists, so an accepted HTTP result can never retire text that + * died between job creation and provider startup. */ + if (!hasQueuedTurnAdmissionSource) { + sendGenerationStarted(); + } await attachConversationCreatedAt(req, conversationId, conversationAnchorPromise).then(() => startupTelemetry?.mark('conversation_resolved'), @@ -2298,6 +2393,21 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit }) : client.sendMessage(text, messageOptions); + if (hasQueuedTurnAdmissionSource) { + const { + settleAgentQueuedTurnExecutionAdmission, + } = require('~/server/services/Agents/triggers'); + await settleAgentQueuedTurnExecutionAdmission(queuedTurnAdmissionSource, { + userId, + ...(tenantId != null && { tenantId }), + conversationId, + clientRequestId, + generationId: streamId, + generationCreatedAt: jobCreatedAt, + }); + sendGenerationStarted(); + } + if (titleEligible && titleTiming === 'immediate') { immediateTitlePromise = addTitle(req, { text: text || getAttachmentTitleText(req.body.files), @@ -2780,6 +2890,14 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit job.abortController.signal.removeEventListener('abort', abortTitleOnJobAbort); acceptsTitleEvents = false; resolveConvoReady(); + if (!res.headersSent) { + sendGenerationJson( + res, + 500, + { error: error.message || 'Failed to start generation' }, + generationProtocolVersion, + ); + } try { await recordEventActorPersistenceFailure(error); } catch (reconciliationError) { @@ -2899,6 +3017,14 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit `[ResumableAgentController] Unhandled error in background generation: ${err.message}`, ); startupTelemetry?.end('error', err); + if (!res.headersSent) { + sendGenerationJson( + res, + 500, + { error: err.message || 'Failed to start generation' }, + generationProtocolVersion, + ); + } let errorFinalized = false; if (!pausePersistenceFailed) { errorFinalized = diff --git a/api/server/controllers/agents/steer.js b/api/server/controllers/agents/steer.js index e6c9a0a19a..ecce5b8067 100644 --- a/api/server/controllers/agents/steer.js +++ b/api/server/controllers/agents/steer.js @@ -203,3 +203,4 @@ module.exports = SteerController; module.exports.SteerDeliveryController = SteerDeliveryController; module.exports.SteerCancelController = SteerCancelController; module.exports.SteerArmController = SteerArmController; +module.exports.createAgentAccessCheck = createAgentAccessCheck; diff --git a/api/server/routes/agents/__tests__/idempotencyLimiter.spec.js b/api/server/routes/agents/__tests__/idempotencyLimiter.spec.js index e7281ee2ed..373266a6c0 100644 --- a/api/server/routes/agents/__tests__/idempotencyLimiter.spec.js +++ b/api/server/routes/agents/__tests__/idempotencyLimiter.spec.js @@ -67,6 +67,11 @@ jest.mock('~/server/controllers/agents/steer', () => { controller.SteerArmController = (_req, _res, next) => next(); return controller; }); +jest.mock('~/server/controllers/agents/queuedTurns', () => ({ + AgentQueuedTurnEnqueueController: (_req, res) => res.status(202).json({ queued: true }), + AgentQueuedTurnListController: (_req, res) => res.status(200).json({ queuedTurns: [] }), + AgentQueuedTurnCancelController: (_req, res) => res.status(200).json({ cancelled: true }), +})); jest.mock('~/models', () => ({})); jest.mock('~/server/services/Schedules', () => ({})); @@ -150,4 +155,29 @@ describe('start-generation idempotency before message limiters', () => { expect(mockIpLimiter).not.toHaveBeenCalled(); expect(mockUserLimiter).not.toHaveBeenCalled(); }); + + it('keeps read-only queued-turn polling outside message admission limiters', async () => { + const responses = await Promise.all( + Array.from({ length: 3 }, () => + request(app).get('/agents/chat/queued-turns').query({ conversationId: 'conversation-1' }), + ), + ); + + expect(responses.map((response) => response.status)).toEqual([200, 200, 200]); + expect(responses[0].body).toEqual({ queuedTurns: [] }); + expect(mockIpLimiter).not.toHaveBeenCalled(); + expect(mockUserLimiter).not.toHaveBeenCalled(); + }); + + it.each([ + ['enqueue', () => request(app).post('/agents/chat/queued-turns').send({ text: 'next' })], + ['cancel', () => request(app).delete('/agents/chat/queued-turns/queued-turn-1')], + ])('keeps queued-turn %s mutations behind message admission limiters', async (_label, send) => { + const response = await send(); + + expect(response.status).toBe(429); + expect(response.body).toEqual({ limited: 'ip' }); + expect(mockIpLimiter).toHaveBeenCalledTimes(1); + expect(mockUserLimiter).not.toHaveBeenCalled(); + }); }); diff --git a/api/server/routes/agents/index.js b/api/server/routes/agents/index.js index b6c237708a..1805d9e135 100644 --- a/api/server/routes/agents/index.js +++ b/api/server/routes/agents/index.js @@ -35,6 +35,11 @@ const { messageUserLimiter, } = require('~/server/middleware'); const SteerController = require('~/server/controllers/agents/steer'); +const { + AgentQueuedTurnEnqueueController, + AgentQueuedTurnListController, + AgentQueuedTurnCancelController, +} = require('~/server/controllers/agents/queuedTurns'); const { GENERATION_PROTOCOL_HEADER, GENERATION_PROTOCOL_V2, @@ -228,7 +233,12 @@ router.get('/chat/stream/:streamId', async (req, res) => { logger.warn(`[AgentStream] Refusing stream with invalid generation identity: ${streamId}`); return sendGenerationJson(res, 403, { error: 'Unauthorized' }, requestProtocolVersion); } - const streamTelemetry = createSseStreamTelemetry({ req, res, streamId, isResume }); + const streamTelemetry = createSseStreamTelemetry({ + req, + res, + streamId, + isResume, + }); res.setHeader('Content-Encoding', 'identity'); res.setHeader('Content-Type', 'text/event-stream'); @@ -383,7 +393,9 @@ router.get('/chat/stream/:streamId', async (req, res) => { final: true, reconcile: true, reconcileReason: generationReplaced ? 'generation_replaced' : 'terminal_payload_missing', - ...(expectedGenerationTerminal && { terminalStatus: currentJob.status }), + ...(expectedGenerationTerminal && { + terminalStatus: currentJob.status, + }), generationCreatedAt: authorizedGenerationCreatedAt, conversation: { conversationId: currentJob?.conversationId ?? job.conversationId ?? streamId, @@ -975,7 +987,9 @@ router.post('/chat/abort', configMiddleware, async (req, res, next) => { // 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.persistenceFailed && - abortResult.pendingSteers?.length > 0 && { pendingSteers: abortResult.pendingSteers }), + abortResult.pendingSteers?.length > 0 && { + pendingSteers: abortResult.pendingSteers, + }), }); } @@ -1076,6 +1090,28 @@ router.post( SteerController.SteerArmController, ); +router.post( + '/chat/queued-turns', + configMiddleware, + ...steerLimiters, + createMessageFilterPii({ + getConfig: (req) => req.config?.messageFilter?.pii, + getFilters: (req) => req.config?.filters, + getFiles, + }), + moderateText, + AgentQueuedTurnEnqueueController, +); +/** Synchronizing durable queue state is read-only and polled while work is + * pending. It must not consume the model-submission admission budget. */ +router.get('/chat/queued-turns', configMiddleware, AgentQueuedTurnListController); +router.delete( + '/chat/queued-turns/:queuedTurnId', + configMiddleware, + ...steerLimiters, + AgentQueuedTurnCancelController, +); + router.use('/', v1); const chatRouter = express.Router(); diff --git a/api/server/services/Agents/triggers.js b/api/server/services/Agents/triggers.js index 0d1eea95af..1548511269 100644 --- a/api/server/services/Agents/triggers.js +++ b/api/server/services/Agents/triggers.js @@ -6,10 +6,20 @@ const { SUBAGENT_COMPLETION_SOURCE, createBackgroundToolCompletionWakeupResolver, BACKGROUND_TOOL_COMPLETION_SOURCE, + createAgentQueuedTurnLifecycle, + AGENT_QUEUED_TURN_SOURCE, GenerationJobManager, } = require('@librechat/api'); const methods = require('~/models'); +const getGenerationAdmissionEvidence = (userId, clientRequestId, streamId, conversationId) => + GenerationJobManager.getGenerationAdmissionEvidence( + userId, + clientRequestId, + streamId, + conversationId, + ); + const subagentCompletionAdapter = createSubagentCompletionWakeupResolver({ methods, getGenerationJob: (conversationId) => GenerationJobManager.getJob(conversationId), @@ -22,23 +32,48 @@ const eventActorAdapter = createAgentEventContinueResolver({ methods, getGenerationJob: (conversationId) => GenerationJobManager.getJob(conversationId), }); +let service; +const queuedTurnLifecycle = createAgentQueuedTurnLifecycle({ + methods, + getGenerationJob: (conversationId) => GenerationJobManager.getJob(conversationId), + getGenerationAdmissionEvidence, + enqueue: (...args) => service.enqueue(...args), + retireDelivery: (...args) => service.retire(...args), + getDelivery: (...args) => service.getDelivery(...args), +}); -const service = createAgentTriggerService({ +service = createAgentTriggerService({ methods, isPrincipalActive: methods.isAgentTriggerPrincipalActive, supportsDetachedActionCompletion: () => GenerationJobManager.supportsDetachedAgentEventActions, + settleSourceBeforeDeadLetter: queuedTurnLifecycle.settleBeforeDeadLetter, prepareContinue: createAgentContinuationResolver({ eventActor: eventActorAdapter, internalSources: new Map([ [SUBAGENT_COMPLETION_SOURCE, subagentCompletionAdapter], [BACKGROUND_TOOL_COMPLETION_SOURCE, backgroundToolCompletionAdapter], + [AGENT_QUEUED_TURN_SOURCE, queuedTurnLifecycle.prepareContinue], ]), }), }); +const initializeAgentTriggerService = async (options) => { + await service.initialize(options); + await queuedTurnLifecycle.initialize(); +}; + +const stopAgentTriggerService = async () => { + await queuedTurnLifecycle.stop(); + await service.stop(); +}; + +const purgeAgentTriggerDeliveriesForUser = async (userId) => { + await service.purgeUser(userId); +}; + module.exports = { - initializeAgentTriggerService: service.initialize, - stopAgentTriggerService: service.stop, + initializeAgentTriggerService, + stopAgentTriggerService, dispatchAgentTrigger: service.dispatch, enqueueAgentTrigger: service.enqueue, getAgentTriggerDelivery: service.getDelivery, @@ -50,5 +85,9 @@ module.exports = { drainAgentTriggerDeliveriesForUser: service.drainUser, prepareAgentTriggerUserPurge: service.prepareUserPurge, cancelAgentTriggerUserPurge: service.cancelUserPurge, - purgeAgentTriggerDeliveriesForUser: service.purgeUser, + purgeAgentTriggerDeliveriesForUser, + scheduleAgentQueuedTurn: queuedTurnLifecycle.schedule, + cancelAgentQueuedTurn: queuedTurnLifecycle.cancel, + settleAgentQueuedTurnExecutionAdmission: queuedTurnLifecycle.recordExecutionAdmission, + verifyAgentQueuedTurnExecutionAdmission: queuedTurnLifecycle.verifyExecutionAdmission, }; diff --git a/api/server/services/Agents/triggers.spec.js b/api/server/services/Agents/triggers.spec.js index 48491d402b..230ae910cf 100644 --- a/api/server/services/Agents/triggers.spec.js +++ b/api/server/services/Agents/triggers.spec.js @@ -2,6 +2,17 @@ const mockCreateAgentTriggerService = jest.fn(); const mockGenerationJobManager = { supportsDetachedAgentEventActions: true, getJob: jest.fn(), + getGenerationAdmissionEvidence: jest.fn(), +}; +const mockQueuedTurnLifecycle = { + prepareContinue: jest.fn(), + settleBeforeDeadLetter: jest.fn(), + recordExecutionAdmission: jest.fn(), + verifyExecutionAdmission: jest.fn(), + initialize: jest.fn(), + stop: jest.fn(), + schedule: jest.fn(), + cancel: jest.fn(), }; jest.mock('@librechat/api', () => ({ @@ -10,8 +21,10 @@ jest.mock('@librechat/api', () => ({ createAgentEventContinueResolver: jest.fn(() => jest.fn()), createBackgroundToolCompletionWakeupResolver: jest.fn(() => jest.fn()), createSubagentCompletionWakeupResolver: jest.fn(() => jest.fn()), + createAgentQueuedTurnLifecycle: jest.fn(() => mockQueuedTurnLifecycle), BACKGROUND_TOOL_COMPLETION_SOURCE: 'background-tool-completion', SUBAGENT_COMPLETION_SOURCE: 'subagent-completion', + AGENT_QUEUED_TURN_SOURCE: 'agent-queued-turn', GenerationJobManager: mockGenerationJobManager, })); diff --git a/client/src/components/Chat/Input/PendingSteerChips.tsx b/client/src/components/Chat/Input/PendingSteerChips.tsx index a2d423886b..eef0566152 100644 --- a/client/src/components/Chat/Input/PendingSteerChips.tsx +++ b/client/src/components/Chat/Input/PendingSteerChips.tsx @@ -2,7 +2,18 @@ import { memo, useMemo, useRef, useState, useCallback } from 'react'; import { useAtomValue } from 'jotai'; import { useRecoilValue } from 'recoil'; import { useToastContext } from '@librechat/client'; -import { X, Zap, Send, Clock, Pencil, Trash2, Paperclip, RotateCcw, TextQuote } from 'lucide-react'; +import { + X, + Zap, + Send, + Clock, + Pencil, + Trash2, + Paperclip, + RotateCcw, + TextQuote, + TriangleAlert, +} from 'lucide-react'; import type { TMessage } from 'librechat-data-provider'; import type { SteeringControls, QueuedMessageContext } from '~/hooks/Chat/useSteering'; import type { PendingSteer, QueuedMessage } from '~/store/families'; @@ -91,6 +102,14 @@ function QueuedRow({ const fileCount = message.files?.length ?? 0; const quoteCount = message.quotes?.length ?? 0; const isRecovered = message.recoverySteerId != null; + const isRejected = message.server?.status === 'rejected'; + const isUnconfirmed = + message.server?.status === 'uncertain' && message.server.reconciliationExpired === true; + const requiresDiscard = isRecovered || message.server?.id != null; + const serverActionable = + message.server == null || + message.server.status === 'rejected' || + (message.server.id != null && message.server.status === 'queued'); const actionPendingRef = useRef(false); const [actionPending, setActionPending] = useState(false); /** A recovered item has a replayable parked source. Edit/remove must first @@ -128,7 +147,8 @@ function QueuedRow({ // generation. Re-steering it would leave or duplicate the parked source; // Edit/remove are safe because `afterDiscard` tombstones that source first. const canSteerNow = steering.duringRunActive && steering.canSteer && !isRecovered; - const showPrimary = canSteerNow || (!steering.duringRunActive && steering.canSendQueuedNow); + const showPrimary = + serverActionable && (canSteerNow || (!steering.duringRunActive && steering.canSendQueuedNow)); /** `canSteer` is defined as false while paused on approval, but the * escalation control must stay visible-and-disabled there — hiding it * during the pause is exactly the discoverability gap this button fixes. */ @@ -140,10 +160,13 @@ function QueuedRow({ key: 'edit', label: localize('com_ui_edit_message'), icon: