🎚️ feat: Configure Agent Event Runtime in YAML (#15128)

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;

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,

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);
+  });
+});

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;

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() };