diff --git a/.env.example b/.env.example index 9e66dbeb10..b63c9605c2 100644 --- a/.env.example +++ b/.env.example @@ -1158,6 +1158,9 @@ HELP_AND_FAQ_URL=https://librechat.ai # USE_REDIS_STREAMS=true # Scheduled chats require shared Redis streams in multi-replica deployments. # Set this only when the deployment truly runs one LibreChat process without Redis. +# Scheduled agents that can pause for approval or ask_user_question require both +# Redis streams for shared action state and a durable shared checkpointer. +# MongoDB is currently the built-in durable checkpointer (and the default). # SCHEDULES_SINGLE_PROCESS=true # Emergency global stop for both automatic and manual scheduled runs. # SCHEDULES_DISABLED=true diff --git a/api/server/controllers/agents/__tests__/resume.spec.js b/api/server/controllers/agents/__tests__/resume.spec.js index cb41952e63..dddf10c4cd 100644 --- a/api/server/controllers/agents/__tests__/resume.spec.js +++ b/api/server/controllers/agents/__tests__/resume.spec.js @@ -382,6 +382,7 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => { files: req.body.files, isTemporary: req.body.isTemporary, conversationCreatedAt: req.conversationCreatedAt, + isScheduledFire: req._isScheduledFire, timezone: req.body.timezone, checkpointNamespace, requestBody, @@ -722,6 +723,8 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => { expect(res.status).toBe(200); await settled; + expect(capturedInit.isScheduledFire).toBe(true); + expect(mockClaimScheduleResume).toHaveBeenCalledWith('schedule-1', scheduledFor, { expectedConfigRevision: 4, automatic: true, diff --git a/api/server/controllers/agents/client.js b/api/server/controllers/agents/client.js index c216e4dfcb..5fe5d1a3eb 100644 --- a/api/server/controllers/agents/client.js +++ b/api/server/controllers/agents/client.js @@ -32,6 +32,7 @@ const { createSubagentUsageSink, anyAgentReplaysReasoningContent, GenerationJobManager, + PENDING_ACTION_EXPIRED_CODE, getTransactionsConfig, resolveRecursionLimit, buildPendingAction, @@ -42,13 +43,18 @@ const { pickResumeContext, getApprovalTtlMs, getAgentCheckpointer, + hasDurableAgentInterruptCheckpoint, isHITLEnabled, + buildToolApprovalHooks, + agentRunUsesCheckpointer, + canAgentGraphPause, + getPluginHookSource, captureAgentCheckpointGeneration, isContentFilterError, deleteAgentCheckpoint, LIBRECHAT_CHECKPOINT_NAMESPACE_KEY, LIBRECHAT_EVENT_ACTOR_INVOCATION_KEY, - agentRequestsAskUserQuestion, + isAskUserQuestionAdminDisabled, attachAskUserQuestionArgs, hydrateResumeRunSteps, createContentIndexOffsetHandlers, @@ -154,19 +160,6 @@ const db = require('~/models'); const loadAgent = (params) => loadAgentFn(params, { getAgent: db.getAgent, getMCPServerTools }); -function getInterruptTtlMs(checkpointerCfg, req) { - const configuredTtlMs = getApprovalTtlMs(checkpointerCfg); - const retention = req?._agentEventBindingRetention; - if (retention?.expiredAt == null) { - return configuredTtlMs; - } - const bindingDeadline = new Date(retention.expiredAt).getTime(); - if (!Number.isFinite(bindingDeadline)) { - return configuredTtlMs; - } - return Math.min(configuredTtlMs, Math.max(0, bindingDeadline - Date.now())); -} - const MEMORY_INPUT_CHARS_PER_TOKEN = 8; function getUserFacingRequestError(baseMessage, error, appConfig) { @@ -3130,6 +3123,43 @@ class AgentClient extends BaseClient { const appConfig = this.options.req?.config; const checkpointerCfg = appConfig?.endpoints?.[EModelEndpoint.agents]?.checkpointer; + if (this.options.req?._isScheduledFire === true) { + if (!GenerationJobManager.isRedis) { + const error = new Error( + 'The agent paused, but its shared action state is unavailable. Please retry the run.', + ); + error.code = 'SCHEDULED_HITL_REQUIRES_SHARED_STORE'; + throw error; + } + let hasDurableInterrupt = false; + try { + hasDurableInterrupt = await hasDurableAgentInterruptCheckpoint( + this.conversationId, + checkpointerCfg, + { + checkpointNamespace: this.checkpointNamespace, + checkpointId: interrupt.checkpointId, + checkpointNs: interrupt.checkpointNs, + interruptId: interrupt.interruptId, + }, + ); + } catch (checkpointError) { + logger.error( + `[AgentClient] Failed to verify scheduled HITL checkpoint for ${this.conversationId} (${this.checkpointNamespace || 'legacy namespace'})`, + checkpointError, + ); + } + if (!hasDurableInterrupt) { + logger.error( + `[AgentClient] Refusing unresumable scheduled HITL pause for ${this.conversationId} (${this.checkpointNamespace || 'legacy namespace'})`, + ); + const error = new Error( + 'The agent paused, but its durable continuation checkpoint is unavailable. Please retry the run.', + ); + error.code = 'HITL_CHECKPOINT_UNAVAILABLE'; + throw error; + } + } // Persist the generation params (temperature, max tokens, custom endpoint params, …) // so an ephemeral-agent resume continues with the SAME settings the run paused on. // The resume payload omits them and they aren't part of the fingerprint, so without @@ -3174,7 +3204,8 @@ class AgentClient extends BaseClient { // thread_id was bound to conversationId at run config (config.configurable); // fall back to it when the SDK doesn't echo threadId on the interrupt. threadId: interrupt.threadId ?? this.conversationId, - ttlMs: getInterruptTtlMs(checkpointerCfg, this.options.req), + ttlMs: getApprovalTtlMs(checkpointerCfg), + expiresAt: this.options.req?._agentEventBindingRetention?.expiredAt, // Pin the graph-determining request fields so resume can't rebuild this paused // run on a different agent/tool set (esp. ephemeral agents, whose agent_id is // undefined so the id guard can't tell two configs apart). @@ -3281,6 +3312,57 @@ class AgentClient extends BaseClient { abortController = new AbortController(); } + /** Scheduled approvals are unattended by definition. Their pending action, + * replay context, and resolution fence must be shared across workers; their + * LangGraph continuation must also use a durable shared checkpointer. Redis + * provides the first half, while handleRunInterrupt verifies the exact durable + * checkpoint before exposing the pause. Refuse unsupported topologies before + * spending on provider work. */ + /** @type {AppConfig['endpoints']['agents']} */ + const agentsEConfig = appConfig.endpoints?.[EModelEndpoint.agents]; + const resolvedToolApprovalHooks = isHITLEnabled(agentsEConfig?.toolApproval) + ? buildToolApprovalHooks({ + userId: this.options.req?.user?.id, + conversationId: this.conversationId, + tenantId: this.options.req?.user?.tenantId, + appConfig, + }) + : undefined; + const topLevelAgents = [this.options.agent, ...(this.agentConfigs?.values() ?? [])]; + const askUserQuestionAdminDisabled = isAskUserQuestionAdminDisabled(appConfig); + const runCanPause = canAgentGraphPause({ + policy: agentsEConfig?.toolApproval, + agents: topLevelAgents, + hostGeneratedToolNames: + this.options.subagentTasks == null ? undefined : [Constants.CHECK_BACKGROUND_TASK], + resolvedProgrammaticHooks: resolvedToolApprovalHooks, + pluginHookSource: getPluginHookSource(), + askUserQuestionAdminDisabled, + }); + const runUsesCheckpointer = agentRunUsesCheckpointer({ + policy: agentsEConfig?.toolApproval, + agents: topLevelAgents, + askUserQuestionAdminDisabled, + }); + if (this.options.req?._isScheduledFire === true && runCanPause) { + if (!GenerationJobManager.isRedis) { + const error = new Error( + 'Scheduled agent runs that can pause require a shared generation store. ' + + 'Enable Redis streams with USE_REDIS_STREAMS=true.', + ); + error.code = 'SCHEDULED_HITL_REQUIRES_SHARED_STORE'; + throw error; + } + if (!(await getAgentCheckpointer(agentsEConfig?.checkpointer))) { + const error = new Error( + 'Scheduled agent runs that can pause require a durable shared checkpointer. ' + + 'Use the default MongoDB checkpointer.', + ); + error.code = 'SCHEDULED_HITL_REQUIRES_DURABLE_CHECKPOINT'; + throw error; + } + } + /** Fire-and-forget: boot each selected stateful environment in * parallel with generation so the first execute_code/bash call lands * on a warm VM. No-op unless a reachable agent resolved @@ -3291,9 +3373,6 @@ class AgentClient extends BaseClient { agents: [this.options.agent, ...(this.agentConfigs?.values() ?? [])], }); - /** @type {AppConfig['endpoints']['agents']} */ - const agentsEConfig = appConfig.endpoints?.[EModelEndpoint.agents]; - config = { runName: 'AgentRun', configurable: { @@ -3555,16 +3634,14 @@ class AgentClient extends BaseClient { // HITL is off or the generation has no remnants. Deliberately unconditional // per HITL turn: any cheaper Redis flag can go stale across replicas/restarts, // while these are two indexed, usually-empty deleteMany operations. - // The gate mirrors createRun's checkpointer condition: the approval policy - // OR an ask_user_question-capable agent (which attaches a checkpointer - // WITHOUT the approval policy). + // Mirror createRun's checkpointer attachment gate. This is deliberately + // broader than pause admission: retries must prune remnants even when a + // policy or request hook changed from pausing to non-pausing. // // Start the prune alongside graph construction. The all-settled barrier // below still guarantees it completes before the graph is exposed or run. const shouldPruneCheckpoint = - streamId && - this.eventActorInvocationId == null && - (isHITLEnabled(agentsEConfig?.toolApproval) || agents.some(agentRequestsAskUserQuestion)); + streamId && this.eventActorInvocationId == null && runUsesCheckpointer; let checkpointPrunePromise = Promise.resolve(); if (shouldPruneCheckpoint && this.checkpointNamespace !== '') { checkpointPrunePromise = deleteAgentCheckpoint( @@ -3627,6 +3704,7 @@ 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, + resolvedToolApprovalHooks, toolInputValidationErrors: this.toolInputValidationErrors, // Mid-run steering: drain queued user messages at each tool-batch // boundary and inject them into graph state. The offset wrapper @@ -3792,6 +3870,15 @@ class AgentClient extends BaseClient { this.applyHideSequentialOutputsFilter(); this.rebaseActivityPhaseBounds(contentBeforeReshape); } catch (err) { + if ( + err?.code === 'SCHEDULED_HITL_REQUIRES_SHARED_STORE' || + err?.code === 'SCHEDULED_HITL_REQUIRES_DURABLE_CHECKPOINT' || + err?.code === 'HITL_CHECKPOINT_UNAVAILABLE' || + err?.code === PENDING_ACTION_EXPIRED_CODE + ) { + logger.warn(`[api/server/controllers/agents/client.js #sendCompletion] ${err.message}`); + throw err; + } if (isContentFilterError(err)) { logger.warn( '[api/server/controllers/agents/client.js #sendCompletion] Blocked by content policy', @@ -3953,6 +4040,14 @@ class AgentClient extends BaseClient { /** @type {AppConfig['endpoints']['agents']} */ const agentsEConfig = appConfig.endpoints?.[EModelEndpoint.agents]; + const resolvedToolApprovalHooks = isHITLEnabled(agentsEConfig?.toolApproval) + ? buildToolApprovalHooks({ + userId: this.options.req?.user?.id, + conversationId: this.conversationId, + tenantId: this.options.req?.user?.tenantId, + appConfig, + }) + : undefined; BaseClient.prototype.setModelBoundStoredMessages.call( this, @@ -4109,6 +4204,7 @@ 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, + resolvedToolApprovalHooks, // Plugin SessionStart hooks match on the lifecycle source; a rebuilt run is a // resume, not a fresh startup. sessionStartSource: 'resume', diff --git a/api/server/controllers/agents/client.test.js b/api/server/controllers/agents/client.test.js index 14c3a0cd19..4cc105219b 100644 --- a/api/server/controllers/agents/client.test.js +++ b/api/server/controllers/agents/client.test.js @@ -6,6 +6,7 @@ const mockRecordCollectedUsage = jest.fn(); const mockDetachedUsageRecorder = jest.fn(); const mockCreateDetachedSubagentUsageRecorder = jest.fn(() => mockDetachedUsageRecorder); const mockGetAgentCheckpointer = jest.fn(); +const mockHasDurableAgentInterruptCheckpoint = jest.fn().mockResolvedValue(true); const mockBuildAgentScopedContext = jest.fn((...args) => jest.requireActual('@librechat/api').buildAgentScopedContext(...args), ); @@ -18,7 +19,14 @@ const mockFormatAgentMessages = jest.fn(() => ({ const { Providers } = require('@librechat/agents'); const { Constants, ContentTypes, EModelEndpoint } = require('librechat-data-provider'); -const { GenerationJobManager, createStreamServices } = require('@librechat/api'); +const { + GenerationJobManager, + createStreamServices, + registerToolApprovalHook, + clearToolApprovalHooks, + getPluginHookSource, + setPluginHookSource, +} = require('@librechat/api'); const BaseClient = require('~/app/clients/BaseClient'); const AgentClient = require('./client'); const { resolveConfigServers } = require('~/server/services/MCP'); @@ -66,6 +74,7 @@ jest.mock('@librechat/api', () => ({ maybePrewarmCodeSandbox: jest.fn(), recordCollectedUsage: (...args) => mockRecordCollectedUsage(...args), getAgentCheckpointer: mockGetAgentCheckpointer, + hasDurableAgentInterruptCheckpoint: (...args) => mockHasDurableAgentInterruptCheckpoint(...args), })); describe('AgentClient - final model-bound content protection', () => { @@ -463,6 +472,7 @@ describe('AgentClient - reasoning label accounting', () => { describe('AgentClient - interrupt discovery persistence', () => { beforeEach(async () => { + mockHasDurableAgentInterruptCheckpoint.mockClear(); await GenerationJobManager.destroy(); GenerationJobManager.configure({ ...createStreamServices(), cleanupOnComplete: false }); GenerationJobManager.initialize(); @@ -567,6 +577,157 @@ describe('AgentClient - interrupt discovery persistence', () => { expect(paused?.metadata.pendingAction.expiresAt).toBeGreaterThanOrEqual(now + 4_900); expect(paused?.metadata.pendingAction.expiresAt).toBeLessThanOrEqual(now + 5_000); }); + + it('does not expose an event-bound pause after its inherited deadline', async () => { + const streamId = 'conversation-expired-event-bound-pause'; + const job = await GenerationJobManager.createJob(streamId, 'user-123', streamId); + const client = new AgentClient({ + req: { + user: { id: 'user-123' }, + body: { endpoint: EModelEndpoint.agents, agent_id: 'agent-123' }, + config: { endpoints: { [EModelEndpoint.agents]: { checkpointer: { ttl: 3600 } } } }, + _agentEventBindingRetention: { + isTemporary: false, + expiredAt: new Date(Date.now() - 1), + }, + }, + res: {}, + agent: { + id: 'agent-123', + endpoint: EModelEndpoint.openAI, + provider: EModelEndpoint.openAI, + model_parameters: { model: 'gpt-4' }, + }, + contentParts: [], + collectedUsage: [], + artifactPromises: [], + }); + client.conversationId = streamId; + client.responseMessageId = 'response-expired-event-bound-pause'; + client.jobCreatedAt = job.createdAt; + + await expect( + client.handleRunInterrupt( + { + getInterrupt: () => ({ + interruptId: 'ask-interrupt', + threadId: streamId, + payload: { + type: 'ask_user_question', + question: { question: 'Proceed?' }, + }, + }), + }, + streamId, + ), + ).rejects.toMatchObject({ code: 'HITL_ACTION_EXPIRED' }); + const liveJob = await GenerationJobManager.getJob(streamId); + expect(liveJob?.status).toBe('running'); + expect(liveJob?.metadata.pendingAction).toBeUndefined(); + }); + + it('does not expose a scheduled pause without its shared action store', async () => { + const streamId = 'scheduled-missing-shared-store'; + const job = await GenerationJobManager.createJob(streamId, 'user-123', streamId); + const client = new AgentClient({ + req: { + user: { id: 'user-123' }, + body: { endpoint: EModelEndpoint.agents, agent_id: 'agent-123' }, + config: { endpoints: { [EModelEndpoint.agents]: {} } }, + _isScheduledFire: true, + }, + res: {}, + agent: { + id: 'agent-123', + endpoint: EModelEndpoint.openAI, + provider: EModelEndpoint.openAI, + model_parameters: { model: 'gpt-4' }, + }, + contentParts: [], + collectedUsage: [], + artifactPromises: [], + }); + client.conversationId = streamId; + client.responseMessageId = 'scheduled-missing-shared-store-response'; + client.jobCreatedAt = job.createdAt; + client.checkpointNamespace = job.metadata.checkpointNamespace; + + await expect( + client.handleRunInterrupt( + { + getInterrupt: () => ({ + interruptId: 'ask-interrupt', + threadId: streamId, + payload: { + type: 'ask_user_question', + question: { question: 'Proceed?' }, + }, + }), + }, + streamId, + ), + ).rejects.toMatchObject({ code: 'SCHEDULED_HITL_REQUIRES_SHARED_STORE' }); + expect(mockHasDurableAgentInterruptCheckpoint).not.toHaveBeenCalled(); + await expect(GenerationJobManager.getJobStatus(streamId)).resolves.toBe('running'); + }); + + it('does not expose a scheduled pause without its durable interrupt checkpoint', async () => { + const isRedisSpy = jest.spyOn(GenerationJobManager, 'isRedis', 'get').mockReturnValue(true); + const streamId = 'scheduled-missing-interrupt-checkpoint'; + const job = await GenerationJobManager.createJob(streamId, 'user-123', streamId); + const client = new AgentClient({ + req: { + user: { id: 'user-123' }, + body: { endpoint: EModelEndpoint.agents, agent_id: 'agent-123' }, + config: { endpoints: { [EModelEndpoint.agents]: {} } }, + _isScheduledFire: true, + }, + res: {}, + agent: { + id: 'agent-123', + endpoint: EModelEndpoint.openAI, + provider: EModelEndpoint.openAI, + model_parameters: { model: 'gpt-4' }, + }, + contentParts: [], + collectedUsage: [], + artifactPromises: [], + }); + client.conversationId = streamId; + client.responseMessageId = 'scheduled-missing-checkpoint-response'; + client.jobCreatedAt = job.createdAt; + client.checkpointNamespace = job.metadata.checkpointNamespace; + mockHasDurableAgentInterruptCheckpoint.mockResolvedValueOnce(false); + + try { + await expect( + client.handleRunInterrupt( + { + getInterrupt: () => ({ + interruptId: 'ask-interrupt', + checkpointId: 'checkpoint-current', + checkpointNs: 'nested-agent', + threadId: streamId, + payload: { + type: 'ask_user_question', + question: { question: 'Proceed?' }, + }, + }), + }, + streamId, + ), + ).rejects.toMatchObject({ code: 'HITL_CHECKPOINT_UNAVAILABLE' }); + expect(mockHasDurableAgentInterruptCheckpoint).toHaveBeenCalledWith(streamId, undefined, { + checkpointNamespace: job.metadata.checkpointNamespace, + checkpointId: 'checkpoint-current', + checkpointNs: 'nested-agent', + interruptId: 'ask-interrupt', + }); + await expect(GenerationJobManager.getJobStatus(streamId)).resolves.toBe('running'); + } finally { + isRedisSpy.mockRestore(); + } + }); }); jest.mock('~/server/services/Config', () => ({ @@ -817,9 +978,280 @@ describe('AgentClient - activity phase completion', () => { describe('AgentClient - startup telemetry', () => { afterEach(() => { + clearToolApprovalHooks(); + mockIsHITLEnabled.mockReturnValue(false); jest.restoreAllMocks(); }); + it('refuses scheduled pause-capable runs before provider startup without Redis', async () => { + mockIsHITLEnabled.mockReturnValue(false); + const createRunBefore = mockCreateRun.mock.calls.length; + const client = new AgentClient({ + req: { + user: { id: 'user-123' }, + body: {}, + config: { endpoints: { [EModelEndpoint.agents]: {} } }, + _isScheduledFire: true, + _resumableStreamId: 'scheduled-hitl-no-redis', + }, + res: {}, + agent: { + id: 'agent-123', + endpoint: EModelEndpoint.openAI, + provider: EModelEndpoint.openAI, + model_parameters: { model: 'gpt-4' }, + tools: [{ name: 'ask_user_question' }], + }, + endpointTokenConfig: {}, + eventHandlers: {}, + contentParts: [], + collectedUsage: [], + artifactPromises: [], + }); + client.conversationId = 'scheduled-hitl-no-redis'; + client.responseMessageId = 'scheduled-hitl-response'; + client.parentMessageId = 'scheduled-hitl-parent'; + + await expect(client.chatCompletion({ payload: [] })).rejects.toMatchObject({ + code: 'SCHEDULED_HITL_REQUIRES_SHARED_STORE', + message: expect.stringContaining('USE_REDIS_STREAMS=true'), + }); + expect(mockCreateRun).toHaveBeenCalledTimes(createRunBefore); + }); + + it('includes host-generated background tools in scheduled pause admission', async () => { + const createRunBefore = mockCreateRun.mock.calls.length; + const client = new AgentClient({ + req: { + user: { id: 'user-123' }, + body: {}, + config: { + endpoints: { + [EModelEndpoint.agents]: { + toolApproval: { + enabled: true, + mode: 'bypass', + ask: ['check_background_task'], + }, + }, + }, + }, + _isScheduledFire: true, + _resumableStreamId: 'scheduled-background-tool', + }, + res: {}, + agent: { + id: 'agent-123', + endpoint: EModelEndpoint.openAI, + provider: EModelEndpoint.openAI, + model_parameters: { model: 'gpt-4' }, + tools: [{ name: 'read_file' }], + }, + subagentTasks: {}, + endpointTokenConfig: {}, + eventHandlers: {}, + contentParts: [], + collectedUsage: [], + artifactPromises: [], + }); + client.conversationId = 'scheduled-background-tool'; + client.responseMessageId = 'scheduled-background-response'; + client.parentMessageId = 'scheduled-background-parent'; + + await expect(client.chatCompletion({ payload: [] })).rejects.toMatchObject({ + code: 'SCHEDULED_HITL_REQUIRES_SHARED_STORE', + }); + expect(mockCreateRun).toHaveBeenCalledTimes(createRunBefore); + }); + + it.each([ + { + name: 'a bypass-only approval policy', + toolApproval: { enabled: true, mode: 'bypass' }, + subagentAgentConfigs: undefined, + }, + { + name: 'an approval rule that matches no selected tool', + toolApproval: { enabled: true, mode: 'bypass', ask: ['approval_probe'] }, + subagentAgentConfigs: undefined, + }, + { + name: 'ask_user_question denied by the approval policy', + toolApproval: { enabled: true, deny: ['ask_*'] }, + primaryTools: [{ name: 'ask_user_question' }], + subagentAgentConfigs: undefined, + }, + { + name: 'ask_user_question on a nested subagent only', + toolApproval: undefined, + subagentAgentConfigs: [ + { + id: 'nested-agent', + tools: [{ name: 'ask_user_question' }], + }, + ], + }, + ])( + 'does not reject scheduled runs for $name', + async ({ toolApproval, primaryTools, subagentAgentConfigs }) => { + mockDeleteAgentCheckpoint.mockReset().mockResolvedValue(undefined); + const processStream = jest.fn().mockResolvedValue(); + mockCreateRun.mockResolvedValueOnce({ + Graph: null, + processStream, + getCalibrationRatio: jest.fn(() => 0), + getInterrupt: jest.fn(() => undefined), + }); + const createRunBefore = mockCreateRun.mock.calls.length; + const client = new AgentClient({ + req: { + user: { id: 'user-123' }, + body: {}, + config: { endpoints: { [EModelEndpoint.agents]: { toolApproval } } }, + _isScheduledFire: true, + _resumableStreamId: 'scheduled-non-pausing-policy', + }, + res: {}, + agent: { + id: 'agent-123', + endpoint: EModelEndpoint.openAI, + provider: EModelEndpoint.openAI, + model_parameters: { model: 'gpt-4' }, + hide_sequential_outputs: false, + tools: primaryTools ?? [{ name: 'read_file' }], + subagentAgentConfigs, + }, + endpointTokenConfig: {}, + eventHandlers: {}, + contentParts: [], + collectedUsage: [], + artifactPromises: [], + }); + client.conversationId = 'scheduled-non-pausing-policy'; + client.checkpointNamespace = 'scheduled-non-pausing-generation'; + client.responseMessageId = 'scheduled-non-pausing-response'; + client.parentMessageId = 'scheduled-non-pausing-parent'; + client.recordCollectedUsage = jest.fn().mockResolvedValue(); + + await expect(client.chatCompletion({ payload: [] })).resolves.toBeUndefined(); + expect(mockCreateRun).toHaveBeenCalledTimes(createRunBefore + 1); + expect(processStream).toHaveBeenCalledTimes(1); + if (toolApproval?.enabled === true) { + expect(mockDeleteAgentCheckpoint).toHaveBeenCalledWith( + 'scheduled-non-pausing-policy', + undefined, + undefined, + { + throwOnError: true, + checkpointNamespace: 'scheduled-non-pausing-generation', + }, + ); + } else { + expect(mockDeleteAgentCheckpoint).not.toHaveBeenCalled(); + } + }, + ); + + it('uses request-scoped hook resolution when deciding whether a scheduled run can pause', async () => { + mockIsHITLEnabled.mockReturnValue(true); + registerToolApprovalHook((context) => + context.userId === 'different-user' ? async () => ({ decision: 'ask' }) : undefined, + ); + const processStream = jest.fn().mockResolvedValue(); + mockCreateRun.mockResolvedValueOnce({ + Graph: null, + processStream, + getCalibrationRatio: jest.fn(() => 0), + getInterrupt: jest.fn(() => undefined), + }); + const client = new AgentClient({ + req: { + user: { id: 'user-123' }, + body: {}, + config: { + endpoints: { + [EModelEndpoint.agents]: { + toolApproval: { enabled: true, mode: 'bypass' }, + }, + }, + }, + _isScheduledFire: true, + _resumableStreamId: 'scheduled-request-scoped-hook', + }, + res: {}, + agent: { + id: 'agent-123', + endpoint: EModelEndpoint.openAI, + provider: EModelEndpoint.openAI, + model_parameters: { model: 'gpt-4' }, + hide_sequential_outputs: false, + tools: [{ name: 'read_file' }], + }, + endpointTokenConfig: {}, + eventHandlers: {}, + contentParts: [], + collectedUsage: [], + artifactPromises: [], + }); + client.conversationId = 'scheduled-request-scoped-hook'; + client.checkpointNamespace = 'scheduled-request-scoped-generation'; + client.responseMessageId = 'scheduled-request-scoped-response'; + client.parentMessageId = 'scheduled-request-scoped-parent'; + client.recordCollectedUsage = jest.fn().mockResolvedValue(); + + await expect(client.chatCompletion({ payload: [] })).resolves.toBeUndefined(); + expect(mockCreateRun.mock.calls.at(-1)?.[0]?.resolvedToolApprovalHooks).toEqual([]); + }); + + it('admits deployment PreToolUse hooks as pause-capable for scheduled runs', async () => { + mockIsHITLEnabled.mockReturnValue(true); + const previousSource = getPluginHookSource(); + setPluginHookSource({ + hasHooks: () => true, + hasToolApprovalHooks: () => true, + register: () => 1, + }); + const client = new AgentClient({ + req: { + user: { id: 'user-123' }, + body: {}, + config: { + endpoints: { + [EModelEndpoint.agents]: { + toolApproval: { enabled: true, mode: 'bypass' }, + }, + }, + }, + _isScheduledFire: true, + _resumableStreamId: 'scheduled-deployment-hook', + }, + res: {}, + agent: { + id: 'agent-123', + endpoint: EModelEndpoint.openAI, + provider: EModelEndpoint.openAI, + model_parameters: { model: 'gpt-4' }, + tools: [{ name: 'write_file' }], + }, + endpointTokenConfig: {}, + eventHandlers: {}, + contentParts: [], + collectedUsage: [], + artifactPromises: [], + }); + client.conversationId = 'scheduled-deployment-hook'; + client.responseMessageId = 'scheduled-deployment-response'; + client.parentMessageId = 'scheduled-deployment-parent'; + + try { + await expect(client.chatCompletion({ payload: [] })).rejects.toMatchObject({ + code: 'SCHEDULED_HITL_REQUIRES_SHARED_STORE', + }); + } finally { + setPluginHookSource(previousSource); + } + }); + it('overlaps run creation with checkpoint pruning and joins both before stream processing', async () => { let releaseCheckpoint; let checkpointStarted; @@ -853,7 +1285,9 @@ describe('AgentClient - startup telemetry', () => { req: { user: { id: 'user-123' }, body: {}, - config: { endpoints: { [EModelEndpoint.agents]: { toolApproval: {} } } }, + config: { + endpoints: { [EModelEndpoint.agents]: { toolApproval: { enabled: true } } }, + }, _resumableStreamId: 'conversation-123', }, res: {}, @@ -863,6 +1297,7 @@ describe('AgentClient - startup telemetry', () => { provider: EModelEndpoint.openAI, model_parameters: { model: 'gpt-4' }, hide_sequential_outputs: false, + tools: [{ name: 'write_file' }], }, endpointTokenConfig: {}, eventHandlers: {}, @@ -1094,7 +1529,9 @@ describe('AgentClient - startup telemetry', () => { req: { user: { id: 'user-123' }, body: {}, - config: { endpoints: { [EModelEndpoint.agents]: { toolApproval: {} } } }, + config: { + endpoints: { [EModelEndpoint.agents]: { toolApproval: { enabled: true } } }, + }, _resumableStreamId: 'conversation-123', }, res: {}, @@ -1104,6 +1541,7 @@ describe('AgentClient - startup telemetry', () => { provider: EModelEndpoint.openAI, model_parameters: { model: 'gpt-4' }, hide_sequential_outputs: false, + tools: [{ name: 'write_file' }], }, endpointTokenConfig: {}, eventHandlers: {}, @@ -1168,7 +1606,9 @@ describe('AgentClient - startup telemetry', () => { req: { user: { id: 'user-123' }, body: {}, - config: { endpoints: { [EModelEndpoint.agents]: { toolApproval: {} } } }, + config: { + endpoints: { [EModelEndpoint.agents]: { toolApproval: { enabled: true } } }, + }, _resumableStreamId: 'conversation-123', }, res: {}, @@ -1178,6 +1618,7 @@ describe('AgentClient - startup telemetry', () => { provider: EModelEndpoint.openAI, model_parameters: { model: 'gpt-4' }, hide_sequential_outputs: false, + tools: [{ name: 'write_file' }], }, endpointTokenConfig: {}, eventHandlers: {}, diff --git a/api/server/controllers/agents/resume.js b/api/server/controllers/agents/resume.js index 0cd6e34b5a..7b8801d5ae 100644 --- a/api/server/controllers/agents/resume.js +++ b/api/server/controllers/agents/resume.js @@ -712,6 +712,13 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle) const scheduleId = job.metadata?.scheduleId; const scheduledFor = job.metadata?.scheduledFor; + // A resumed schedule is still an unattended scheduled run. Preserve that + // provenance on the rebuilt client so every later pause (a second approval + // or follow-up question) must prove its durable checkpoint before the job is + // exposed as `requires_action` again. + if (scheduleId) { + req._isScheduledFire = true; + } if ( scheduleId && !(await isScheduleLive(scheduleId, job.metadata?.scheduleConfigRevision, { diff --git a/api/server/index.js b/api/server/index.js index 795ca8ac6f..8796f3dbca 100644 --- a/api/server/index.js +++ b/api/server/index.js @@ -38,6 +38,7 @@ const { getDeploymentPluginHookCapabilities, registerDeploymentPluginHooks, hasDeploymentPluginHooks, + hasDeploymentPluginToolApprovalHooks, setPluginHookSource, loadToolApprovalHooks, maybeInjectQueryDevtoolsBootstrap, @@ -220,6 +221,7 @@ const startServer = async () => { // agents -> plugins import (see agents/hooks/source.ts). setPluginHookSource({ hasHooks: hasDeploymentPluginHooks, + hasToolApprovalHooks: hasDeploymentPluginToolApprovalHooks, register: registerDeploymentPluginHooks, }); await initializeDeploymentSkills({ diff --git a/librechat.example.yaml b/librechat.example.yaml index e5d63e3053..f0eef9fc6b 100644 --- a/librechat.example.yaml +++ b/librechat.example.yaml @@ -200,6 +200,9 @@ interface: # A deployment with Redis-backed resumable streams can run this on every replica. # A single-process deployment without Redis must also set # `SCHEDULES_SINGLE_PROCESS=true`; unsafe multi-replica writes fail closed. + # Scheduled agents that can pause require both `USE_REDIS_STREAMS=true` for + # shared action state and a durable shared checkpointer for graph continuation. + # MongoDB is currently the built-in durable checkpointer (and the default). # schedules: # use: true # create: true diff --git a/packages/api/src/agents/__tests__/run-summarization.test.ts b/packages/api/src/agents/__tests__/run-summarization.test.ts index 083ef084ba..55c61514ef 100644 --- a/packages/api/src/agents/__tests__/run-summarization.test.ts +++ b/packages/api/src/agents/__tests__/run-summarization.test.ts @@ -11,7 +11,7 @@ import type { BaseMessage } from '@langchain/core/messages'; import type { SubagentTaskConfig } from '@librechat/agents'; import type { AppConfig } from '@librechat/data-schemas'; import type { ModelBoundChatModelCallback } from '~/middleware/modelBoundContent'; -import { createRun } from '~/agents/run'; +import { createRun, isAskUserQuestionAdminDisabled } from '~/agents/run'; // Mock winston logger — `format` must be callable so @librechat/data-schemas // dist module-load completes cleanly; see api/test/__mocks__/logger.js. @@ -107,6 +107,21 @@ function makeAgent( }; } +describe('isAskUserQuestionAdminDisabled', () => { + it('applies includedTools precedence and the filteredTools fallback', () => { + expect(isAskUserQuestionAdminDisabled(undefined)).toBe(false); + expect(isAskUserQuestionAdminDisabled({ includedTools: ['calculator'] } as AppConfig)).toBe( + true, + ); + expect( + isAskUserQuestionAdminDisabled({ includedTools: ['ask_user_question'] } as AppConfig), + ).toBe(false); + expect( + isAskUserQuestionAdminDisabled({ filteredTools: ['ask_user_question'] } as AppConfig), + ).toBe(true); + }); +}); + type TestRunAgent = ReturnType & { subagentAgentConfigs?: TestRunAgent[]; }; diff --git a/packages/api/src/agents/checkpointer.integration.spec.ts b/packages/api/src/agents/checkpointer.integration.spec.ts index aca60f7134..c7ea1fecd4 100644 --- a/packages/api/src/agents/checkpointer.integration.spec.ts +++ b/packages/api/src/agents/checkpointer.integration.spec.ts @@ -5,6 +5,7 @@ import { MongoDBSaver } from '@langchain/langgraph-checkpoint-mongodb'; import { emptyCheckpoint, ERROR, INTERRUPT } from '@langchain/langgraph-checkpoint'; import { getAgentCheckpointer, + hasDurableAgentInterruptCheckpoint, captureAgentCheckpointGeneration, deleteAgentCheckpoint, deleteAgentCheckpoints, @@ -49,6 +50,7 @@ async function seedInterruptCheckpoint( saver: MongoDBSaver, threadId: string, checkpointNamespace = '', + interruptId = 'interrupt-current', ) { const { config, checkpoint, metadata } = putArgs(threadId, checkpointNamespace); await saver.putWrites( @@ -59,7 +61,7 @@ async function seedInterruptCheckpoint( checkpoint_id: checkpoint.id, }, }, - [[INTERRUPT, 'approve?']], + [[INTERRUPT, { id: interruptId, value: 'approve?' }]], 'task-1', ); await saver.put(config, checkpoint, metadata); @@ -98,6 +100,68 @@ describe('checkpointer (mongodb-memory-server integration)', () => { expect(ttlIndex?.expireAfterSeconds).toBe(3600); }); + it('verifies that an interrupt write has its matching durable checkpoint', async () => { + const saver = await getAgentCheckpointer(MONGO_CFG); + expect(saver).toBeDefined(); + const missingIdentity = { + checkpointId: 'missing-checkpoint', + interruptId: 'interrupt-current', + }; + await expect( + hasDurableAgentInterruptCheckpoint('verified-pause', MONGO_CFG, missingIdentity), + ).resolves.toBe(false); + + const checkpoint = await seedInterruptCheckpoint(saver!, 'verified-pause'); + const identity = { + checkpointId: checkpoint.id, + interruptId: 'interrupt-current', + }; + + await expect( + hasDurableAgentInterruptCheckpoint('verified-pause', MONGO_CFG, identity), + ).resolves.toBe(true); + await expect( + hasDurableAgentInterruptCheckpoint('verified-pause', MONGO_CFG, missingIdentity), + ).resolves.toBe(false); + await expect( + hasDurableAgentInterruptCheckpoint('verified-pause', MONGO_CFG, { + ...identity, + interruptId: 'interrupt-stale', + }), + ).resolves.toBe(false); + await mongoose.connection.db!.collection('agent_checkpoints').deleteMany({ + thread_id: 'verified-pause', + }); + await expect( + hasDurableAgentInterruptCheckpoint('verified-pause', MONGO_CFG, identity), + ).resolves.toBe(false); + }); + + it("uses Mongoose's selected database instead of the MongoClient URI default", async () => { + await mongoose.disconnect(); + await mongoose.connect(mongoServer.getUri('driver_default'), { dbName: 'active_app' }); + __resetCheckpointerForTests(); + + const saver = await getAgentCheckpointer(MONGO_CFG); + expect(saver).toBeDefined(); + await seedInterruptCheckpoint(saver!, 'db-selection-pause'); + + expect( + await mongoose.connection.db!.collection('agent_checkpoints').countDocuments({ + thread_id: 'db-selection-pause', + }), + ).toBe(1); + expect( + await mongoose.connection + .getClient() + .db('driver_default') + .collection('agent_checkpoints') + .countDocuments({ + thread_id: 'db-selection-pause', + }), + ).toBe(0); + }); + it('returns undefined for the memory type (SDK MemorySaver fallback) even when connected', async () => { expect(await getAgentCheckpointer({ type: 'memory' })).toBeUndefined(); }); diff --git a/packages/api/src/agents/checkpointer.ts b/packages/api/src/agents/checkpointer.ts index 9f995d0cb2..277056a64e 100644 --- a/packages/api/src/agents/checkpointer.ts +++ b/packages/api/src/agents/checkpointer.ts @@ -403,6 +403,9 @@ export class LazyMongoSaver extends MongoDBSaver { if (isEventActorInvocation(config)) { await this.assertCheckpointFitsDocument(config, checkpoint, metadata); const persisted = await super.put(toStorageCheckpointConfig(config), checkpoint, metadata); + logger.debug( + `[checkpointer] Persisted durable checkpoint for thread ${config.configurable?.thread_id ?? 'unknown'} (${checkpoint.id})`, + ); return fromStorageCheckpointConfig(persisted, config); } if (this.writeAnchorIds.delete(checkpoint.id)) { @@ -413,6 +416,9 @@ export class LazyMongoSaver extends MongoDBSaver { sweepStale(this.persistedIds, (t) => t); this.persistedIds.set(checkpoint.id, Date.now()); const persisted = await super.put(toStorageCheckpointConfig(config), checkpoint, metadata); + logger.debug( + `[checkpointer] Persisted durable checkpoint for thread ${config.configurable?.thread_id ?? 'unknown'} (${checkpoint.id})`, + ); // `assertCheckpointFitsDocument` awaits a (potentially slow) serialization AFTER the // anchor was consumed above but BEFORE `persistedIds` was set — a bookkeeping-only // `putWrites` dispatched in that window sees neither marker and parks its batch. Flush @@ -512,7 +518,7 @@ export class LazyMongoSaver extends MongoDBSaver { return; } logger.debug( - `[checkpointer] Persisting durable checkpoint for thread ${threadId ?? 'unknown'}: ${bytes} bytes`, + `[checkpointer] Prepared durable checkpoint for thread ${threadId ?? 'unknown'}: ${bytes} bytes`, ); } } @@ -587,6 +593,62 @@ export function getApprovalTtlMs(cfg: TCheckpointerConfig | undefined): number { return resolveCheckpointerConfig(cfg).ttlSeconds * 1000; } +/** + * Prove that the durable saver contains a complete interrupt checkpoint for one generation. + * + * A pending Redis action is useful only when LangGraph can reload the state it + * interrupted. Read the exact checkpoint selected by the current interrupt and + * require its matching interrupt id, so an older retained pause cannot satisfy + * verification for a missing or misrouted re-pause. + * This runs once per interrupt, never on the ordinary generation path. + */ +export async function hasDurableAgentInterruptCheckpoint( + threadId: string, + cfg?: TCheckpointerConfig, + options?: { + checkpointNamespace?: string; + checkpointId: string; + checkpointNs?: string; + interruptId: string; + }, +): Promise { + if (!threadId || !options?.checkpointId || !options.interruptId) { + return false; + } + const saver = await getAgentCheckpointer(cfg); + if (!saver) { + return false; + } + + const checkpointNamespace = options?.checkpointNamespace ?? ''; + const tuple = await saver.getTuple({ + configurable: { + thread_id: threadId, + checkpoint_ns: options.checkpointNs ?? '', + checkpoint_id: options.checkpointId, + ...(checkpointNamespace !== '' && { + [LIBRECHAT_CHECKPOINT_NAMESPACE_KEY]: checkpointNamespace, + }), + }, + }); + if (tuple?.checkpoint.id !== options.checkpointId) { + return false; + } + return (tuple.pendingWrites ?? []).some((write) => { + if (write[1] !== INTERRUPT) { + return false; + } + const values = Array.isArray(write[2]) ? write[2] : [write[2]]; + return values.some( + (value) => + value != null && + typeof value === 'object' && + 'id' in value && + value.id === options.interruptId, + ); + }); +} + /** * One saver per process, built lazily on first use so `setup()` (index creation) * runs exactly once. Keyed by the resolved settings so a config change rebuilds. @@ -711,6 +773,13 @@ async function buildMongoSaver( client: mongoose.connection.getClient() as unknown as ConstructorParameters< typeof MongoDBSaver >[0]['client'], + // MongoDBSaver calls MongoClient.db(dbName). Passing no name makes that + // resolve from the driver's URI default, which is not guaranteed to be the + // database Mongoose selected (for example when Mongoose connected with a + // dbName override). Every capture/delete path below uses connection.db, so + // bind the saver to that exact database as well or a pause can be written to + // one database while LibreChat looks for it in another. + dbName: mongoose.connection.db?.databaseName, checkpointCollectionName: resolved.checkpointCollectionName, checkpointWritesCollectionName: resolved.checkpointWritesCollectionName, // TTL index on `upserted_at`: an unresolved paused run is reclaimed after the diff --git a/packages/api/src/agents/hitl/admission.spec.ts b/packages/api/src/agents/hitl/admission.spec.ts new file mode 100644 index 0000000000..170654e25a --- /dev/null +++ b/packages/api/src/agents/hitl/admission.spec.ts @@ -0,0 +1,163 @@ +import type { PluginHookSource } from '~/agents/hooks/source'; +import type { ToolApprovalAdmissionAgent } from './admission'; +import type { ToolApprovalHook } from './hooks'; +import { agentRunUsesCheckpointer, canAgentGraphPause } from './admission'; + +const askHook: ToolApprovalHook = async () => ({ decision: 'ask' }); + +function pluginSource( + hasToolApprovalHooks: PluginHookSource['hasToolApprovalHooks'], +): PluginHookSource { + return { + hasHooks: () => true, + hasToolApprovalHooks, + register: () => 0, + }; +} + +describe('canAgentGraphPause', () => { + test.each([ + ['configured tool names', { tools: ['read_file'] }, 'read_file'], + ['loaded tool objects', { tools: [{ name: 'read_file' }] }, 'read_file'], + ['tool definitions', { toolDefinitions: [{ name: 'read_file' }] }, 'read_file'], + ['tool registries', { toolRegistry: new Map([['read_file', {}]]) }, 'read_file'], + ])('discovers %s', (_label, agent, toolName) => { + expect( + canAgentGraphPause({ + policy: { enabled: true, mode: 'bypass', ask: [toolName] }, + agents: [agent], + }), + ).toBe(true); + }); + + test.each([ + ['initialized children', { subagentAgentConfigs: [{ tools: ['write_file'] }] }], + ['lazy children', { lazySubagentConfigs: [{ tools: ['write_file'] }] }], + ['graph members', { subagentGraphConfigs: [{ memberConfigs: [{ tools: ['write_file'] }] }] }], + ['graph member metadata', { subagentGraphMemberMetadata: [{ tools: ['write_file'] }] }], + ])('intersects approval policy with %s', (_label, agent) => { + expect( + canAgentGraphPause({ + policy: { enabled: true, mode: 'bypass', ask: ['write_*'] }, + agents: [agent], + }), + ).toBe(true); + }); + + test('fails closed for an unresolved lazy tool surface that could pause', () => { + expect( + canAgentGraphPause({ + policy: { enabled: true, mode: 'bypass', ask: ['write_*'] }, + agents: [{ lazySubagentConfigs: [{}] }], + }), + ).toBe(true); + expect( + canAgentGraphPause({ + policy: { enabled: true, mode: 'bypass' }, + agents: [{ lazySubagentConfigs: [{}] }], + }), + ).toBe(false); + }); + + test('does not match an approval rule outside the reachable tool surface', () => { + expect( + canAgentGraphPause({ + policy: { enabled: true, mode: 'bypass', ask: ['delete_*'] }, + agents: [{ tools: ['read_file'] }], + }), + ).toBe(false); + }); + + test('includes host-generated runtime tools in approval admission', () => { + expect( + canAgentGraphPause({ + policy: { enabled: true, mode: 'bypass', ask: ['check_background_task'] }, + agents: [{}], + hostGeneratedToolNames: ['check_background_task'], + }), + ).toBe(true); + }); + + test('matches static and request-scoped rules against MCP aliases', () => { + const agent: ToolApprovalAdmissionAgent = { + tools: ['mcp__server__read_file'], + mcpToolAliases: [{ name: 'mcp__server__read_file', aliasName: 'read_file' }], + }; + expect( + canAgentGraphPause({ + policy: { enabled: true, mode: 'bypass', ask: ['read_file'] }, + agents: [agent], + }), + ).toBe(true); + expect( + canAgentGraphPause({ + policy: { enabled: true, mode: 'bypass' }, + agents: [agent], + resolvedProgrammaticHooks: [{ hook: askHook, matcher: '^read_file$' }], + }), + ).toBe(true); + }); + + test('asks deployment hook sources only about concrete runtime tool names', () => { + const hasToolApprovalHooks = jest.fn( + (names?: readonly string[]) => names?.includes('write_file') === true, + ); + expect( + canAgentGraphPause({ + policy: { enabled: true, mode: 'bypass' }, + agents: [{ tools: ['read_file', 'write_file'] }], + pluginHookSource: pluginSource(hasToolApprovalHooks), + }), + ).toBe(true); + expect(hasToolApprovalHooks).toHaveBeenCalledWith(['read_file']); + expect(hasToolApprovalHooks).toHaveBeenCalledWith(['write_file']); + }); + + test('classifies top-level ask_user_question unless it is filtered or denied', () => { + const agents = [{ tools: ['ask_user_question'] }]; + expect(canAgentGraphPause({ policy: undefined, agents })).toBe(true); + expect( + canAgentGraphPause({ + policy: { enabled: true, deny: ['ask_*'] }, + agents, + }), + ).toBe(false); + expect( + canAgentGraphPause({ + policy: { enabled: true }, + agents, + askUserQuestionAdminDisabled: true, + }), + ).toBe(false); + }); + + test('does not promote nested ask_user_question to a parent pause capability', () => { + expect( + canAgentGraphPause({ + policy: undefined, + agents: [{ subagentAgentConfigs: [{ tools: ['ask_user_question'] }] }], + }), + ).toBe(false); + }); +}); + +describe('agentRunUsesCheckpointer', () => { + test('tracks checkpointer attachment independently from current pause capability', () => { + const policy = { enabled: true, mode: 'bypass' as const }; + const agents = [{ tools: ['read_file'] }]; + expect(canAgentGraphPause({ policy, agents })).toBe(false); + expect(agentRunUsesCheckpointer({ policy, agents })).toBe(true); + }); + + test('uses the same top-level ask-tool admin gate as createRun', () => { + const agents = [{ tools: ['ask_user_question'] }]; + expect(agentRunUsesCheckpointer({ policy: undefined, agents })).toBe(true); + expect( + agentRunUsesCheckpointer({ + policy: undefined, + agents, + askUserQuestionAdminDisabled: true, + }), + ).toBe(false); + }); +}); diff --git a/packages/api/src/agents/hitl/admission.ts b/packages/api/src/agents/hitl/admission.ts new file mode 100644 index 0000000000..0c6405f140 --- /dev/null +++ b/packages/api/src/agents/hitl/admission.ts @@ -0,0 +1,185 @@ +import type { TToolApprovalPolicy } from 'librechat-data-provider'; +import type { PluginHookSource } from '~/agents/hooks/source'; +import type { MCPToolAlias } from '~/tools/classification'; +import type { ResolvedToolApprovalHook } from './hooks'; +import { + healToolApprovalPolicy, + isHITLEnabled, + isToolApprovalPauseCapable, + isToolDeniedByApprovalPolicy, +} from './policy'; +import { ASK_USER_QUESTION_TOOL_NAME } from './askUserQuestionTool'; +import { resolvedToolApprovalHooksCanMatch } from './hooks'; + +interface ApprovalToolReference { + readonly name?: string; +} + +interface ApprovalToolRegistry { + keys(): Iterable; + has(name: string): boolean; +} + +interface ApprovalSubagentGraph { + readonly memberConfigs?: readonly (ToolApprovalAdmissionAgent | null | undefined)[]; +} + +export interface ToolApprovalAdmissionAgent { + readonly tools?: readonly (string | ApprovalToolReference)[]; + readonly toolRegistry?: ApprovalToolRegistry; + readonly toolDefinitions?: readonly ApprovalToolReference[]; + readonly mcpToolAliases?: readonly MCPToolAlias[]; + readonly subagentAgentConfigs?: readonly (ToolApprovalAdmissionAgent | null | undefined)[]; + readonly lazySubagentConfigs?: readonly (ToolApprovalAdmissionAgent | null | undefined)[]; + readonly subagentGraphMemberMetadata?: readonly (ToolApprovalAdmissionAgent | null | undefined)[]; + readonly subagentGraphConfigs?: readonly ApprovalSubagentGraph[]; +} + +export interface ToolApprovalAdmissionInput { + readonly policy: TToolApprovalPolicy | undefined; + readonly agents: readonly (ToolApprovalAdmissionAgent | null | undefined)[]; + readonly hostGeneratedToolNames?: readonly string[]; + readonly resolvedProgrammaticHooks?: readonly ResolvedToolApprovalHook[]; + readonly pluginHookSource?: PluginHookSource; + readonly askUserQuestionAdminDisabled?: boolean; +} + +function agentHasTool(agent: ToolApprovalAdmissionAgent, toolName: string): boolean { + return ( + agent.tools?.some((tool) => (typeof tool === 'string' ? tool : tool.name) === toolName) === + true || + agent.toolRegistry?.has(toolName) === true || + agent.toolDefinitions?.some((definition) => definition.name === toolName) === true + ); +} + +function collectApprovalAgents(roots: readonly (ToolApprovalAdmissionAgent | null | undefined)[]): { + agents: ToolApprovalAdmissionAgent[]; + hasLazyToolSurface: boolean; +} { + const agents: ToolApprovalAdmissionAgent[] = []; + const visited = new Set(); + const pending = [...roots]; + let hasLazyToolSurface = false; + + for (let index = 0; index < pending.length; index++) { + const agent = pending[index]; + if (agent == null || visited.has(agent)) { + continue; + } + visited.add(agent); + agents.push(agent); + pending.push(...(agent.subagentAgentConfigs ?? [])); + if ((agent.lazySubagentConfigs?.length ?? 0) > 0) { + hasLazyToolSurface = true; + pending.push(...(agent.lazySubagentConfigs ?? [])); + } + pending.push(...(agent.subagentGraphMemberMetadata ?? [])); + for (const graph of agent.subagentGraphConfigs ?? []) { + pending.push(...(graph.memberConfigs ?? [])); + } + } + + return { agents, hasLazyToolSurface }; +} + +/** + * Whether an initialized run can pause through tool approval or a top-level + * `ask_user_question`. Eager tools are matched exactly across every subagent + * form; unresolved lazy surfaces are classified conservatively. The interrupt + * boundary remains the final fail-closed durability check. + */ +export function canAgentGraphPause({ + policy, + agents, + hostGeneratedToolNames = [], + resolvedProgrammaticHooks = [], + pluginHookSource, + askUserQuestionAdminDisabled = false, +}: ToolApprovalAdmissionInput): boolean { + const asksUserQuestion = + !askUserQuestionAdminDisabled && + !isToolDeniedByApprovalPolicy(policy, ASK_USER_QUESTION_TOOL_NAME) && + agents.some((agent) => agent != null && agentHasTool(agent, ASK_USER_QUESTION_TOOL_NAME)); + if (!isHITLEnabled(policy)) { + return asksUserQuestion; + } + + const approvalGraph = collectApprovalAgents(agents); + const toolNames = new Set(); + const aliases: MCPToolAlias[] = []; + const aliasesByToolName = new Map(); + const addToolName = (name: unknown) => { + if (typeof name === 'string' && name !== ASK_USER_QUESTION_TOOL_NAME) { + toolNames.add(name); + } + }; + + for (const name of hostGeneratedToolNames) { + addToolName(name); + } + + for (const agent of approvalGraph.agents) { + for (const tool of agent.tools ?? []) { + addToolName(typeof tool === 'string' ? tool : tool.name); + } + if (agent.toolRegistry) { + for (const name of agent.toolRegistry.keys()) { + addToolName(name); + } + } + for (const definition of agent.toolDefinitions ?? []) { + addToolName(definition.name); + } + for (const alias of agent.mcpToolAliases ?? []) { + aliases.push(alias); + const names = aliasesByToolName.get(alias.name) ?? []; + names.push(alias.aliasName); + aliasesByToolName.set(alias.name, names); + } + } + + const effectivePolicy = healToolApprovalPolicy(policy, aliases); + const knownToolCanPause = Array.from(toolNames).some((toolName) => { + const matcherNames = [toolName, ...(aliasesByToolName.get(toolName) ?? [])]; + const requestHookCanAsk = resolvedToolApprovalHooksCanMatch( + resolvedProgrammaticHooks, + matcherNames, + ); + const pluginHookCanAsk = pluginHookSource?.hasToolApprovalHooks?.([toolName]) === true; + return isToolApprovalPauseCapable(effectivePolicy, requestHookCanAsk || pluginHookCanAsk, [ + toolName, + ]); + }); + if (knownToolCanPause) { + return true; + } + if (approvalGraph.hasLazyToolSurface) { + const unresolvedHookCanAsk = + resolvedProgrammaticHooks.length > 0 || pluginHookSource?.hasToolApprovalHooks?.() === true; + if (isToolApprovalPauseCapable(effectivePolicy, unresolvedHookCanAsk)) { + return true; + } + } + return asksUserQuestion; +} + +/** + * Whether `createRun` attaches a checkpointer for this initialization. + * Cleanup follows attachment, not current pause capability: a retry must not + * restore remnants written before a policy or request-hook change. + */ +export function agentRunUsesCheckpointer({ + policy, + agents, + askUserQuestionAdminDisabled = false, +}: Pick< + ToolApprovalAdmissionInput, + 'policy' | 'agents' | 'askUserQuestionAdminDisabled' +>): boolean { + return ( + isHITLEnabled(policy) || + (!askUserQuestionAdminDisabled && + agents.some((agent) => agent != null && agentHasTool(agent, ASK_USER_QUESTION_TOOL_NAME))) + ); +} diff --git a/packages/api/src/agents/hitl/hooks.spec.ts b/packages/api/src/agents/hitl/hooks.spec.ts index 1ab1c7ea21..2e763ae0dc 100644 --- a/packages/api/src/agents/hitl/hooks.spec.ts +++ b/packages/api/src/agents/hitl/hooks.spec.ts @@ -4,6 +4,7 @@ import { getRegisteredToolApprovalHookCount, clearToolApprovalHooks, buildToolApprovalHooks, + resolvedToolApprovalHooksCanMatch, } from './hooks'; const denyHook: ToolApprovalHook = async () => ({ decision: 'deny' }); @@ -74,4 +75,18 @@ describe('tool-approval hook registry', () => { expect(buildToolApprovalHooks({})).toEqual([]); }); }); + + describe('resolvedToolApprovalHooksCanMatch', () => { + test('matches only hooks that can run for the selected tool names', () => { + const hooks = [{ hook: denyHook, matcher: '^approval_probe$' }]; + expect(resolvedToolApprovalHooksCanMatch(hooks, ['read_file'])).toBe(false); + expect(resolvedToolApprovalHooksCanMatch(hooks, ['approval_probe'])).toBe(true); + }); + + test('treats an unscoped hook as matching any nonempty tool surface', () => { + const hooks = [{ hook: denyHook }]; + expect(resolvedToolApprovalHooksCanMatch(hooks, [])).toBe(false); + expect(resolvedToolApprovalHooksCanMatch(hooks, ['read_file'])).toBe(true); + }); + }); }); diff --git a/packages/api/src/agents/hitl/hooks.ts b/packages/api/src/agents/hitl/hooks.ts index 826b58882d..a443370db1 100644 --- a/packages/api/src/agents/hitl/hooks.ts +++ b/packages/api/src/agents/hitl/hooks.ts @@ -44,6 +44,11 @@ export type ToolApprovalHookFactory = ( context: ToolApprovalHookContext, ) => ToolApprovalHook | undefined; +export interface ResolvedToolApprovalHook { + hook: ToolApprovalHook; + matcher?: string; +} + interface RegisteredHook { factory: ToolApprovalHookFactory; /** Optional regex matched against the tool name (the `PreToolUse` matcher `pattern`). */ @@ -106,8 +111,8 @@ export function clearToolApprovalHooks(): void { */ export function buildToolApprovalHooks( context: ToolApprovalHookContext, -): Array<{ hook: ToolApprovalHook; matcher?: string }> { - const built: Array<{ hook: ToolApprovalHook; matcher?: string }> = []; +): ResolvedToolApprovalHook[] { + const built: ResolvedToolApprovalHook[] = []; for (const { factory, matcher } of registeredHooks) { const hook = factory(context); if (hook) { @@ -116,3 +121,25 @@ export function buildToolApprovalHooks( } return built; } + +/** Whether any resolved hook matcher can run for one of the supplied tool names. */ +export function resolvedToolApprovalHooksCanMatch( + hooks: readonly ResolvedToolApprovalHook[], + toolNames: readonly string[], +): boolean { + return hooks.some(({ matcher }) => { + if (matcher == null) { + return toolNames.length > 0; + } + let regex: RegExp; + try { + regex = new RegExp(matcher); + } catch { + return false; + } + return toolNames.some((name) => { + regex.lastIndex = 0; + return regex.test(name); + }); + }); +} diff --git a/packages/api/src/agents/hitl/index.ts b/packages/api/src/agents/hitl/index.ts index 39ad15c8ed..3688fa1ba1 100644 --- a/packages/api/src/agents/hitl/index.ts +++ b/packages/api/src/agents/hitl/index.ts @@ -1,4 +1,5 @@ export * from './policy'; +export * from './admission'; export * from './runtime'; export * from './resume'; export * from './preflight'; diff --git a/packages/api/src/agents/hitl/policy.spec.ts b/packages/api/src/agents/hitl/policy.spec.ts index e95f3b034a..331e9464b8 100644 --- a/packages/api/src/agents/hitl/policy.spec.ts +++ b/packages/api/src/agents/hitl/policy.spec.ts @@ -17,8 +17,60 @@ import { applyResumeContext, applyResumeModelParameters, exemptAskUserQuestionFromApproval, + isToolApprovalPauseCapable, + isToolDeniedByApprovalPolicy, } from './policy'; +describe('isToolApprovalPauseCapable', () => { + it('recognizes the default ask fallback and explicit ask rules', () => { + expect(isToolApprovalPauseCapable({ enabled: true })).toBe(true); + expect(isToolApprovalPauseCapable({ enabled: true, mode: 'bypass', ask: ['write_*'] })).toBe( + true, + ); + }); + + it('excludes policies that can only allow or deny', () => { + expect(isToolApprovalPauseCapable({ enabled: true, mode: 'bypass' })).toBe(false); + expect(isToolApprovalPauseCapable({ enabled: true, mode: 'dontAsk' })).toBe(false); + expect(isToolApprovalPauseCapable({ enabled: true, allow: ['*'] })).toBe(false); + expect(isToolApprovalPauseCapable({ enabled: true, deny: ['*'], ask: ['write_*'] }, true)).toBe( + false, + ); + }); + + it('treats a programmatic hook as pause-capable when policy does not deny every tool', () => { + expect(isToolApprovalPauseCapable({ enabled: true, mode: 'bypass' }, true)).toBe(true); + }); + + it('intersects approval rules with the selected run tool surface', () => { + const policy = { enabled: true, mode: 'bypass' as const, ask: ['write_*'] }; + expect(isToolApprovalPauseCapable(policy, false, [])).toBe(false); + expect(isToolApprovalPauseCapable(policy, false, ['read_file'])).toBe(false); + expect(isToolApprovalPauseCapable(policy, false, ['write_file'])).toBe(true); + expect(isToolApprovalPauseCapable({ enabled: true }, false, ['read_file'])).toBe(true); + }); + + it('keeps deny precedence when a matching programmatic hook can ask', () => { + const policy = { enabled: true, mode: 'bypass' as const, deny: ['delete_*'] }; + expect(isToolApprovalPauseCapable(policy, true, ['delete_file'])).toBe(false); + expect(isToolApprovalPauseCapable(policy, true, ['write_file'])).toBe(true); + }); +}); + +describe('isToolDeniedByApprovalPolicy', () => { + it('matches exact and wildcard denies only when approval is enabled', () => { + expect( + isToolDeniedByApprovalPolicy({ enabled: true, deny: ['ask_*'] }, 'ask_user_question'), + ).toBe(true); + expect( + isToolDeniedByApprovalPolicy({ enabled: true, deny: ['write_*'] }, 'ask_user_question'), + ).toBe(false); + expect( + isToolDeniedByApprovalPolicy({ enabled: false, deny: ['ask_*'] }, 'ask_user_question'), + ).toBe(false); + }); +}); + describe('resolveToolApprovalPolicy', () => { test('returns the endpoint policy unchanged (single layer wired today)', () => { const endpoint: TToolApprovalPolicy = { enabled: true, mode: 'default', deny: ['rm'] }; @@ -263,6 +315,17 @@ describe('buildPendingAction', () => { expect(action.expiresAt).toBeGreaterThanOrEqual(before); expect(action.expiresAt).toBeLessThanOrEqual(after); }); + + test('caps the TTL at an inherited absolute deadline without a second clock read', () => { + const deadline = Date.now() + 1_000; + const action = buildPendingAction(toolApprovalPayload, { + ...ctx, + ttlMs: 5_000, + expiresAt: new Date(deadline), + }); + + expect(action.expiresAt).toBe(deadline); + }); }); describe('toClientPendingAction', () => { diff --git a/packages/api/src/agents/hitl/policy.ts b/packages/api/src/agents/hitl/policy.ts index d3350ec063..7112cbd443 100644 --- a/packages/api/src/agents/hitl/policy.ts +++ b/packages/api/src/agents/hitl/policy.ts @@ -76,10 +76,63 @@ export function resolveToolApprovalPolicy( * any multi-process deployment. Pair this predicate with the checkpointer * assignment at the `Run.create` call site. */ -export function isHITLEnabled(policy: TToolApprovalPolicy | undefined): boolean { +export function isHITLEnabled( + policy: TToolApprovalPolicy | undefined, +): policy is NonNullable { return policy?.enabled === true; } +/** + * Whether the configured policy can structurally return `ask` for any tool. + * + * `bypass` and `dontAsk` are non-pausing fallbacks; only an explicit ask rule + * or a programmatic hook can tighten them to `ask`. A catch-all deny remains + * non-pausing because deny wins over every hook/rule, while a catch-all allow + * removes the default mode's unmatched-tool ask fallback. More-specific pattern + * overlap is intentionally treated conservatively because the run's complete + * lazy tool surface is not known at admission time. + */ +export function isToolApprovalPauseCapable( + policy: TToolApprovalPolicy | undefined, + hasProgrammaticHooks = false, + toolNames?: readonly string[], +): boolean { + if (!isHITLEnabled(policy)) { + return false; + } + const enabledPolicy = policy; + if (toolNames != null) { + const names = Array.from(new Set(toolNames.filter((name) => name.length > 0))); + if (names.length === 0) { + return false; + } + const matches = (patterns: string[] | undefined, name: string): boolean => + patterns?.some((pattern) => globToRegex(pattern).test(name)) === true; + return names.some((name) => { + if (matches(enabledPolicy.deny, name)) { + return false; + } + if (hasProgrammaticHooks || matches(enabledPolicy.ask, name)) { + return true; + } + if (matches(enabledPolicy.allow, name)) { + return false; + } + return enabledPolicy.mode !== 'bypass' && enabledPolicy.mode !== 'dontAsk'; + }); + } + if (policy?.deny?.includes('*')) { + return false; + } + if (hasProgrammaticHooks || (policy?.ask?.length ?? 0) > 0) { + return true; + } + if (policy?.mode === 'bypass' || policy?.mode === 'dontAsk') { + return false; + } + return policy?.allow?.includes('*') !== true; +} + /** * Map a LibreChat tool-approval policy to the SDK's `ToolPolicyConfig`. * @@ -93,6 +146,17 @@ function globToRegex(pattern: string): RegExp { return new RegExp('^' + escaped.replace(/\*/g, '.*') + '$'); } +/** Whether an enabled static policy unconditionally denies one concrete tool name. */ +export function isToolDeniedByApprovalPolicy( + policy: TToolApprovalPolicy | undefined, + toolName: string, +): boolean { + return ( + isHITLEnabled(policy) && + policy.deny?.some((pattern) => globToRegex(pattern).test(toolName)) === true + ); +} + /** * Extends each `toolApproval` pattern list with the names of tools whose * OTHER spelling matches, so admin YAML keeps applying when a tool's key @@ -252,6 +316,8 @@ export interface PendingActionContext { responseMessageId?: string; /** Optional TTL (ms). When set, `expiresAt = createdAt + ttlMs`. */ ttlMs?: number; + /** Optional absolute upper bound inherited from an enclosing event binding. */ + expiresAt?: Date | string | number; /** Override actionId; defaults to a fresh uuid. */ actionId?: string; /** SDK interrupt id (`RunInterruptResult.interruptId`) for cross-process resume. */ @@ -661,6 +727,25 @@ export function buildPendingAction( ctx: PendingActionContext, ): Agents.PendingAction { const createdAt = Date.now(); + const ttlExpiresAt = typeof ctx.ttlMs === 'number' ? createdAt + ctx.ttlMs : undefined; + let absoluteExpiresAt: number | undefined; + if (typeof ctx.expiresAt === 'number') { + absoluteExpiresAt = ctx.expiresAt; + } else if (ctx.expiresAt instanceof Date) { + absoluteExpiresAt = ctx.expiresAt.getTime(); + } else if (typeof ctx.expiresAt === 'string') { + absoluteExpiresAt = new Date(ctx.expiresAt).getTime(); + } + const finiteAbsoluteExpiresAt = Number.isFinite(absoluteExpiresAt) + ? absoluteExpiresAt + : undefined; + let expiresAt = ttlExpiresAt; + if (finiteAbsoluteExpiresAt != null) { + expiresAt = + ttlExpiresAt == null + ? finiteAbsoluteExpiresAt + : Math.min(ttlExpiresAt, finiteAbsoluteExpiresAt); + } return { actionId: ctx.actionId ?? randomUUID(), streamId: ctx.streamId, @@ -669,7 +754,7 @@ export function buildPendingAction( responseMessageId: ctx.responseMessageId, payload, createdAt, - expiresAt: typeof ctx.ttlMs === 'number' ? createdAt + ctx.ttlMs : undefined, + expiresAt, interruptId: ctx.interruptId, threadId: ctx.threadId, requestFingerprint: ctx.requestFingerprint, diff --git a/packages/api/src/agents/hitl/runtime.spec.ts b/packages/api/src/agents/hitl/runtime.spec.ts index 46142fffcc..65ae715075 100644 --- a/packages/api/src/agents/hitl/runtime.spec.ts +++ b/packages/api/src/agents/hitl/runtime.spec.ts @@ -88,6 +88,19 @@ describe('buildHITLRunWiring host-hook composition', () => { ); }); + test('reuses request-scoped hooks resolved by admission without invoking factories twice', () => { + const hook = async () => ({ decision: 'ask' as const }); + const factory = jest.fn(() => hook); + registerToolApprovalHook(factory); + const resolved = [{ hook }]; + factory.mockClear(); + + const wiring = buildHITLRunWiring({ enabled: true }, {}, [], resolved); + + expect(factory).not.toHaveBeenCalled(); + expect(wiring?.hooks.getMatchers('PreToolUse')).toHaveLength(2); + }); + test('matches lazy aliases without changing host-hook ordering', async () => { const hook = jest.fn(async () => ({ decision: 'deny' as const })); registerToolApprovalHook(() => hook, { diff --git a/packages/api/src/agents/hitl/runtime.ts b/packages/api/src/agents/hitl/runtime.ts index fd35edd7be..e732a3b060 100644 --- a/packages/api/src/agents/hitl/runtime.ts +++ b/packages/api/src/agents/hitl/runtime.ts @@ -1,7 +1,7 @@ import { HookRegistry, createToolPolicyHook } from '@librechat/agents'; import type { TToolApprovalPolicy } from 'librechat-data-provider'; +import type { ResolvedToolApprovalHook, ToolApprovalHookContext } from './hooks'; import type { MCPToolAlias } from '~/tools/classification'; -import type { ToolApprovalHookContext } from './hooks'; import { isHITLEnabled, mapToolApprovalPolicy } from './policy'; import { buildToolApprovalHooks } from './hooks'; @@ -40,13 +40,14 @@ export function buildHITLRunWiring( policy: TToolApprovalPolicy | undefined, context: ToolApprovalHookContext = {}, mcpToolAliases: readonly MCPToolAlias[] = [], + resolvedProgrammaticHooks?: readonly ResolvedToolApprovalHook[], ): HITLRunWiring | undefined { if (!isHITLEnabled(policy)) { return undefined; } const registry = new HookRegistry(); - let activePolicy = policy; + let activePolicy: TToolApprovalPolicy | undefined = policy; const aliases = [...mcpToolAliases]; const registeredAliases = new Set( aliases.map(({ name, aliasName }) => `${name}\u0000${aliasName}`), @@ -60,7 +61,7 @@ export function buildHITLRunWiring( }); // Host-registered programmatic hooks — context-aware, layered after the static-policy hook. - const programmaticHooks = buildToolApprovalHooks(context); + const programmaticHooks = resolvedProgrammaticHooks ?? buildToolApprovalHooks(context); for (const { hook, matcher } of programmaticHooks) { if (matcher == null) { registry.register('PreToolUse', { hooks: [hook] }); diff --git a/packages/api/src/agents/hooks/source.ts b/packages/api/src/agents/hooks/source.ts index 8774d77910..a73a66d7e9 100644 --- a/packages/api/src/agents/hooks/source.ts +++ b/packages/api/src/agents/hooks/source.ts @@ -16,6 +16,8 @@ export interface PluginHookSourceOptions { */ export interface PluginHookSource { hasHooks(): boolean; + /** Whether a ready deployment hook can participate in tool-approval decisions. */ + hasToolApprovalHooks?(toolNames?: readonly string[]): boolean; register(options: PluginHookSourceOptions): number; } diff --git a/packages/api/src/agents/run.ts b/packages/api/src/agents/run.ts index dbc18fab69..f1b9eeb246 100644 --- a/packages/api/src/agents/run.ts +++ b/packages/api/src/agents/run.ts @@ -46,6 +46,7 @@ import type { Callbacks } from '@langchain/core/callbacks/manager'; import type { AppConfig, IUser } from '@librechat/data-schemas'; import type { ModelBoundChatModelCallback } from '~/middleware/modelBoundContent'; import type { ToolInputValidationError } from '~/agents/toolValidation'; +import type { ResolvedToolApprovalHook } from '~/agents/hitl/hooks'; import type { ResolvedAlwaysApplySkill } from '~/agents/skills'; import type { MCPToolAlias } from '~/tools/classification'; import type { SubagentUsageEvent } from '~/agents/usage'; @@ -1091,7 +1092,7 @@ export function agentRequestsAskUserQuestion(agent: { * it to the model, attaching checkpointers, and pausing runs — for a run-pausing * tool the filter must be an actual kill switch. */ -function isAskUserQuestionAdminDisabled(appConfig?: AppConfig): boolean { +export function isAskUserQuestionAdminDisabled(appConfig?: AppConfig): boolean { const included = appConfig?.includedTools; if (included != null && included.length > 0) { return !included.includes(ASK_USER_QUESTION_TOOL_NAME); @@ -1392,6 +1393,7 @@ export async function createRun({ activityPhase, eventActorCheckpointing = false, hitlCapable = false, + resolvedToolApprovalHooks, toolInputValidationErrors, sessionStartSource, streaming = true, @@ -1490,6 +1492,11 @@ export async function createRun({ * final response / `[DONE]` with the tool call left unresolved). */ hitlCapable?: boolean; + /** + * Request-scoped approval hooks already resolved by the scheduled-run admission guard. + * Reuse them here so a context-aware factory is evaluated exactly once for the run. + */ + resolvedToolApprovalHooks?: readonly ResolvedToolApprovalHook[]; /** Plugin-hook SessionStart lifecycle source: 'startup' (default) or 'resume' on HITL-rebuild paths. */ sessionStartSource?: string; /** Request-scoped tool input failures consumed by the completion handler. */ @@ -1881,6 +1888,7 @@ export async function createRun({ appConfig, }, mcpToolAliases, + resolvedToolApprovalHooks, ) : undefined; registerResolvedMCPToolAliases = (resolvedAgent) => { diff --git a/packages/api/src/plugins/runtime.spec.ts b/packages/api/src/plugins/runtime.spec.ts index 0a3660a293..42163c435b 100644 --- a/packages/api/src/plugins/runtime.spec.ts +++ b/packages/api/src/plugins/runtime.spec.ts @@ -6,6 +6,7 @@ import { getDeploymentPluginHookCapabilities, registerDeploymentPluginHooks, hasDeploymentPluginHooks, + hasDeploymentPluginToolApprovalHooks, } from './runtime'; import { getPluginHookSource, setPluginHookSource } from '~/agents/hooks'; import { initializeDeploymentPlugins } from './deployment'; @@ -306,6 +307,9 @@ describe('registerDeploymentPluginHooks', () => { }); const source = getPluginHookSource(); expect(source?.hasHooks()).toBe(true); + expect(hasDeploymentPluginToolApprovalHooks()).toBe(true); + expect(hasDeploymentPluginToolApprovalHooks(['unmatched_tool'])).toBe(false); + expect(hasDeploymentPluginToolApprovalHooks(['write_file'])).toBe(true); const registry = new HookRegistry(); expect(source?.register({ registry, context: { sessionId: 'conversation-seam' } })).toBe(4); setPluginHookSource(undefined); diff --git a/packages/api/src/plugins/runtime.ts b/packages/api/src/plugins/runtime.ts index 68cc391812..2b1b926f41 100644 --- a/packages/api/src/plugins/runtime.ts +++ b/packages/api/src/plugins/runtime.ts @@ -1,5 +1,6 @@ import { createHash } from 'node:crypto'; import { logger } from '@librechat/data-schemas'; +import { matchesQuery } from '@librechat/agents'; import type { HookRegistry, HookOutput } from '@librechat/agents'; import type { PluginHookExecutionRequest, @@ -113,6 +114,18 @@ export function hasDeploymentPluginHooks(): boolean { return getExecutableHookPlugins().length > 0; } +/** Whether any executable deployment-plugin hook can return a tool decision. */ +export function hasDeploymentPluginToolApprovalHooks(toolNames?: readonly string[]): boolean { + return getExecutableHookPlugins().some((plugin) => + plugin.hooks?.plan.entries.some((entry) => { + if (entry.status !== 'ready' || entry.targetEvent !== 'PreToolUse') { + return false; + } + return toolNames == null || toolNames.some((name) => matchesQuery(entry.matcher, name)); + }), + ); +} + export interface RegisterDeploymentPluginHooksOptions { registry: HookRegistry; context?: PluginHookRuntimeContext; diff --git a/packages/api/src/stream/ApprovalLifecycle.ts b/packages/api/src/stream/ApprovalLifecycle.ts index 75cbd6dd29..b9bae96e01 100644 --- a/packages/api/src/stream/ApprovalLifecycle.ts +++ b/packages/api/src/stream/ApprovalLifecycle.ts @@ -8,6 +8,7 @@ import type { } from '~/stream/interfaces/IJobStore'; import type { ActivityPhaseSnapshot } from '~/agents/activityPhases/runtime'; import { + JobStatusTransitionDeadlineError, isPendingActionExpired, isPendingActionStale, PAUSE_PERSISTENCE_TIMEOUT_ERROR, @@ -35,6 +36,17 @@ export interface ApprovalPauseOptions { persistencePending?: boolean; } +export const PENDING_ACTION_EXPIRED_CODE = 'HITL_ACTION_EXPIRED'; + +export class PendingActionExpiredError extends Error { + readonly code: typeof PENDING_ACTION_EXPIRED_CODE = PENDING_ACTION_EXPIRED_CODE; + + constructor() { + super('The pending action expired before it could be exposed for review'); + this.name = 'PendingActionExpiredError'; + } +} + const PAUSE_PERSISTENCE_ACTION_PREFIX = 'pause-persistence:'; export function pausePersistenceActionId(actionId: string): string { @@ -85,6 +97,9 @@ export class ApprovalLifecycle { pendingAction: Agents.PendingAction, options: ApprovalPauseOptions = {}, ): Promise { + if (isPendingActionExpired({ pendingAction })) { + throw new PendingActionExpiredError(); + } const job = await this.store.getJob(streamId); if ( !job || @@ -107,28 +122,37 @@ export class ApprovalLifecycle { : 0, ); const persistenceStartedAt = Date.now(); - const ok = await this.store.transitionStatus(streamId, { - from: 'running', - to: 'requires_action', - // pendingActionId is the flat mirror the atomic resolve/expire guard on. - patch: { - pendingAction, - pendingActionId: - options.persistencePending === true - ? pausePersistenceActionId(pendingAction.actionId) - : pendingAction.actionId, - ...(options.persistencePending === true && { - terminalPersistencePending: true, - terminalPersistenceStartedAt: persistenceStartedAt, - }), - ...(discoveredTools != null && discoveredTools.length > 0 - ? { discoveredTools: [...discoveredTools] } - : {}), - ...(activityPhaseSnapshot != null ? { activityPhaseSnapshot } : {}), - }, - expectCreatedAt: expectedCreatedAt, - steerReceiptTtlSeconds: pauseReceiptTtl, - }); + let ok: boolean; + try { + ok = await this.store.transitionStatus(streamId, { + from: 'running', + to: 'requires_action', + // pendingActionId is the flat mirror the atomic resolve/expire guard on. + patch: { + pendingAction, + pendingActionId: + options.persistencePending === true + ? pausePersistenceActionId(pendingAction.actionId) + : pendingAction.actionId, + ...(options.persistencePending === true && { + terminalPersistencePending: true, + terminalPersistenceStartedAt: persistenceStartedAt, + }), + ...(discoveredTools != null && discoveredTools.length > 0 + ? { discoveredTools: [...discoveredTools] } + : {}), + ...(activityPhaseSnapshot != null ? { activityPhaseSnapshot } : {}), + }, + expectCreatedAt: expectedCreatedAt, + notAfterMs: pendingAction.expiresAt, + steerReceiptTtlSeconds: pauseReceiptTtl, + }); + } catch (error) { + if (error instanceof JobStatusTransitionDeadlineError) { + throw new PendingActionExpiredError(); + } + throw error; + } if (ok) { this.callbacks.onPaused?.(streamId, expectedCreatedAt); logger.debug( diff --git a/packages/api/src/stream/__tests__/RedisJobStore.spec.ts b/packages/api/src/stream/__tests__/RedisJobStore.spec.ts index 80ddae19eb..865f60a6dd 100644 --- a/packages/api/src/stream/__tests__/RedisJobStore.spec.ts +++ b/packages/api/src/stream/__tests__/RedisJobStore.spec.ts @@ -200,9 +200,9 @@ describe('RedisJobStore', () => { }), ).resolves.toBe(false); - // ARGV[4] is the terminal job TTL. The persistence owner/recovery path + // ARGV[5] is the terminal job TTL. The persistence owner/recovery path // gets five minutes even when ordinary completed records are immediate. - expect(evalTransition.mock.calls[0][15]).toBe('300'); + expect(evalTransition.mock.calls[0][16]).toBe('300'); }); test('retains the generation epoch beyond the paused job TTL', async () => { @@ -223,9 +223,9 @@ describe('RedisJobStore', () => { expect(transitionCall[0]).toContain( 'redis.call("SET", KEYS[8], currentCreatedAt, "EX", ttl + generationEpochGraceTtl)', ); - expect(transitionCall[15]).toBe('4621'); - expect(transitionCall[16]).toBe('0'); - expect(transitionCall[20]).toBe('300'); + expect(transitionCall[16]).toBe('4621'); + expect(transitionCall[17]).toBe('0'); + expect(transitionCall[21]).toBe('300'); }); test('seeds the guarded epoch when reaping a legacy job without a marker', async () => { @@ -991,9 +991,9 @@ describe('RedisJobStore', () => { evalTransition.mock.invocationCallOrder[0], ); const transitionCall = evalTransition.mock.calls[0]; - expect(transitionCall[15]).toBe('86400'); - expect(transitionCall[17]).toBe('86400'); + expect(transitionCall[16]).toBe('86400'); expect(transitionCall[18]).toBe('86400'); + expect(transitionCall[19]).toBe('86400'); }); test('acknowledges evidence TTLs without removing a successor retry member', async () => { diff --git a/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts b/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts index 204136ca5b..70d62e738d 100644 --- a/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts +++ b/packages/api/src/stream/__tests__/RedisJobStore.stream_integration.spec.ts @@ -4,6 +4,7 @@ import type { Agents } from 'librechat-data-provider'; import type { Redis, Cluster } from 'ioredis'; import type { SteerQueueItem, SteerReceipt } from '../interfaces/IJobStore'; import { + JobStatusTransitionDeadlineError, PAUSE_PERSISTENCE_TIMEOUT_ERROR, STEER_ENQUEUE_RECEIPT_FULL, } from '../interfaces/IJobStore'; @@ -129,6 +130,33 @@ describe('RedisJobStore Integration Tests', () => { await store.destroy(); }); + test('atomically rejects a status transition after its deadline', async () => { + if (!ioredisClient) { + return; + } + + const { RedisJobStore } = await import('../implementations/RedisJobStore'); + const store = new RedisJobStore(ioredisClient); + await store.initialize(); + + const streamId = `test-transition-deadline-${Date.now()}`; + await store.createJob(streamId, 'deadline-user', streamId); + + await expect( + store.transitionStatus(streamId, { + from: 'running', + to: 'requires_action', + notAfterMs: Date.now() - 1, + }), + ).rejects.toMatchObject({ + name: JobStatusTransitionDeadlineError.name, + notAfterMs: expect.any(Number), + }); + await expect(store.getJob(streamId)).resolves.toMatchObject({ status: 'running' }); + + await store.destroy(); + }); + test('returns the exact predecessor captured by the atomic replacement script', async () => { if (!ioredisClient) { return; diff --git a/packages/api/src/stream/__tests__/pendingAction.spec.ts b/packages/api/src/stream/__tests__/pendingAction.spec.ts index faf71428f0..84592dbf0c 100644 --- a/packages/api/src/stream/__tests__/pendingAction.spec.ts +++ b/packages/api/src/stream/__tests__/pendingAction.spec.ts @@ -1,10 +1,10 @@ import type { Agents } from 'librechat-data-provider'; +import { ApprovalLifecycle, PendingActionExpiredError } from '~/stream/ApprovalLifecycle'; import { InMemoryEventTransport } from '~/stream/implementations/InMemoryEventTransport'; import { buildPendingAction, buildToolApprovalPayload } from '~/agents/hitl/policy'; import { PAUSE_PERSISTENCE_TIMEOUT_ERROR } from '~/stream/interfaces/IJobStore'; import { InMemoryJobStore } from '~/stream/implementations/InMemoryJobStore'; import { GenerationJobManagerClass } from '~/stream/GenerationJobManager'; -import { ApprovalLifecycle } from '~/stream/ApprovalLifecycle'; jest.spyOn(console, 'log').mockImplementation(); @@ -60,6 +60,42 @@ describe('ApprovalLifecycle via GenerationJobManager.approvals (in-memory)', () } }); + test('refuses an action whose deadline passed before the pause transition', async () => { + const streamId = 'stream-expired-before-pause'; + await manager.createJob(streamId, 'user-1'); + + await expect( + manager.approvals.pause(streamId, buildAction(streamId, { expiresAt: Date.now() - 1 })), + ).rejects.toBeInstanceOf(PendingActionExpiredError); + await expect(manager.getJob(streamId)).resolves.toMatchObject({ status: 'running' }); + await expect(manager.approvals.peek(streamId)).resolves.toBeNull(); + }); + + test('refuses an action that expires while the pause transition is waiting on storage', async () => { + const streamId = 'stream-expired-during-pause'; + await manager.createJob(streamId, 'user-1'); + const expiresAt = Date.now() + 1000; + const originalTransition = jobStore.transitionStatus.bind(jobStore); + const now = jest.spyOn(Date, 'now').mockReturnValue(expiresAt - 1); + const transition = jest + .spyOn(jobStore, 'transitionStatus') + .mockImplementationOnce(async (...args) => { + now.mockReturnValue(expiresAt); + return originalTransition(...args); + }); + + try { + await expect( + manager.approvals.pause(streamId, buildAction(streamId, { expiresAt })), + ).rejects.toBeInstanceOf(PendingActionExpiredError); + } finally { + transition.mockRestore(); + now.mockRestore(); + } + await expect(manager.getJob(streamId)).resolves.toMatchObject({ status: 'running' }); + await expect(manager.approvals.peek(streamId)).resolves.toBeNull(); + }); + test('a later ask retains a legacy answer for ordered cross-replica reconstruction', async () => { const streamId = 'stream-repause-retains-legacy-answer'; await manager.createJob(streamId, 'user-1'); @@ -610,12 +646,15 @@ describe('ApprovalLifecycle via GenerationJobManager.approvals (in-memory)', () test('treats a past-expiresAt record as gone (lazy expiry)', async () => { const streamId = 'stream-expired-peek'; await manager.createJob(streamId, 'user-1'); - await manager.approvals.pause( - streamId, - buildAction(streamId, { expiresAt: Date.now() - 1000 }), - ); + const expiresAt = Date.now() + 1000; + await manager.approvals.pause(streamId, buildAction(streamId, { expiresAt })); - expect(await manager.approvals.peek(streamId)).toBeNull(); + const now = jest.spyOn(Date, 'now').mockReturnValue(expiresAt + 1); + try { + expect(await manager.approvals.peek(streamId)).toBeNull(); + } finally { + now.mockRestore(); + } }); }); @@ -724,13 +763,16 @@ describe('ApprovalLifecycle via GenerationJobManager.approvals (in-memory)', () test('an expired pending action expires instead of resuming', async () => { const streamId = 'stream-resolve-expired'; await manager.createJob(streamId, 'user-1'); - await manager.approvals.pause( - streamId, - buildAction(streamId, { expiresAt: Date.now() - 1000 }), - ); + const expiresAt = Date.now() + 1000; + await manager.approvals.pause(streamId, buildAction(streamId, { expiresAt })); - expect(await manager.approvals.resolve(streamId)).toBe(false); - expect(await manager.getJobStatus(streamId)).toBe('aborted'); + const now = jest.spyOn(Date, 'now').mockReturnValue(expiresAt + 1); + try { + expect(await manager.approvals.resolve(streamId)).toBe(false); + expect(await manager.getJobStatus(streamId)).toBe('aborted'); + } finally { + now.mockRestore(); + } }); }); @@ -813,13 +855,19 @@ describe('ApprovalLifecycle via GenerationJobManager.approvals (in-memory)', () scheduledFor: '2026-08-17T12:00:00.000Z', }); const lifecycle = new ApprovalLifecycle(jobStore); - await lifecycle.pause(streamId, buildAction(streamId, { expiresAt: Date.now() - 1 })); + const expiresAt = Date.now() + 1_000; + await lifecycle.pause(streamId, buildAction(streamId, { expiresAt })); const onApprovalExpired = jest.fn(async () => undefined); manager.setApprovalExpiredHandler(onApprovalExpired); - await ( - manager as unknown as { expireStaleApprovals: () => Promise } - ).expireStaleApprovals(); + const now = jest.spyOn(Date, 'now').mockReturnValue(expiresAt + 1); + try { + await ( + manager as unknown as { expireStaleApprovals: () => Promise } + ).expireStaleApprovals(); + } finally { + now.mockRestore(); + } expect(onApprovalExpired).toHaveBeenCalledWith( streamId, @@ -1182,13 +1230,16 @@ describe('ApprovalLifecycle via GenerationJobManager.approvals (in-memory)', () test('excludes a pending-approval job whose prompt has expired', async () => { const streamId = 'stream-expired-active'; await manager.createJob(streamId, 'user-exp'); - await manager.approvals.pause( - streamId, - buildAction(streamId, { expiresAt: Date.now() - 1000 }), - ); + const expiresAt = Date.now() + 1000; + await manager.approvals.pause(streamId, buildAction(streamId, { expiresAt })); - // Still requires_action, but the prompt is past expiry → no longer active. - expect(await manager.getActiveJobIdsForUser('user-exp')).not.toContain(streamId); + const now = jest.spyOn(Date, 'now').mockReturnValue(expiresAt + 1); + try { + // Still requires_action, but the prompt is past expiry → no longer active. + expect(await manager.getActiveJobIdsForUser('user-exp')).not.toContain(streamId); + } finally { + now.mockRestore(); + } }); }); }); diff --git a/packages/api/src/stream/implementations/InMemoryJobStore.ts b/packages/api/src/stream/implementations/InMemoryJobStore.ts index 2f186e8e7d..a2e67b2be7 100644 --- a/packages/api/src/stream/implementations/InMemoryJobStore.ts +++ b/packages/api/src/stream/implementations/InMemoryJobStore.ts @@ -23,6 +23,7 @@ import type { } from '~/stream/interfaces/IJobStore'; import type { RecoveredSteerPayload } from '~/stream/SteerRecovery'; import { + JobStatusTransitionDeadlineError, JobPredecessorMismatchError, STEER_ENQUEUE_NOT_RUNNING, STEER_ENQUEUE_QUEUE_FULL, @@ -770,6 +771,9 @@ export class InMemoryJobStore implements IJobStoreV2 { if (args.expectCreatedAt != null && job.createdAt !== args.expectCreatedAt) { return false; } + if (args.notAfterMs != null && Date.now() >= args.notAfterMs) { + throw new JobStatusTransitionDeadlineError(args.notAfterMs); + } if (['complete', 'error', 'aborted'].includes(args.to)) { if (!this.isParkedRecoveryCompatible(streamId, job)) { return false; diff --git a/packages/api/src/stream/implementations/RedisJobStore.ts b/packages/api/src/stream/implementations/RedisJobStore.ts index 3e430e6868..63769e9568 100644 --- a/packages/api/src/stream/implementations/RedisJobStore.ts +++ b/packages/api/src/stream/implementations/RedisJobStore.ts @@ -28,6 +28,7 @@ import type { ResolvedAskUserQuestion } from '~/agents/hitl/resume'; import type { RecoveredSteerPayload } from '~/stream/SteerRecovery'; import { JobCreationSupersededError, + JobStatusTransitionDeadlineError, JobPredecessorMismatchError, STEER_ENQUEUE_NOT_RUNNING, STEER_QUEUE_MAX_DEPTH, @@ -107,6 +108,7 @@ function assertCreateIdempotencyArguments( * from, * expectActionId | "", * expectCreatedAt | "", + * notAfterMs | "", * ttl, * terminal ("0" | "1"), * chunksAfterComplete, @@ -124,14 +126,17 @@ const JOB_CAS_LUA = 'if redis.call("HGET", KEYS[1], "status") ~= ARGV[1] then return 0 end ' + 'if ARGV[2] ~= "" and redis.call("HGET", KEYS[1], "pendingActionId") ~= ARGV[2] then return 0 end ' + 'if ARGV[3] ~= "" and redis.call("HGET", KEYS[1], "createdAt") ~= ARGV[3] then return 0 end ' + + 'if ARGV[4] ~= "" then local now = redis.call("TIME") ' + + 'local nowMs = tonumber(now[1]) * 1000 + math.floor(tonumber(now[2]) / 1000) ' + + 'if nowMs >= tonumber(ARGV[4]) then return -1 end end ' + 'local currentCreatedAt = redis.call("HGET", KEYS[1], "createdAt") ' + - 'local ttl = tonumber(ARGV[4]) ' + - 'local terminal = ARGV[5] == "1" ' + - 'local chunksTtl = tonumber(ARGV[6]) ' + - 'local runStepsTtl = tonumber(ARGV[7]) ' + - 'local parkedTtl = tonumber(ARGV[8]) ' + - 'local generationEpochGraceTtl = tonumber(ARGV[9]) ' + - 'local receiptTtl = tonumber(ARGV[10]) ' + + 'local ttl = tonumber(ARGV[5]) ' + + 'local terminal = ARGV[6] == "1" ' + + 'local chunksTtl = tonumber(ARGV[7]) ' + + 'local runStepsTtl = tonumber(ARGV[8]) ' + + 'local parkedTtl = tonumber(ARGV[9]) ' + + 'local generationEpochGraceTtl = tonumber(ARGV[10]) ' + + 'local receiptTtl = tonumber(ARGV[11]) ' + 'local ownerUserId = redis.call("HGET", KEYS[1], "userId") ' + 'local ownerTenantId = redis.call("HGET", KEYS[1], "tenantId") ' + 'local generationProtocol = redis.call("HGET", KEYS[1], "generationProtocolVersion") == "2" and 2 or 1 ' + @@ -163,8 +168,8 @@ const JOB_CAS_LUA = 'or (item.createdAt and (type(item.createdAt) ~= "number" or item.createdAt < 0)) ' + 'or (item.recoveringCreatedAt and (type(item.recoveringCreatedAt) ~= "number" or item.recoveringCreatedAt < 0)) then return 0 end ' + 'validatedPrior[#validatedPrior + 1] = item end end end ' + - 'local hdelCount = tonumber(ARGV[12]) ' + - 'local idx = 13 ' + + 'local hdelCount = tonumber(ARGV[13]) ' + + 'local idx = 14 ' + 'for i = 1, hdelCount do redis.call("HDEL", KEYS[1], ARGV[idx]) idx = idx + 1 end ' + 'local hset = {} ' + 'for i = idx, #ARGV do hset[#hset + 1] = ARGV[i] end ' + @@ -227,7 +232,7 @@ const JOB_CAS_LUA = 'redis.call("DEL", KEYS[5], KEYS[6]) ' + 'if chunksTtl == 0 then redis.call("DEL", KEYS[3]) else redis.call("EXPIRE", KEYS[3], chunksTtl) end ' + 'if runStepsTtl == 0 then redis.call("DEL", KEYS[4]) else redis.call("EXPIRE", KEYS[4], runStepsTtl) end ' + - 'if ARGV[11] == "1" then if #items == 0 then return "[]" end return cjson.encode(items) end ' + + 'if ARGV[12] == "1" then if #items == 0 then return "[]" end return cjson.encode(items) end ' + 'else ' + 'redis.call("EXPIRE", KEYS[3], ttl) ' + 'redis.call("EXPIRE", KEYS[4], ttl) ' + @@ -2514,7 +2519,7 @@ export class RedisJobStore implements IJobStoreV2 { args: JobStatusTransition, returnDrainedSteers: boolean, ): Promise { - const { from, to, patch, clear, expectActionId, expectCreatedAt } = args; + const { from, to, patch, clear, expectActionId, expectCreatedAt, notAfterMs } = args; const key = KEYS.job(streamId); // status + patch become HSET pairs; serializeJob skips undefined, so @@ -2573,6 +2578,7 @@ export class RedisJobStore implements IJobStoreV2 { from, expectActionId ?? '', expectCreatedAt != null ? String(expectCreatedAt) : '', + notAfterMs != null ? String(notAfterMs) : '', String(ttl), terminal ? '1' : '0', String( @@ -2591,6 +2597,9 @@ export class RedisJobStore implements IJobStoreV2 { ...clearFields, ...fields, ); + if (result === -1 && notAfterMs != null) { + throw new JobStatusTransitionDeadlineError(notAfterMs); + } if (returnDrainedSteers ? typeof result !== 'string' : result !== 1) { return null; } diff --git a/packages/api/src/stream/index.ts b/packages/api/src/stream/index.ts index 61f788fc05..3c745f8f10 100644 --- a/packages/api/src/stream/index.ts +++ b/packages/api/src/stream/index.ts @@ -38,6 +38,12 @@ export { STEER_QUEUE_MAX_DEPTH, } from './interfaces/IJobStore'; export { SteeringLifecycle, toPendingSteer } from './SteeringLifecycle'; +export { + ApprovalLifecycle, + PendingActionExpiredError, + PENDING_ACTION_EXPIRED_CODE, +} from './ApprovalLifecycle'; +export type { ApprovalLifecycleCallbacks, ApprovalPauseOptions } from './ApprovalLifecycle'; export { assertJobStoreV2, getMissingJobStoreV2Methods, diff --git a/packages/api/src/stream/interfaces/IJobStore.ts b/packages/api/src/stream/interfaces/IJobStore.ts index 0f64820664..41aa3eaf23 100644 --- a/packages/api/src/stream/interfaces/IJobStore.ts +++ b/packages/api/src/stream/interfaces/IJobStore.ts @@ -565,6 +565,17 @@ export const STEER_ENQUEUE_QUEUE_FULL = -2; * so idempotency evidence is never evicted inside its recovery window. */ export const STEER_ENQUEUE_RECEIPT_FULL = -3; +/** The store rejected a status CAS because its atomic deadline had elapsed. */ +export class JobStatusTransitionDeadlineError extends Error { + readonly notAfterMs: number; + + constructor(notAfterMs: number) { + super('The status transition deadline elapsed before the transition could commit'); + this.name = 'JobStatusTransitionDeadlineError'; + this.notAfterMs = notAfterMs; + } +} + /** * Arguments for an atomic {@link IJobStore.transitionStatus} compare-and-set. */ @@ -589,6 +600,12 @@ export interface JobStatusTransition { * the same stream ID. */ expectCreatedAt?: number; + /** + * Additional guard: reject the transition when the store's clock has reached + * this absolute deadline. The comparison is part of the same atomic operation + * as the status change, so queueing or storage latency cannot publish stale state. + */ + notAfterMs?: number; /** Extend all current steer receipts in the SAME atomic step as this * transition. Used by running→requires_action so no enqueue can land between * a pre-pause TTL pass and the status CAS. */ diff --git a/packages/data-provider/src/config.ts b/packages/data-provider/src/config.ts index f7c6d81154..1d517f5889 100644 --- a/packages/data-provider/src/config.ts +++ b/packages/data-provider/src/config.ts @@ -903,7 +903,7 @@ export type ToolApprovalMode = z.infer; * * Shape mirrors `@librechat/agents`'s `ToolPolicyConfig` so the host can map it * directly into `createToolPolicyHook(config)`. The SDK does the evaluation - * (`deny → bypass → allow → ask → dontAsk → fallthrough(ask)`); this config + * (`deny → ask → allow → bypass → dontAsk → fallthrough(ask)`); this config * just describes the surface. * * Conventions: