From 8969ee4b187b3170fedac3a1072c491ea67e1683 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Sun, 23 Aug 2026 02:37:33 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=8E=9A=EF=B8=8F=20feat:=20Configure=20Age?= =?UTF-8?q?nt=20Event=20Runtime=20in=20YAML=20(#15128)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 28 ----------- api/server/index.js | 2 + .../middleware/limiters/messageLimiters.js | 48 ++++++++++--------- .../limiters/messageLimiters.spec.js | 26 ++++++++++ .../Endpoints/agents/subagentThreadStore.js | 20 ++++---- .../agents/subagentThreadStore.spec.js | 29 +++++++++-- librechat.example.yaml | 12 +++++ packages/api/src/agents/triggers/README.md | 10 ++-- packages/api/src/app/agents.spec.ts | 40 ++++++++++++++++ packages/api/src/app/agents.ts | 18 +++++++ packages/api/src/app/checks.spec.ts | 12 +++++ packages/api/src/app/index.ts | 1 + packages/api/src/app/limits.ts | 1 + packages/data-provider/src/config.spec.ts | 34 +++++++++++++ packages/data-provider/src/config.ts | 17 +++++++ 15 files changed, 233 insertions(+), 65 deletions(-) create mode 100644 api/server/middleware/limiters/messageLimiters.spec.js create mode 100644 packages/api/src/app/agents.spec.ts create mode 100644 packages/api/src/app/agents.ts diff --git a/.env.example b/.env.example index 188b1efb49..4d47b6b064 100644 --- a/.env.example +++ b/.env.example @@ -724,15 +724,6 @@ LIMIT_MESSAGE_USER=false MESSAGE_USER_MAX=40 MESSAGE_USER_WINDOW=1 -# Authenticated agent-event admission uses a separate API-key bucket because -# delivery execution later consumes the normal message-user limit. -AGENT_EVENT_USER_MAX=40 -AGENT_EVENT_USER_WINDOW=1 - -# Enable only after every API replica runs a release that understands -# source-bound child actor continuations. -ENABLE_AGENT_EVENT_CHILD_TURNS=false - ILLEGAL_MODEL_REQ_SCORE=5 #========================# @@ -1221,25 +1212,6 @@ OPENWEATHER_API_KEY= # or # COHERE_API_KEY=your_cohere_api_key -#===========================# -# Agent Trigger Delivery # -#===========================# - -# Base URL used by trusted in-process event producers to dispatch agent fires, continuations, -# and steers. -# Defaults to this process's bound listener. Set only when internal trigger admission must -# traverse another trusted HTTP origin, such as a TLS front door. -# AGENT_TRIGGERS_SELF_URL=http://127.0.0.1:3080 - -# Automatically continue a saved parent agent after a detached subagent settles. -# Rolling-deploy safety: deploy support with this disabled first, wait until every API -# replica is upgraded, then enable it in a subsequent rollout. -# ENABLE_SUBAGENT_COMPLETION_WAKEUPS=false - -# Trusted event adapters enqueue through the shared durable trigger service. Mongo-backed -# leases make its workers safe across replicas; successful delivery records expire after -# 90 days, while dead letters remain available for explicit operator requeue. - #======================# # MCP Configuration # #======================# diff --git a/api/server/index.js b/api/server/index.js index 539836cef6..1c4a91244f 100644 --- a/api/server/index.js +++ b/api/server/index.js @@ -43,6 +43,7 @@ const { updateInterfacePermissions, configureMessageFilterRegexValidator, configureFileConfigRegexEngine, + configureAgentEventRuntime, createScheduleWriteGate, waitForKeyvRedisClient, } = require('@librechat/api'); @@ -175,6 +176,7 @@ const startServer = async () => { logger.error('[sweepOrphanedPreviews] Background sweep failed:', err); }); const appConfig = await getAppConfig({ baseOnly: true }); + configureAgentEventRuntime(appConfig?.endpoints?.agents?.eventDriven); initializeFileStorage(appConfig); const projectRoot = path.resolve(__dirname, '../..'); // Plugin hooks execute only when the operator opts in via DEPLOYMENT_PLUGIN_HOOKS; diff --git a/api/server/middleware/limiters/messageLimiters.js b/api/server/middleware/limiters/messageLimiters.js index 2e05820032..630be717d8 100644 --- a/api/server/middleware/limiters/messageLimiters.js +++ b/api/server/middleware/limiters/messageLimiters.js @@ -9,8 +9,6 @@ const { MESSAGE_IP_WINDOW = 1, MESSAGE_USER_MAX = 40, MESSAGE_USER_WINDOW = 1, - AGENT_EVENT_USER_MAX = 40, - AGENT_EVENT_USER_WINDOW = 1, MESSAGE_VIOLATION_SCORE: score, } = process.env; @@ -22,10 +20,6 @@ const userWindowMs = MESSAGE_USER_WINDOW * 60 * 1000; const userMax = MESSAGE_USER_MAX; const userWindowInMinutes = userWindowMs / 60000; -const agentEventUserWindowMs = AGENT_EVENT_USER_WINDOW * 60 * 1000; -const agentEventUserMax = AGENT_EVENT_USER_MAX; -const agentEventUserWindowInMinutes = agentEventUserWindowMs / 60000; - /** * Creates either an IP/User message request rate limiter for excessive requests * that properly logs and denies the violation. @@ -85,23 +79,31 @@ const messageUserLimiter = rateLimit(userLimiterOptions); * consumes the normal message-user bucket when it executes the delivery, so * sharing that limiter here would charge every event twice. */ -const agentEventUserLimiter = rateLimit({ - windowMs: agentEventUserWindowMs, - max: agentEventUserMax, - handler: async (req, res) => { - const type = ViolationTypes.MESSAGE_LIMIT; - const errorMessage = { - type, - max: agentEventUserMax, - limiter: 'agent_event_principal', - windowInMinutes: agentEventUserWindowInMinutes, - }; - await logViolation(req, res, type, errorMessage, score); - return await denyRequest(req, res, errorMessage); - }, - keyGenerator: (req) => String(req.apiKeyId ?? req.user?.id), - store: limiterCache('agent_event_user_limiter'), -}); +let configuredAgentEventUserLimiter; +const agentEventUserLimiter = (req, res, next) => { + if (configuredAgentEventUserLimiter == null) { + const max = Number(process.env.AGENT_EVENT_USER_MAX ?? 40); + const windowInMinutes = Number(process.env.AGENT_EVENT_USER_WINDOW ?? 1); + configuredAgentEventUserLimiter = rateLimit({ + windowMs: windowInMinutes * 60 * 1000, + max, + handler: async (limitedReq, limitedRes) => { + const type = ViolationTypes.MESSAGE_LIMIT; + const errorMessage = { + type, + max, + limiter: 'agent_event_principal', + windowInMinutes, + }; + await logViolation(limitedReq, limitedRes, type, errorMessage, score); + return await denyRequest(limitedReq, limitedRes, errorMessage); + }, + keyGenerator: (limitedReq) => String(limitedReq.apiKeyId ?? limitedReq.user?.id), + store: limiterCache('agent_event_user_limiter'), + }); + } + return configuredAgentEventUserLimiter(req, res, next); +}; module.exports = { agentEventUserLimiter, diff --git a/api/server/middleware/limiters/messageLimiters.spec.js b/api/server/middleware/limiters/messageLimiters.spec.js new file mode 100644 index 0000000000..c2ff78be62 --- /dev/null +++ b/api/server/middleware/limiters/messageLimiters.spec.js @@ -0,0 +1,26 @@ +const mockLimiter = jest.fn((_req, _res, next) => next()); +const mockRateLimit = jest.fn(() => mockLimiter); + +jest.mock('express-rate-limit', () => mockRateLimit); +jest.mock('@librechat/api', () => ({ + limiterCache: jest.fn(() => ({})), + removePorts: jest.fn(), +})); +jest.mock('~/server/middleware/denyRequest', () => jest.fn()); +jest.mock('~/cache', () => ({ logViolation: jest.fn() })); + +describe('agent event rate limiter', () => { + it('reads YAML-projected limits lazily after startup configuration', () => { + process.env.AGENT_EVENT_USER_MAX = '80'; + process.env.AGENT_EVENT_USER_WINDOW = '2'; + const { agentEventUserLimiter } = require('./messageLimiters'); + const next = jest.fn(); + + agentEventUserLimiter({ apiKeyId: 'key-1' }, {}, next); + + expect(mockRateLimit).toHaveBeenLastCalledWith( + expect.objectContaining({ max: 80, windowMs: 120_000 }), + ); + expect(mockLimiter).toHaveBeenCalledWith({ apiKeyId: 'key-1' }, {}, next); + }); +}); diff --git a/api/server/services/Endpoints/agents/subagentThreadStore.js b/api/server/services/Endpoints/agents/subagentThreadStore.js index a42e9cee84..fd1f406178 100644 --- a/api/server/services/Endpoints/agents/subagentThreadStore.js +++ b/api/server/services/Endpoints/agents/subagentThreadStore.js @@ -14,12 +14,10 @@ const { const db = require('~/models'); const { enqueueAgentTrigger } = require('../../Agents/triggers'); -/** Keep producers off for the first rollout so older trigger workers cannot - * permanently reject the new `continue` envelope. Enable only after every API - * replica runs a release that understands completion wakeups. */ -const completionWakeupsEnabled = isEnabled(process.env.ENABLE_SUBAGENT_COMPLETION_WAKEUPS); const GENERATION_DRAIN_TIMEOUT_MS = 45_000; const GENERATION_DRAIN_POLL_MS = 100; +const completionWakeupHandler = createSubagentCompletionWakeupHandler(enqueueAgentTrigger); +const completionWakeupsEnabled = () => isEnabled(process.env.ENABLE_SUBAGENT_COMPLETION_WAKEUPS); async function cancelUnroutedGeneration({ userId, tenantId, taskId }) { let job = await GenerationJobManager.getJob(taskId); @@ -78,9 +76,12 @@ const subagentThreadTaskStore = createSubagentThreadTaskStore( renewOwnerAdmission: db.renewSubagentAdmission, releaseOwnerAdmission: db.releaseSubagentAdmission, cancelUnroutedTask: cancelUnroutedGeneration, - ...(completionWakeupsEnabled && { - onTaskPrepared: createSubagentCompletionWakeupHandler(enqueueAgentTrigger), - }), + onTaskPrepared: (registration) => { + if (!completionWakeupsEnabled()) { + return; + } + return completionWakeupHandler(registration); + }, }, ); @@ -138,5 +139,8 @@ async function configureSubagentTaskRouting() { } module.exports = subagentThreadTaskStore; -module.exports.completionWakeupsEnabled = completionWakeupsEnabled; +Object.defineProperty(module.exports, 'completionWakeupsEnabled', { + enumerable: true, + get: completionWakeupsEnabled, +}); module.exports.configureSubagentTaskRouting = configureSubagentTaskRouting; diff --git a/api/server/services/Endpoints/agents/subagentThreadStore.spec.js b/api/server/services/Endpoints/agents/subagentThreadStore.spec.js index a75ae5c2a8..615661dd52 100644 --- a/api/server/services/Endpoints/agents/subagentThreadStore.spec.js +++ b/api/server/services/Endpoints/agents/subagentThreadStore.spec.js @@ -5,6 +5,7 @@ const mockTaskStore = { destroyTaskControlTransport: jest.fn().mockResolvedValue(undefined), destroyActivityStream: jest.fn(), }; +const mockCompletionWakeupHandler = jest.fn().mockResolvedValue(undefined); jest.mock('@librechat/api', () => ({ cacheConfig: { USE_REDIS: true, REDIS_KEY_PREFIX: 'test:' }, @@ -13,7 +14,7 @@ jest.mock('@librechat/api', () => ({ registerShutdownTask: jest.fn(), duplicateIoRedisClient: jest.fn(), createSubagentThreadTaskStore: jest.fn(() => mockTaskStore), - createSubagentCompletionWakeupHandler: jest.fn(), + createSubagentCompletionWakeupHandler: jest.fn(() => mockCompletionWakeupHandler), RedisSubagentTaskControlTransport: jest.fn(), RedisEventTransport: jest.fn(), SubagentActivityStream: jest.fn(), @@ -44,13 +45,35 @@ jest.mock('../../Agents/triggers', () => ({ enqueueAgentTrigger: jest.fn(), })); -const { ioredisClient, registerShutdownTask, duplicateIoRedisClient } = require('@librechat/api'); -const { configureSubagentTaskRouting } = require('./subagentThreadStore'); +const { + ioredisClient, + isEnabled, + registerShutdownTask, + duplicateIoRedisClient, + createSubagentThreadTaskStore, +} = require('@librechat/api'); +const subagentThreadTaskStore = require('./subagentThreadStore'); +const { configureSubagentTaskRouting } = subagentThreadTaskStore; +const taskStoreOptions = createSubagentThreadTaskStore.mock.calls[0][1]; const activityPrepareRegistration = registerShutdownTask.mock.calls.find( ([name]) => name === 'subagent activity streams prepare', ); describe('subagent thread Redis lifecycle', () => { + it('reads completion wakeup rollout state at task preparation time', async () => { + isEnabled.mockReturnValueOnce(false); + + await taskStoreOptions.onTaskPrepared({ taskId: 'disabled' }); + expect(mockCompletionWakeupHandler).not.toHaveBeenCalled(); + isEnabled.mockReturnValueOnce(true); + await taskStoreOptions.onTaskPrepared({ taskId: 'enabled' }); + expect(mockCompletionWakeupHandler).toHaveBeenCalledWith({ taskId: 'enabled' }); + isEnabled.mockReturnValueOnce(true); + expect(subagentThreadTaskStore.completionWakeupsEnabled).toBe(true); + isEnabled.mockReturnValueOnce(false); + expect(subagentThreadTaskStore.completionWakeupsEnabled).toBe(false); + }); + it('closes activity SSE before drain and disconnects its subscriber after drain', async () => { const taskSubscriber = { disconnect: jest.fn() }; const activitySubscriber = { disconnect: jest.fn() }; diff --git a/librechat.example.yaml b/librechat.example.yaml index 2c1fd6bcf3..0128c56214 100644 --- a/librechat.example.yaml +++ b/librechat.example.yaml @@ -361,6 +361,10 @@ registration: # - '127.0.0.1:8080' # rateLimits: +# # Authenticated agent-event admission has its own API-key-principal bucket. +# agentEvents: +# userMax: 40 +# userWindowInMinutes: 1 # fileUploads: # ipMax: 100 # ipWindowInMinutes: 60 # Rate limit window for file uploads per IP @@ -559,6 +563,14 @@ endpoints: # # (optional) Limit the workspace scopes users may select. Omit to allow all three. # statefulCodeSessions: # allowedEnvironments: ["user", "agent-user", "conversation"] + # # (optional) Process-wide event-driven agent rollout controls. Deploy support to + # # every API replica before enabling either producer during a subsequent rollout. + # eventDriven: + # childTurns: false + # completionWakeups: false + # # Optional trusted origin for internal trigger delivery. By default, LibreChat + # # uses its own bound listener; override only for an internal TLS/front-door route. + # # selfUrl: 'https://librechat.internal' # # "run_in_background" makes Code Interpreter tools eligible by default and enables per-tool MCP opt-in. # # "tool_intents" enables live model-written labels for native tools and opted-in MCP tools. # # (optional) Require user approval before matching tool calls. Disabled by default. diff --git a/packages/api/src/agents/triggers/README.md b/packages/api/src/agents/triggers/README.md index 3c15c163c6..cc1d919c0a 100644 --- a/packages/api/src/agents/triggers/README.md +++ b/packages/api/src/agents/triggers/README.md @@ -114,9 +114,13 @@ Register a direct child agent once under the same Remote Agents API key that wil The parent must be an ordinary agent conversation, and the target must be enabled in that parent agent's direct `subagents.agent_ids` list (or be an allowed self-spawn). The reserved child conversation is hidden from conversation lists and remains read-only to human chat routes. -`ENABLE_AGENT_EVENT_CHILD_TURNS` defaults to false; enable it only after every API replica runs a -release that understands bound child continuations, otherwise an older worker could permanently -reject a new envelope during a rolling deployment. +`endpoints.agents.eventDriven.childTurns` defaults to false; enable it only after every API replica +runs a release that understands bound child continuations, otherwise an older worker could +permanently reject a new envelope during a rolling deployment. The legacy +`ENABLE_AGENT_EVENT_CHILD_TURNS` environment variable remains a compatibility fallback. +`AGENT_TRIGGERS_SELF_URL` likewise remains a compatibility fallback for +`endpoints.agents.eventDriven.selfUrl`; most deployments should omit both and use the bound +listener. ```http POST /api/agents/v1/events/bindings diff --git a/packages/api/src/app/agents.spec.ts b/packages/api/src/app/agents.spec.ts new file mode 100644 index 0000000000..7462451373 --- /dev/null +++ b/packages/api/src/app/agents.spec.ts @@ -0,0 +1,40 @@ +import { configureAgentEventRuntime } from './agents'; + +describe('configureAgentEventRuntime', () => { + const originalEnvironment = process.env; + + beforeEach(() => { + process.env = { ...originalEnvironment }; + delete process.env.ENABLE_AGENT_EVENT_CHILD_TURNS; + delete process.env.ENABLE_SUBAGENT_COMPLETION_WAKEUPS; + delete process.env.AGENT_TRIGGERS_SELF_URL; + }); + + afterAll(() => { + process.env = originalEnvironment; + }); + + it('projects explicit base-config rollout flags into the legacy runtime seam', () => { + configureAgentEventRuntime({ + childTurns: true, + completionWakeups: false, + selfUrl: 'https://triggers.internal', + }); + + expect(process.env.ENABLE_AGENT_EVENT_CHILD_TURNS).toBe('true'); + expect(process.env.ENABLE_SUBAGENT_COMPLETION_WAKEUPS).toBe('false'); + expect(process.env.AGENT_TRIGGERS_SELF_URL).toBe('https://triggers.internal'); + }); + + it('preserves environment fallbacks when the YAML fields are omitted', () => { + process.env.ENABLE_AGENT_EVENT_CHILD_TURNS = 'true'; + process.env.ENABLE_SUBAGENT_COMPLETION_WAKEUPS = 'true'; + process.env.AGENT_TRIGGERS_SELF_URL = 'https://legacy.internal'; + + configureAgentEventRuntime(undefined); + + expect(process.env.ENABLE_AGENT_EVENT_CHILD_TURNS).toBe('true'); + expect(process.env.ENABLE_SUBAGENT_COMPLETION_WAKEUPS).toBe('true'); + expect(process.env.AGENT_TRIGGERS_SELF_URL).toBe('https://legacy.internal'); + }); +}); diff --git a/packages/api/src/app/agents.ts b/packages/api/src/app/agents.ts new file mode 100644 index 0000000000..d90c2d293b --- /dev/null +++ b/packages/api/src/app/agents.ts @@ -0,0 +1,18 @@ +import type { TAgentsEndpoint } from 'librechat-data-provider'; + +type AgentEventRuntimeConfig = NonNullable; + +const setBooleanEnvironmentFallback = (name: string, value?: boolean): void => { + if (value != null) { + process.env[name] = String(value); + } +}; + +/** Applies base-config rollout flags before the HTTP listener accepts agent events. */ +export const configureAgentEventRuntime = (config?: AgentEventRuntimeConfig): void => { + setBooleanEnvironmentFallback('ENABLE_AGENT_EVENT_CHILD_TURNS', config?.childTurns); + setBooleanEnvironmentFallback('ENABLE_SUBAGENT_COMPLETION_WAKEUPS', config?.completionWakeups); + if (config?.selfUrl != null) { + process.env.AGENT_TRIGGERS_SELF_URL = config.selfUrl; + } +}; diff --git a/packages/api/src/app/checks.spec.ts b/packages/api/src/app/checks.spec.ts index cab5b727f9..0d570acbb7 100644 --- a/packages/api/src/app/checks.spec.ts +++ b/packages/api/src/app/checks.spec.ts @@ -355,4 +355,16 @@ describe('handleRateLimits', () => { expect(process.env.STT_USER_MAX).toEqual('30'); expect(process.env.STT_USER_WINDOW).toEqual('20'); }); + + it('should set authenticated agent-event admission limits', () => { + handleRateLimits({ + agentEvents: { + userMax: 80, + userWindowInMinutes: 2, + }, + }); + + expect(process.env.AGENT_EVENT_USER_MAX).toEqual('80'); + expect(process.env.AGENT_EVENT_USER_WINDOW).toEqual('2'); + }); }); diff --git a/packages/api/src/app/index.ts b/packages/api/src/app/index.ts index 8dff0f8f9f..bd924896eb 100644 --- a/packages/api/src/app/index.ts +++ b/packages/api/src/app/index.ts @@ -8,5 +8,6 @@ export * from './resolve'; export * from './shutdown'; export * from './server'; export * from './origin'; +export * from './agents'; export { resolveBuildInfo } from './build'; export type { BuildInfo } from './build'; diff --git a/packages/api/src/app/limits.ts b/packages/api/src/app/limits.ts index 1f1ad583f5..809c47d8f2 100644 --- a/packages/api/src/app/limits.ts +++ b/packages/api/src/app/limits.ts @@ -11,6 +11,7 @@ export const handleRateLimits = (rateLimits?: TCustomConfig['rateLimits']): void } const rateLimitKeys = { + agentEvents: 'AGENT_EVENT', fileUploads: RateLimitPrefix.FILE_UPLOAD, conversationsImport: RateLimitPrefix.IMPORT, tts: RateLimitPrefix.TTS, diff --git a/packages/data-provider/src/config.spec.ts b/packages/data-provider/src/config.spec.ts index b06064ceab..bdaab9367a 100644 --- a/packages/data-provider/src/config.spec.ts +++ b/packages/data-provider/src/config.spec.ts @@ -61,6 +61,40 @@ describe('bedrockEndpointSchema', () => { }); }); +describe('agent event runtime config', () => { + it('accepts rollout flags and agent-event admission limits', () => { + const result = configSchema.safeParse({ + version: '1.0', + endpoints: { + agents: { + eventDriven: { + childTurns: true, + completionWakeups: false, + selfUrl: 'https://triggers.internal', + }, + }, + }, + rateLimits: { + agentEvents: { userMax: 80, userWindowInMinutes: 2 }, + }, + }); + + expect(result.success).toBe(true); + if (!result.success) { + return; + } + expect(result.data.endpoints?.agents?.eventDriven).toEqual({ + childTurns: true, + completionWakeups: false, + selfUrl: 'https://triggers.internal', + }); + expect(result.data.rateLimits?.agentEvents).toEqual({ + userMax: 80, + userWindowInMinutes: 2, + }); + }); +}); + describe('speechTab schema', () => { it.each(['browser', 'external', 'openai', 'azureOpenAI'])( 'accepts the speech-to-text engine "%s"', diff --git a/packages/data-provider/src/config.ts b/packages/data-provider/src/config.ts index e512f36da4..6e91fc2452 100644 --- a/packages/data-provider/src/config.ts +++ b/packages/data-provider/src/config.ts @@ -1039,6 +1039,17 @@ export const agentsEndpointSchema = baseEndpointSchema allowedEnvironments: z.array(z.enum(STATEFUL_CODE_ENVIRONMENTS)).min(1), }) .optional(), + /** Process-wide event-driven agent rollout controls. Configure these from the base + * deployment config only so every API replica exposes the same wire capabilities. */ + eventDriven: z + .object({ + childTurns: z.boolean().optional(), + completionWakeups: z.boolean().optional(), + /** Optional trusted origin for in-process trigger delivery. The bound + * listener remains the default and is safer for most deployments. */ + selfUrl: z.string().url().optional(), + }) + .optional(), skills: z .object({ maxCatalogSkills: z.number().int().min(1).max(100).optional(), @@ -1394,6 +1405,12 @@ export enum RateLimitPrefix { } export const rateLimitSchema = z.object({ + agentEvents: z + .object({ + userMax: z.number().int().positive().optional(), + userWindowInMinutes: z.number().positive().optional(), + }) + .optional(), fileUploads: z .object({ ipMax: z.number().optional(),