From 3ebef4c84e04cd94f39fb74886a1f643f16e60db Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Sat, 22 Aug 2026 01:09:04 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=93=A8=20feat:=20Add=20Authenticated=20Ag?= =?UTF-8?q?ent=20Event=20Ingress=20(#15110)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: add authenticated agent event ingress * style: sort agent ingress imports * fix: harden agent event ingress * fix: bind event provenance to API keys * fix: inspect event input with legacy PII filters * fix: scope event status reads to source keys * fix: bind event status reads to remote sources --- .env.example | 5 + .../middleware/limiters/messageLimiters.js | 30 ++ .../routes/agents/__tests__/abort.spec.js | 1 + .../routes/agents/__tests__/events.spec.js | 82 +++++ api/server/routes/agents/middleware.js | 9 +- api/server/routes/agents/openai.js | 34 +- api/server/services/Agents/triggers.js | 1 + packages/api/src/agents/triggers/README.md | 37 +++ packages/api/src/agents/triggers/index.ts | 1 + .../api/src/agents/triggers/ingress.spec.ts | 306 ++++++++++++++++++ packages/api/src/agents/triggers/ingress.ts | 280 ++++++++++++++++ .../agents/triggers/service.delivery.spec.ts | 1 + packages/api/src/agents/triggers/service.ts | 17 + packages/api/src/apiKeys/middleware.spec.ts | 76 +++++ packages/api/src/apiKeys/middleware.ts | 39 ++- .../src/middleware/messageFilterPii.spec.ts | 10 + packages/api/src/protection/adapters/chat.ts | 5 + packages/api/src/protection/legacy.ts | 1 + .../src/methods/triggerDelivery.spec.ts | 72 +++++ .../src/methods/triggerDelivery.ts | 29 ++ .../data-schemas/src/types/triggerDelivery.ts | 13 + 21 files changed, 1033 insertions(+), 16 deletions(-) create mode 100644 api/server/routes/agents/__tests__/events.spec.js create mode 100644 packages/api/src/agents/triggers/ingress.spec.ts create mode 100644 packages/api/src/agents/triggers/ingress.ts create mode 100644 packages/api/src/apiKeys/middleware.spec.ts diff --git a/.env.example b/.env.example index 22b4265882..3b525166f9 100644 --- a/.env.example +++ b/.env.example @@ -724,6 +724,11 @@ 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 + ILLEGAL_MODEL_REQ_SCORE=5 #========================# diff --git a/api/server/middleware/limiters/messageLimiters.js b/api/server/middleware/limiters/messageLimiters.js index 4f1d72076f..2e05820032 100644 --- a/api/server/middleware/limiters/messageLimiters.js +++ b/api/server/middleware/limiters/messageLimiters.js @@ -9,6 +9,8 @@ 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; @@ -20,6 +22,10 @@ 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. @@ -74,7 +80,31 @@ const messageIpLimiter = rateLimit(ipLimiterOptions); */ const messageUserLimiter = rateLimit(userLimiterOptions); +/** + * Event admission has its own API-principal bucket. The durable worker later + * 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'), +}); + module.exports = { + agentEventUserLimiter, messageIpLimiter, messageUserLimiter, }; diff --git a/api/server/routes/agents/__tests__/abort.spec.js b/api/server/routes/agents/__tests__/abort.spec.js index baec3d32cb..b99d0fb59a 100644 --- a/api/server/routes/agents/__tests__/abort.spec.js +++ b/api/server/routes/agents/__tests__/abort.spec.js @@ -64,6 +64,7 @@ jest.mock('~/server/middleware', () => ({ next(); }, moderateText: (req, res, next) => next(), + agentEventUserLimiter: (req, res, next) => next(), messageIpLimiter: (req, res, next) => next(), configMiddleware: (req, res, next) => next(), messageUserLimiter: (req, res, next) => next(), diff --git a/api/server/routes/agents/__tests__/events.spec.js b/api/server/routes/agents/__tests__/events.spec.js new file mode 100644 index 0000000000..126703f757 --- /dev/null +++ b/api/server/routes/agents/__tests__/events.spec.js @@ -0,0 +1,82 @@ +const express = require('express'); +const request = require('supertest'); + +const mockEnqueueAgentTrigger = jest.fn(); +const mockGetAgentTriggerDeliveryStatus = jest.fn(); +const mockEnqueueEvent = jest.fn((_req, res) => res.status(202).json({ id: 'trigger-1' })); +const mockGetEvent = jest.fn((_req, res) => res.status(200).json({ status: 'succeeded' })); +let mockIngressDependencies; +const mockCreateAgentTriggerIngressHandlers = jest.fn((dependencies) => { + mockIngressDependencies = dependencies; + return { + enqueueEvent: mockEnqueueEvent, + getEvent: mockGetEvent, + }; +}); + +jest.mock('@librechat/api', () => ({ + createAgentTriggerIngressHandlers: mockCreateAgentTriggerIngressHandlers, + createMessageFilterPii: () => (_req, _res, next) => next(), +})); + +jest.mock('~/server/controllers/agents/openai', () => ({ + OpenAIChatCompletionController: jest.fn(), + ListModelsController: jest.fn(), + GetModelController: jest.fn(), +})); + +jest.mock('~/server/services/Agents/triggers', () => ({ + enqueueAgentTrigger: mockEnqueueAgentTrigger, + getAgentTriggerDeliveryStatus: mockGetAgentTriggerDeliveryStatus, +})); + +jest.mock('~/server/middleware', () => ({ + agentEventUserLimiter: (_req, _res, next) => next(), + configMiddleware: (_req, _res, next) => next(), +})); + +jest.mock('../middleware', () => ({ + preAuthTenantMiddleware: (_req, _res, next) => next(), + requireRemoteAgentAuth: (req, _res, next) => { + req.user = { id: 'user-1' }; + next(); + }, + checkRemoteAgentsFeature: (_req, _res, next) => next(), + checkAgentPermission: (_req, _res, next) => next(), + checkAgentTriggerPermission: (_req, _res, next) => next(), +})); + +const router = require('../openai'); + +describe('Remote Agents event routes', () => { + const app = express(); + app.use(express.json()); + app.use('/api/agents/v1', router); + + beforeEach(() => { + mockEnqueueEvent.mockClear(); + mockGetEvent.mockClear(); + }); + + it('wires durable event admission to the trigger service', async () => { + const response = await request(app) + .post('/api/agents/v1/events') + .send({ + target: { agentId: 'agent-1' }, + }); + + expect(response.status).toBe(202); + expect(mockIngressDependencies).toEqual({ + enqueue: mockEnqueueAgentTrigger, + getDeliveryStatus: mockGetAgentTriggerDeliveryStatus, + }); + expect(mockEnqueueEvent).toHaveBeenCalledTimes(1); + }); + + it('wires owner-scoped event status reads to the ingress handler', async () => { + const response = await request(app).get('/api/agents/v1/events/trigger-1'); + + expect(response.status).toBe(200); + expect(mockGetEvent).toHaveBeenCalledTimes(1); + }); +}); diff --git a/api/server/routes/agents/middleware.js b/api/server/routes/agents/middleware.js index efb4a8c0b0..49cd18425b 100644 --- a/api/server/routes/agents/middleware.js +++ b/api/server/routes/agents/middleware.js @@ -4,6 +4,7 @@ const { preAuthTenantMiddleware, createRequireApiKeyAuth, createRemoteAgentAuth, + createCheckAgentTriggerAccess, createCheckRemoteAgentAccess, } = require('@librechat/api'); const { getEffectivePermissions } = require('~/server/services/PermissionService'); @@ -29,13 +30,17 @@ const checkRemoteAgentsFeature = generateCheckAccess({ getRoleByName: db.getRoleByName, }); -const checkAgentPermission = createCheckRemoteAgentAccess({ +const agentAccessDependencies = { getAgent: db.getAgent, getEffectivePermissions, -}); +}; + +const checkAgentPermission = createCheckRemoteAgentAccess(agentAccessDependencies); +const checkAgentTriggerPermission = createCheckAgentTriggerAccess(agentAccessDependencies); module.exports = { checkAgentPermission, + checkAgentTriggerPermission, preAuthTenantMiddleware, requireRemoteAgentAuth, checkRemoteAgentsFeature, diff --git a/api/server/routes/agents/openai.js b/api/server/routes/agents/openai.js index fa7f9b26c8..44008bb35f 100644 --- a/api/server/routes/agents/openai.js +++ b/api/server/routes/agents/openai.js @@ -6,6 +6,8 @@ * * Usage: * POST /v1/chat/completions - Chat with an agent + * POST /v1/events - Durably deliver a source-neutral event + * GET /v1/events/:id - Read an event delivery status and result * GET /v1/models - List available agents * GET /v1/models/:model - Get agent details * @@ -17,26 +19,56 @@ * } */ const express = require('express'); +const { createAgentTriggerIngressHandlers, createMessageFilterPii } = require('@librechat/api'); const { OpenAIChatCompletionController, ListModelsController, GetModelController, } = require('~/server/controllers/agents/openai'); -const { configMiddleware } = require('~/server/middleware'); +const { agentEventUserLimiter, configMiddleware } = require('~/server/middleware'); +const { + enqueueAgentTrigger, + getAgentTriggerDeliveryStatus, +} = require('~/server/services/Agents/triggers'); const { checkAgentPermission, + checkAgentTriggerPermission, preAuthTenantMiddleware, requireRemoteAgentAuth, checkRemoteAgentsFeature, } = require('./middleware'); const router = express.Router(); +const eventHandlers = createAgentTriggerIngressHandlers({ + enqueue: enqueueAgentTrigger, + getDeliveryStatus: getAgentTriggerDeliveryStatus, +}); router.use(preAuthTenantMiddleware); router.use(requireRemoteAgentAuth); router.use(configMiddleware); router.use(checkRemoteAgentsFeature); +/** + * @route POST /v1/events + * @desc Durably deliver a source-neutral event to an agent + * @access Private (API key auth required) + */ +router.post( + '/events', + agentEventUserLimiter, + createMessageFilterPii({ getConfig: (req) => req.config?.messageFilter?.pii }), + checkAgentTriggerPermission, + eventHandlers.enqueueEvent, +); + +/** + * @route GET /v1/events/:id + * @desc Read the authenticated owner's delivery status and result + * @access Private (API key auth required) + */ +router.get('/events/:id', eventHandlers.getEvent); + /** * @route POST /v1/chat/completions * @desc OpenAI-compatible chat completions with agents diff --git a/api/server/services/Agents/triggers.js b/api/server/services/Agents/triggers.js index 2279a0da2a..e662e1bb31 100644 --- a/api/server/services/Agents/triggers.js +++ b/api/server/services/Agents/triggers.js @@ -20,6 +20,7 @@ module.exports = { dispatchAgentTrigger: service.dispatch, enqueueAgentTrigger: service.enqueue, getAgentTriggerDelivery: service.getDelivery, + getAgentTriggerDeliveryStatus: service.getDeliveryStatus, getAgentTriggerDeadLetters: service.getDeadLetters, requeueAgentTrigger: service.requeue, drainAgentTriggerDeliveriesForUser: service.drainUser, diff --git a/packages/api/src/agents/triggers/README.md b/packages/api/src/agents/triggers/README.md index f4da03ac9a..610188837d 100644 --- a/packages/api/src/agents/triggers/README.md +++ b/packages/api/src/agents/triggers/README.md @@ -68,3 +68,40 @@ await enqueueAgentTrigger( `getAgentTriggerDeadLetters` and `requeueAgentTrigger` are intentionally trusted in-process operations. Exposing them through an admin API requires a separate authorization and audit layer. + +## Remote event ingress + +Authenticated controllers and source adapters can enqueue the same durable envelope through +`POST /api/agents/v1/events`. The endpoint uses Remote Agents API-key authentication, the remote +agents feature permission, and the target agent's existing remote-view ACL. Send exactly one +`Idempotency-Key` header and keep it stable when retrying the same source-event-to-target delivery. +The authenticated user, tenant, API-key source identity, request id, and receive time are always +supplied by LibreChat. Remote callers do not choose `event.source`; provider-specific webhook +adapters may verify their native signature and map verified provider metadata into the trusted +in-process adapter contract above. + +```http +POST /api/agents/v1/events +Authorization: Bearer +Idempotency-Key: webhook-42-resource-7 +Content-Type: application/json + +{ + "mode": "fire", + "event": { + "id": "resource-7-ready-3", + "type": "resource.ready", + "occurredAt": 1786967999000, + "payload": { "resourceId": "resource-7" } + }, + "target": { "agentId": "agent-id" }, + "input": "Resource resource-7 is ready. Inspect it and report the result.", + "orderingKey": "resource-7" +} +``` + +A successful admission returns `202 Accepted`, an opaque delivery `id`, and a `Location` header. +Poll that location to read `pending`, `leased`, `succeeded`, or `dead` state. Successful fire +results include the conversation and generation identity needed for a later `steer` event. Status +responses never expose the stored source payload, ordering key, retry history, or worker identity. +Callers must sanitize `event.payload`; credentials and transport secrets must not be persisted. diff --git a/packages/api/src/agents/triggers/index.ts b/packages/api/src/agents/triggers/index.ts index ab2c692942..e3e2f8c6a3 100644 --- a/packages/api/src/agents/triggers/index.ts +++ b/packages/api/src/agents/triggers/index.ts @@ -2,5 +2,6 @@ export * from './dispatch'; export * from './delivery'; export * from './envelope'; export * from './host'; +export * from './ingress'; export * from './service'; export * from './engine'; diff --git a/packages/api/src/agents/triggers/ingress.spec.ts b/packages/api/src/agents/triggers/ingress.spec.ts new file mode 100644 index 0000000000..754b646a41 --- /dev/null +++ b/packages/api/src/agents/triggers/ingress.spec.ts @@ -0,0 +1,306 @@ +import express from 'express'; +import request from 'supertest'; +import type { Application } from 'express'; +import type { AgentTriggerIngressDependencies } from './ingress'; +import type { AgentTriggerStoredRecord } from './service'; +import { AgentTriggerServiceUnavailableError } from './service'; +import { createAgentTriggerIngressHandlers } from './ingress'; + +const USER_ID = '507f1f77bcf86cd799439011'; +const API_KEY_ID = '68a1c312abc123abc123abcf'; +const DELIVERY_KEY = `trigger_${'a'.repeat(64)}`; +const AVAILABLE_AT = new Date('2026-08-17T12:00:00.000Z'); +const CREATED_AT = new Date('2026-08-17T11:59:59.000Z'); + +function delivery(overrides: Partial = {}): AgentTriggerStoredRecord { + return { + id: '68a1c312abc123abc123abc1', + user: USER_ID, + deliveryKey: DELIVERY_KEY, + fingerprint: 'fingerprint', + orderingKey: 'ordering-key', + laneSequence: 1, + envelope: { secret: 'not-public' }, + status: 'pending', + attempts: 0, + availableAt: AVAILABLE_AT, + createdAt: CREATED_AT, + history: [ + { + attempt: 1, + outcome: 'retry', + at: CREATED_AT, + workerId: 'private-worker-id', + }, + ], + ...overrides, + }; +} + +function dependencies( + overrides: Partial = {}, +): AgentTriggerIngressDependencies { + return { + enqueue: jest.fn(async () => ({ + id: '68a1c312abc123abc123abc1', + deliveryKey: DELIVERY_KEY, + status: 'pending' as const, + availableAt: AVAILABLE_AT, + replayed: false, + })), + getDeliveryStatus: jest.fn(async () => delivery()), + now: () => 1_755_430_000_000, + createRequestId: () => 'generated-request-id', + ...overrides, + }; +} + +function createApp( + deps: AgentTriggerIngressDependencies, + user: { id: string; role?: string; tenantId?: string } | null = { + id: USER_ID, + role: 'USER', + tenantId: 'tenant-1', + }, +): Application { + const app = express(); + const handlers = createAgentTriggerIngressHandlers(deps); + app.use(express.json()); + app.use((req, _res, next) => { + Object.assign(req, { + user: user ?? undefined, + apiKeyId: API_KEY_ID, + requestId: 'request-from-context', + }); + next(); + }); + app.post('/api/agents/v1/events', handlers.enqueueEvent); + app.get('/api/agents/v1/events/:id', handlers.getEvent); + return app; +} + +function fireEvent() { + return { + mode: 'fire', + event: { + id: 'event-1', + type: 'resource.ready', + occurredAt: 1_755_429_900_000, + source: { id: 'webhook-1', type: 'webhook' }, + payload: { resourceId: 'resource-1' }, + }, + target: { agentId: 'agent-1' }, + input: 'Handle resource-1.', + orderingKey: 'resource-1', + principal: { id: 'attacker-controlled' }, + }; +} + +describe('agent trigger event ingress', () => { + it('builds a trusted envelope and returns an opaque delivery status URL', async () => { + const deps = dependencies(); + const response = await request(createApp(deps)) + .post('/api/agents/v1/events') + .set('Idempotency-Key', 'source-delivery-1') + .send(fireEvent()); + + expect(response.status).toBe(202); + expect(response.headers.location).toBe(`/api/agents/v1/events/${DELIVERY_KEY}`); + expect(response.body).toEqual({ + id: DELIVERY_KEY, + status: 'pending', + availableAt: AVAILABLE_AT.toISOString(), + replayed: false, + }); + expect(deps.enqueue).toHaveBeenCalledWith( + { + version: 1, + mode: 'fire', + requestId: 'request-from-context', + deliveryId: 'source-delivery-1', + receivedAt: 1_755_430_000_000, + principal: { userId: USER_ID, role: 'USER', tenantId: 'tenant-1' }, + event: { + ...fireEvent().event, + source: { id: API_KEY_ID, type: 'remote_api_key' }, + }, + target: { agentId: 'agent-1' }, + input: 'Handle resource-1.', + }, + { orderingKey: 'resource-1' }, + ); + }); + + it('admits fenced steer events through the same source-neutral contract', async () => { + const deps = dependencies(); + const response = await request(createApp(deps)) + .post('/api/agents/v1/events') + .set('Idempotency-Key', 'steer-delivery-1') + .send({ + mode: 'steer', + event: { + id: 'clock-expired-1', + type: 'clock.expired', + occurredAt: 1_755_429_950_000, + source: { id: 'match-1', type: 'tournament-controller' }, + }, + target: { + agentId: 'agent-1', + conversationId: 'conversation-1', + generationCreatedAt: 1_755_429_940_000, + preempt: true, + }, + input: 'Your clock expired. Submit immediately.', + }); + + expect(response.status).toBe(202); + expect(deps.enqueue).toHaveBeenCalledWith( + expect.objectContaining({ + mode: 'steer', + deliveryId: 'steer-delivery-1', + target: { + agentId: 'agent-1', + conversationId: 'conversation-1', + generationCreatedAt: 1_755_429_940_000, + preempt: true, + }, + }), + {}, + ); + }); + + it('fails closed when the idempotency header is absent or duplicated', async () => { + const deps = dependencies(); + const app = createApp(deps); + + const missing = await request(app).post('/api/agents/v1/events').send(fireEvent()); + const duplicated = await request(app) + .post('/api/agents/v1/events') + .set('Idempotency-Key', 'delivery-1,delivery-2') + .send(fireEvent()); + + expect(missing.status).toBe(400); + expect(duplicated.status).toBe(400); + expect(deps.enqueue).not.toHaveBeenCalled(); + }); + + it('rejects malformed event bodies before enqueueing', async () => { + const deps = dependencies(); + const response = await request(createApp(deps)) + .post('/api/agents/v1/events') + .set('Idempotency-Key', 'delivery-1') + .send({ ...fireEvent(), mode: 'unknown' }); + + expect(response.status).toBe(400); + expect(response.body.error.code).toBe('invalid_event'); + expect(deps.enqueue).not.toHaveBeenCalled(); + }); + + it('reports idempotency conflicts and temporary delivery unavailability', async () => { + const conflict = dependencies({ + enqueue: jest.fn(async () => { + const error = new Error(`Delivery ${DELIVERY_KEY} was reused with different content`); + error.name = 'AgentTriggerDeliveryConflictError'; + throw error; + }), + }); + const unavailable = dependencies({ + enqueue: jest.fn(async () => { + throw new AgentTriggerServiceUnavailableError('Delivery service is starting'); + }), + }); + + const conflictResponse = await request(createApp(conflict)) + .post('/api/agents/v1/events') + .set('Idempotency-Key', 'delivery-1') + .send(fireEvent()); + const unavailableResponse = await request(createApp(unavailable)) + .post('/api/agents/v1/events') + .set('Idempotency-Key', 'delivery-1') + .send(fireEvent()); + + expect(conflictResponse.status).toBe(409); + expect(conflictResponse.body.error.code).toBe('idempotency_conflict'); + expect(unavailableResponse.status).toBe(503); + expect(unavailableResponse.body.error.code).toBe('trigger_delivery_unavailable'); + }); + + it('returns a safe owner-scoped delivery projection', async () => { + const settledAt = new Date('2026-08-17T12:00:01.000Z'); + const attemptedAt = new Date('2026-08-17T12:00:00.500Z'); + const deps = dependencies({ + getDeliveryStatus: jest.fn(async () => + delivery({ + status: 'dead', + attempts: 3, + settledAt, + lastError: { + code: 'FORBIDDEN', + message: 'Agent access was revoked', + certainty: 'definite', + retryable: false, + attemptedAt, + status: 403, + }, + }), + ), + }); + + const response = await request(createApp(deps)).get(`/api/agents/v1/events/${DELIVERY_KEY}`); + + expect(response.status).toBe(200); + expect(response.body).toEqual({ + id: DELIVERY_KEY, + status: 'dead', + attempts: 3, + availableAt: AVAILABLE_AT.toISOString(), + createdAt: CREATED_AT.toISOString(), + settledAt: settledAt.toISOString(), + error: { + code: 'FORBIDDEN', + message: 'Agent access was revoked', + certainty: 'definite', + retryable: false, + attemptedAt: attemptedAt.toISOString(), + status: 403, + }, + }); + expect(response.body).not.toHaveProperty('envelope'); + expect(response.body).not.toHaveProperty('history'); + expect(response.body).not.toHaveProperty('orderingKey'); + }); + + it('uses an API-key, owner, and tenant-scoped status projection and hides missing deliveries', async () => { + const deps = dependencies({ getDeliveryStatus: jest.fn(async () => null) }); + const response = await request(createApp(deps)).get(`/api/agents/v1/events/${DELIVERY_KEY}`); + + expect(response.status).toBe(404); + expect(response.body.error.code).toBe('event_not_found'); + expect(deps.getDeliveryStatus).toHaveBeenCalledWith( + DELIVERY_KEY, + USER_ID, + API_KEY_ID, + 'tenant-1', + ); + }); + + it('rejects malformed delivery ids without querying storage', async () => { + const deps = dependencies(); + const response = await request(createApp(deps)).get('/api/agents/v1/events/not-a-key'); + + expect(response.status).toBe(400); + expect(response.body.error.code).toBe('invalid_event'); + expect(deps.getDeliveryStatus).not.toHaveBeenCalled(); + }); + + it('requires an authenticated principal even when mounted incorrectly', async () => { + const deps = dependencies(); + const response = await request(createApp(deps, null)) + .post('/api/agents/v1/events') + .set('Idempotency-Key', 'delivery-1') + .send(fireEvent()); + + expect(response.status).toBe(401); + expect(deps.enqueue).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/api/src/agents/triggers/ingress.ts b/packages/api/src/agents/triggers/ingress.ts new file mode 100644 index 0000000000..1c591f4aa1 --- /dev/null +++ b/packages/api/src/agents/triggers/ingress.ts @@ -0,0 +1,280 @@ +import { randomUUID } from 'node:crypto'; +import { logger } from '@librechat/data-schemas'; +import type { Request, RequestHandler, Response } from 'express'; +import type { + AgentFireTarget, + AgentSteerTarget, + AgentTriggerEvent, + AgentTriggerMode, +} from './envelope'; +import type { AgentTriggerEnqueueOptions } from './delivery'; +import type { AgentTriggerService } from './service'; +import { AgentTriggerEnvelopeError, createAgentTriggerEnvelope } from './envelope'; +import { AgentTriggerServiceUnavailableError } from './service'; +import { AgentTriggerDeliveryError } from './delivery'; + +const IDEMPOTENCY_HEADER = 'idempotency-key'; +const MAX_IDEMPOTENCY_KEY_LENGTH = 256; +const IDEMPOTENCY_KEY_PATTERN = /^[A-Za-z0-9._~:/+=-]+$/; +const DELIVERY_KEY_PATTERN = /^trigger_[a-f0-9]{64}$/; + +interface AgentTriggerIngressUser { + id: string; + role?: string; + tenantId?: string; +} + +interface AgentTriggerIngressRequest extends Request { + apiKeyId?: { toString(): string } | string; + requestId?: string; + user?: AgentTriggerIngressUser; +} + +interface AgentTriggerIngressBody { + mode?: AgentTriggerMode; + event?: AgentTriggerEvent; + target?: AgentFireTarget | AgentSteerTarget; + input?: string; + orderingKey?: string; +} + +export interface AgentTriggerIngressDependencies { + enqueue: AgentTriggerService['enqueue']; + getDeliveryStatus: AgentTriggerService['getDeliveryStatus']; + now?: () => number; + createRequestId?: () => string; +} + +class AgentTriggerIngressError extends TypeError { + constructor(message: string) { + super(message); + this.name = 'AgentTriggerIngressError'; + } +} + +class AgentTriggerAuthenticationError extends Error { + constructor() { + super('Authenticated user is required'); + this.name = 'AgentTriggerAuthenticationError'; + } +} + +function sendError(res: Response, status: number, code: string, message: string): void { + res.status(status).json({ + error: { + message, + type: status >= 500 ? 'server_error' : 'invalid_request_error', + code, + }, + }); +} + +function rawHeaderValues(req: Request, name: string): string[] { + const values: string[] = []; + for (let index = 0; index < req.rawHeaders.length; index += 2) { + if (req.rawHeaders[index]?.toLowerCase() === name) { + values.push(req.rawHeaders[index + 1] ?? ''); + } + } + if (values.length > 0) { + return values; + } + + const fallback = req.headers[name]; + if (Array.isArray(fallback)) { + return fallback; + } + return fallback == null ? [] : [fallback]; +} + +function requireIdempotencyKey(req: Request): string { + const values = rawHeaderValues(req, IDEMPOTENCY_HEADER); + if (values.length !== 1) { + throw new AgentTriggerIngressError('Exactly one Idempotency-Key header is required'); + } + const key = values[0].trim(); + if ( + key.length === 0 || + key.length > MAX_IDEMPOTENCY_KEY_LENGTH || + !IDEMPOTENCY_KEY_PATTERN.test(key) + ) { + throw new AgentTriggerIngressError( + `Idempotency-Key must contain 1-${MAX_IDEMPOTENCY_KEY_LENGTH} visible token characters`, + ); + } + return key; +} + +function requireUser(req: AgentTriggerIngressRequest): AgentTriggerIngressUser { + if (typeof req.user?.id !== 'string' || req.user.id.trim() === '') { + throw new AgentTriggerAuthenticationError(); + } + return req.user; +} + +function requireSourceKeyId(req: AgentTriggerIngressRequest): string { + const sourceKeyId = req.apiKeyId?.toString().trim(); + if (sourceKeyId == null || sourceKeyId === '') { + throw new AgentTriggerAuthenticationError(); + } + return sourceKeyId; +} + +function requireBody(value: object | null | undefined): AgentTriggerIngressBody { + if (value == null || typeof value !== 'object' || Array.isArray(value)) { + throw new AgentTriggerIngressError('Event body must be an object'); + } + return value as AgentTriggerIngressBody; +} + +function enqueueOptions(body: AgentTriggerIngressBody): AgentTriggerEnqueueOptions { + if (body.orderingKey != null && typeof body.orderingKey !== 'string') { + throw new AgentTriggerIngressError('orderingKey must be a string'); + } + return body.orderingKey == null ? {} : { orderingKey: body.orderingKey }; +} + +function toPublicDelivery(delivery: Awaited>) { + if (delivery == null) { + return null; + } + return { + id: delivery.deliveryKey, + status: delivery.status, + attempts: delivery.attempts, + availableAt: delivery.availableAt.toISOString(), + createdAt: delivery.createdAt.toISOString(), + ...(delivery.settledAt != null && { settledAt: delivery.settledAt.toISOString() }), + ...(delivery.result !== undefined && { result: delivery.result }), + ...(delivery.lastError != null && { + error: { + code: delivery.lastError.code, + message: delivery.lastError.message, + certainty: delivery.lastError.certainty, + retryable: delivery.lastError.retryable, + attemptedAt: delivery.lastError.attemptedAt.toISOString(), + ...(delivery.lastError.status != null && { status: delivery.lastError.status }), + }, + }), + }; +} + +function handleIngressError(res: Response, error: unknown): void { + if (error instanceof AgentTriggerAuthenticationError) { + sendError(res, 401, 'invalid_api_key', error.message); + return; + } + if ( + error instanceof AgentTriggerIngressError || + error instanceof AgentTriggerEnvelopeError || + error instanceof AgentTriggerDeliveryError + ) { + sendError(res, 400, 'invalid_event', error.message); + return; + } + if (error instanceof Error && error.name === 'AgentTriggerDeliveryConflictError') { + sendError(res, 409, 'idempotency_conflict', error.message); + return; + } + if (error instanceof AgentTriggerServiceUnavailableError) { + sendError(res, 503, 'trigger_delivery_unavailable', error.message); + return; + } + logger.error('[agent-trigger-ingress] request failed:', error); + sendError(res, 500, 'internal_error', 'Failed to process agent event'); +} + +export function createAgentTriggerIngressHandlers(deps: AgentTriggerIngressDependencies): { + enqueueEvent: RequestHandler; + getEvent: RequestHandler; +} { + const now = deps.now ?? Date.now; + const createRequestId = deps.createRequestId ?? randomUUID; + + const enqueueEvent: RequestHandler = async (baseReq, res) => { + const req = baseReq as AgentTriggerIngressRequest; + try { + const user = requireUser(req); + const sourceKeyId = requireSourceKeyId(req); + const body = requireBody(req.body); + const requestId = req.requestId?.trim() || createRequestId(); + const deliveryId = requireIdempotencyKey(req); + const receivedAt = now(); + const principal = { + id: user.id, + ...(user.role != null && { role: user.role }), + ...(user.tenantId != null && { tenantId: user.tenantId }), + }; + const common = { + requestId, + deliveryId, + receivedAt, + principal, + event: { + ...(body.event as AgentTriggerEvent), + source: { id: sourceKeyId, type: 'remote_api_key' }, + }, + input: body.input as string, + }; + const envelope = + body.mode === 'fire' + ? createAgentTriggerEnvelope({ + ...common, + mode: body.mode, + target: body.target as AgentFireTarget, + }) + : createAgentTriggerEnvelope({ + ...common, + mode: body.mode as 'steer', + target: body.target as AgentSteerTarget, + }); + const receipt = await deps.enqueue(envelope, enqueueOptions(body)); + + logger.info('[agent-trigger-ingress] delivery accepted', { + delivery_key: receipt.deliveryKey, + mode: envelope.mode, + agent_id: envelope.target.agentId, + user_id: user.id, + tenant_id: user.tenantId, + replayed: receipt.replayed, + }); + const collectionPath = req.originalUrl.split('?')[0].replace(/\/+$/, ''); + res.setHeader('Location', `${collectionPath}/${encodeURIComponent(receipt.deliveryKey)}`); + res.status(202).json({ + id: receipt.deliveryKey, + status: receipt.status, + availableAt: receipt.availableAt.toISOString(), + replayed: receipt.replayed, + }); + } catch (error) { + handleIngressError(res, error); + } + }; + + const getEvent: RequestHandler = async (baseReq, res) => { + const req = baseReq as AgentTriggerIngressRequest; + try { + const user = requireUser(req); + const sourceKeyId = requireSourceKeyId(req); + const deliveryKey = req.params.id; + if (!DELIVERY_KEY_PATTERN.test(deliveryKey)) { + throw new AgentTriggerIngressError('Event delivery id is invalid'); + } + const delivery = await deps.getDeliveryStatus( + deliveryKey, + user.id, + sourceKeyId, + user.tenantId, + ); + if (delivery == null) { + sendError(res, 404, 'event_not_found', 'Agent event delivery not found'); + return; + } + res.status(200).json(toPublicDelivery(delivery)); + } catch (error) { + handleIngressError(res, error); + } + }; + + return { enqueueEvent, getEvent }; +} diff --git a/packages/api/src/agents/triggers/service.delivery.spec.ts b/packages/api/src/agents/triggers/service.delivery.spec.ts index f9e0f7a329..98163adfa6 100644 --- a/packages/api/src/agents/triggers/service.delivery.spec.ts +++ b/packages/api/src/agents/triggers/service.delivery.spec.ts @@ -74,6 +74,7 @@ function deliveryMethods( retryAgentTriggerDelivery: jest.fn(async () => true), deadLetterAgentTriggerDelivery: jest.fn(async () => true), getAgentTriggerDelivery: jest.fn(async () => null), + getAgentTriggerDeliveryStatus: jest.fn(async () => null), getAgentTriggerDeadLetters: jest.fn(async () => []), requeueAgentTriggerDelivery: jest.fn(async () => null), countActiveAgentTriggerDeliveriesByUser: jest.fn(async () => 0), diff --git a/packages/api/src/agents/triggers/service.ts b/packages/api/src/agents/triggers/service.ts index dc2694bd07..0c14850f66 100644 --- a/packages/api/src/agents/triggers/service.ts +++ b/packages/api/src/agents/triggers/service.ts @@ -1,4 +1,5 @@ import { logger, runAsSystem } from '@librechat/data-schemas'; +import type { AgentTriggerDeliveryStatusRecord } from '@librechat/data-schemas'; import type { AgentTriggerDeliveryFailure, AgentTriggerDeliveryEngine, @@ -94,6 +95,12 @@ export interface AgentTriggerDeliveryPersistence { retryAgentTriggerDelivery: AgentTriggerDeliveryStore['retry']; deadLetterAgentTriggerDelivery: AgentTriggerDeliveryStore['dead']; getAgentTriggerDelivery: (deliveryKey: string) => Promise; + getAgentTriggerDeliveryStatus: ( + deliveryKey: string, + userId: string, + sourceKeyId: string, + tenantId?: string, + ) => Promise; getAgentTriggerDeadLetters: (limit?: number) => Promise; requeueAgentTriggerDelivery: ( id: string, @@ -124,6 +131,12 @@ export interface AgentTriggerService { options?: AgentTriggerEnqueueOptions, ) => Promise; getDelivery: (deliveryKey: string) => Promise; + getDeliveryStatus: ( + deliveryKey: string, + userId: string, + sourceKeyId: string, + tenantId?: string, + ) => Promise; getDeadLetters: (limit?: number) => Promise; requeue: (id: string, availableAt?: Date) => Promise; drainUser: (userId: string) => Promise; @@ -394,6 +407,10 @@ export function createAgentTriggerService(deps: AgentTriggerServiceDeps = {}): A }, getDelivery: (deliveryKey) => runAsSystem(async () => requireMethods().getAgentTriggerDelivery(deliveryKey)), + getDeliveryStatus: (deliveryKey, userId, sourceKeyId, tenantId) => + runAsSystem(async () => + requireMethods().getAgentTriggerDeliveryStatus(deliveryKey, userId, sourceKeyId, tenantId), + ), getDeadLetters: (limit) => runAsSystem(async () => requireMethods().getAgentTriggerDeadLetters(limit)), requeue: (id, availableAt = new Date()) => diff --git a/packages/api/src/apiKeys/middleware.spec.ts b/packages/api/src/apiKeys/middleware.spec.ts new file mode 100644 index 0000000000..8b282aacf2 --- /dev/null +++ b/packages/api/src/apiKeys/middleware.spec.ts @@ -0,0 +1,76 @@ +import express from 'express'; +import request from 'supertest'; +import { Types } from 'mongoose'; +import { PermissionBits } from 'librechat-data-provider'; +import { createCheckAgentTriggerAccess, createCheckRemoteAgentAccess } from './middleware'; + +describe('createCheckRemoteAgentAccess', () => { + it('preserves model-based authorization for existing remote agent routes', async () => { + const getAgent = jest.fn(async () => ({ _id: new Types.ObjectId() })); + const checkAccess = createCheckRemoteAgentAccess({ + getAgent, + getEffectivePermissions: jest.fn(async () => PermissionBits.VIEW), + }); + const app = express(); + app.use(express.json()); + app.use((req, _res, next) => { + Object.assign(req, { user: { id: new Types.ObjectId().toString(), role: 'USER' } }); + next(); + }); + app.post('/chat', checkAccess, (_req, res) => { + res.status(204).send(); + }); + + const response = await request(app).post('/chat').send({ model: 'agent-1' }); + + expect(response.status).toBe(204); + expect(getAgent).toHaveBeenCalledWith({ id: 'agent-1' }); + }); +}); + +describe('createCheckAgentTriggerAccess', () => { + it('authorizes the actual event target instead of a top-level model field', async () => { + const getAgent = jest.fn(async () => ({ _id: new Types.ObjectId() })); + const getEffectivePermissions = jest.fn(async () => PermissionBits.VIEW); + const checkAccess = createCheckAgentTriggerAccess({ getAgent, getEffectivePermissions }); + const app = express(); + app.use(express.json()); + app.use((req, _res, next) => { + Object.assign(req, { user: { id: new Types.ObjectId().toString(), role: 'USER' } }); + next(); + }); + app.post('/events', checkAccess, (_req, res) => { + res.status(204).send(); + }); + + const response = await request(app) + .post('/events') + .send({ + model: 'decoy-agent', + target: { agentId: 'target-agent' }, + }); + + expect(response.status).toBe(204); + expect(getAgent).toHaveBeenCalledWith({ id: 'target-agent' }); + expect(getAgent).not.toHaveBeenCalledWith({ id: 'decoy-agent' }); + }); + + it('does not fall back to model when the event target is absent', async () => { + const getAgent = jest.fn(async () => ({ _id: new Types.ObjectId() })); + const checkAccess = createCheckAgentTriggerAccess({ + getAgent, + getEffectivePermissions: jest.fn(async () => PermissionBits.VIEW), + }); + const app = express(); + app.use(express.json()); + app.post('/events', checkAccess, (_req, res) => { + res.status(204).send(); + }); + + const response = await request(app).post('/events').send({ model: 'decoy-agent' }); + + expect(response.status).toBe(400); + expect(response.body.error.code).toBe('missing_model'); + expect(getAgent).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/api/src/apiKeys/middleware.ts b/packages/api/src/apiKeys/middleware.ts index 626a8dd5df..82031e9038 100644 --- a/packages/api/src/apiKeys/middleware.ts +++ b/packages/api/src/apiKeys/middleware.ts @@ -1,6 +1,6 @@ import { logger } from '@librechat/data-schemas'; import { ResourceType, PermissionBits, hasPermissions } from 'librechat-data-provider'; -import type { Request, Response, NextFunction } from 'express'; +import type { Request, Response, NextFunction, RequestHandler } from 'express'; import type { IUser } from '@librechat/data-schemas'; import type { Types } from 'mongoose'; import { getRemoteAgentPermissions } from './service'; @@ -35,6 +35,8 @@ export interface RemoteAgentAccessRequest extends ApiKeyAuthRequest { agentPermissions?: number; } +type AgentIdResolver = (req: RemoteAgentAccessRequest) => string | undefined; + export function createRequireApiKeyAuth(deps: ApiKeyAuthDependencies) { return async ( req: ApiKeyAuthRequest, @@ -108,35 +110,37 @@ export function createRequireApiKeyAuth(deps: ApiKeyAuthDependencies) { }; } -export function createCheckRemoteAgentAccess(deps: RemoteAgentAccessDependencies) { - return async ( - req: RemoteAgentAccessRequest, - res: Response, - next: NextFunction, - ): Promise => { - const agentId = req.body?.model || req.params?.model; +function createAgentAccessMiddleware( + deps: RemoteAgentAccessDependencies, + resolveAgentId: AgentIdResolver, +): RequestHandler { + return async (baseReq, res, next): Promise => { + const req = baseReq as RemoteAgentAccessRequest; + const agentId = resolveAgentId(req); - if (!agentId) { - return res.status(400).json({ + if (typeof agentId !== 'string' || agentId.trim() === '') { + res.status(400).json({ error: { message: 'Model (agent ID) is required', type: 'invalid_request_error', code: 'missing_model', }, }); + return; } try { const agent = await deps.getAgent({ id: agentId }); if (!agent) { - return res.status(404).json({ + res.status(404).json({ error: { message: `Agent not found: ${agentId}`, type: 'invalid_request_error', code: 'model_not_found', }, }); + return; } const userId = req.user?.id || ''; @@ -144,13 +148,14 @@ export function createCheckRemoteAgentAccess(deps: RemoteAgentAccessDependencies const permissions = await getRemoteAgentPermissions(deps, userId, req.user?.role, agent._id); if (!hasPermissions(permissions, PermissionBits.VIEW)) { - return res.status(403).json({ + res.status(403).json({ error: { message: `No remote access to agent: ${agentId}`, type: 'permission_error', code: 'access_denied', }, }); + return; } req.agent = agent; @@ -159,7 +164,7 @@ export function createCheckRemoteAgentAccess(deps: RemoteAgentAccessDependencies next(); } catch (error) { logger.error('[checkRemoteAgentAccess] Error checking agent access:', error); - return res.status(500).json({ + res.status(500).json({ error: { message: 'Internal server error while checking agent access', type: 'server_error', @@ -169,3 +174,11 @@ export function createCheckRemoteAgentAccess(deps: RemoteAgentAccessDependencies } }; } + +export function createCheckRemoteAgentAccess(deps: RemoteAgentAccessDependencies): RequestHandler { + return createAgentAccessMiddleware(deps, (req) => req.body?.model || req.params?.model); +} + +export function createCheckAgentTriggerAccess(deps: RemoteAgentAccessDependencies): RequestHandler { + return createAgentAccessMiddleware(deps, (req) => req.body?.target?.agentId); +} diff --git a/packages/api/src/middleware/messageFilterPii.spec.ts b/packages/api/src/middleware/messageFilterPii.spec.ts index 7f0f71ed85..11a6c9f2b6 100644 --- a/packages/api/src/middleware/messageFilterPii.spec.ts +++ b/packages/api/src/middleware/messageFilterPii.spec.ts @@ -250,6 +250,16 @@ describe('messageFilterPii middleware', () => { }); }); + it('rejects model-bound event input before durable ingress can persist it', () => { + const { capturedRes, nextCalls } = runMiddleware( + {}, + { input: 'dispatch this with sk-proj-FAKE1234567890ABCDEF' }, + ); + expect(nextCalls).toBe(0); + expect(capturedRes.status).toBe(400); + expect(capturedRes.body).toMatchObject({ error: 'message_filter_pii_block' }); + }); + it('rejects with 400 when a Bearer header is present', () => { const { capturedRes, nextCalls } = runMiddleware( {}, diff --git a/packages/api/src/protection/adapters/chat.ts b/packages/api/src/protection/adapters/chat.ts index 491abe4e76..84c09eac04 100644 --- a/packages/api/src/protection/adapters/chat.ts +++ b/packages/api/src/protection/adapters/chat.ts @@ -19,6 +19,7 @@ export interface ChatSubmissionDecision { } export interface ChatSubmissionBody extends ModelParameterContentInput { + readonly input?: string; readonly text?: string; readonly quotes?: unknown; readonly answer?: string; @@ -103,6 +104,10 @@ export function extractChatContent( fragments.push(createFragment('chat.text', '/text', text, 'message', 'text')); } + if (typeof body?.input === 'string' && body.input.length > 0) { + fragments.push(createFragment('chat.input', '/input', body.input, 'message', 'text')); + } + const quoteEntries = getReferencedQuoteEntries(body?.quotes); if (quoteEntries != null) { const quotes: string[] = []; diff --git a/packages/api/src/protection/legacy.ts b/packages/api/src/protection/legacy.ts index 000575e144..82bf14867d 100644 --- a/packages/api/src/protection/legacy.ts +++ b/packages/api/src/protection/legacy.ts @@ -42,6 +42,7 @@ export function isLegacyPiiFragment(fragment: TextContentFragment): boolean { } return ( fragment.id === 'chat.text' || + fragment.id === 'chat.input' || fragment.id === 'chat.answer' || /^chat\.quote\.\d+$/.test(fragment.id) || /^chat\.decision\.\d+\.(?:response|reason)$/.test(fragment.id) || diff --git a/packages/data-schemas/src/methods/triggerDelivery.spec.ts b/packages/data-schemas/src/methods/triggerDelivery.spec.ts index 6e1e2dc292..5667b40c3e 100644 --- a/packages/data-schemas/src/methods/triggerDelivery.spec.ts +++ b/packages/data-schemas/src/methods/triggerDelivery.spec.ts @@ -103,6 +103,78 @@ describe('agent trigger delivery methods', () => { expect(await Delivery.countDocuments()).toBe(1); }); + it('projects public status while enforcing API key, owner, and tenant in the query', async () => { + const user = new mongoose.Types.ObjectId(); + const queued = await methods.enqueueAgentTriggerDelivery( + enqueueInput({ + user, + tenantId: 'tenant-1', + envelope: { + event: { source: { id: 'source-key-1', type: 'remote_api_key' } }, + privatePayload: 'x'.repeat(1024), + }, + }), + ); + + const status = await methods.getAgentTriggerDeliveryStatus( + queued.delivery.deliveryKey, + user, + 'source-key-1', + 'tenant-1', + ); + + expect(status).toEqual({ + deliveryKey: queued.delivery.deliveryKey, + status: 'pending', + attempts: 0, + availableAt: START, + createdAt: expect.any(Date), + }); + expect(status).not.toHaveProperty('envelope'); + expect(status).not.toHaveProperty('orderingKey'); + expect(status).not.toHaveProperty('history'); + await expect( + methods.getAgentTriggerDeliveryStatus( + queued.delivery.deliveryKey, + new mongoose.Types.ObjectId(), + 'source-key-1', + 'tenant-1', + ), + ).resolves.toBeNull(); + + const internal = await methods.enqueueAgentTriggerDelivery( + enqueueInput({ + user, + tenantId: 'tenant-1', + envelope: { event: { source: { id: 'source-key-1', type: 'schedule' } } }, + }), + ); + await expect( + methods.getAgentTriggerDeliveryStatus( + internal.delivery.deliveryKey, + user, + 'source-key-1', + 'tenant-1', + ), + ).resolves.toBeNull(); + await expect( + methods.getAgentTriggerDeliveryStatus( + queued.delivery.deliveryKey, + user, + 'source-key-1', + 'tenant-2', + ), + ).resolves.toBeNull(); + await expect( + methods.getAgentTriggerDeliveryStatus( + queued.delivery.deliveryKey, + user, + 'source-key-2', + 'tenant-1', + ), + ).resolves.toBeNull(); + }); + it('grants one atomic winner across concurrent claims', async () => { await methods.enqueueAgentTriggerDelivery(enqueueInput()); const claims = await Promise.all( diff --git a/packages/data-schemas/src/methods/triggerDelivery.ts b/packages/data-schemas/src/methods/triggerDelivery.ts index 0e718caecc..ff380650ce 100644 --- a/packages/data-schemas/src/methods/triggerDelivery.ts +++ b/packages/data-schemas/src/methods/triggerDelivery.ts @@ -3,6 +3,7 @@ import type { AgentTriggerDeliveryClaim, AgentTriggerDeliveryFailure, AgentTriggerDeliveryRecord, + AgentTriggerDeliveryStatusRecord, AgentTriggerOrderingBlock, IAgentTriggerDelivery, IAgentTriggerDeliveryDocument, @@ -94,6 +95,12 @@ export interface AgentTriggerDeliveryMethods { }, ) => Promise; getAgentTriggerDelivery: (deliveryKey: string) => Promise; + getAgentTriggerDeliveryStatus: ( + deliveryKey: string, + user: string | Types.ObjectId, + sourceKeyId: string, + tenantId?: string, + ) => Promise; getAgentTriggerDeadLetters: (limit?: number) => Promise; requeueAgentTriggerDelivery: ( id: string, @@ -814,6 +821,27 @@ export function createAgentTriggerDeliveryMethods( return delivery == null ? null : toRecord(delivery); } + async function getAgentTriggerDeliveryStatus( + deliveryKey: string, + user: string | Types.ObjectId, + sourceKeyId: string, + tenantId?: string, + ): Promise { + const tenantScope = + tenantId == null ? { tenantId: null } : { $or: [{ tenantId: null }, { tenantId }] }; + const delivery = await Delivery() + .findOne({ + deliveryKey, + user, + 'envelope.event.source.id': sourceKeyId, + 'envelope.event.source.type': 'remote_api_key', + ...tenantScope, + }) + .select('-_id deliveryKey status attempts availableAt createdAt settledAt result lastError') + .lean(); + return delivery; + } + async function getAgentTriggerDeadLetters(limit = 50): Promise { if (!Number.isSafeInteger(limit) || limit <= 0) { throw new TypeError('Agent trigger dead-letter limit must be a positive integer'); @@ -982,6 +1010,7 @@ export function createAgentTriggerDeliveryMethods( retryAgentTriggerDelivery, deadLetterAgentTriggerDelivery, getAgentTriggerDelivery, + getAgentTriggerDeliveryStatus, getAgentTriggerDeadLetters, requeueAgentTriggerDelivery, countActiveAgentTriggerDeliveriesByUser, diff --git a/packages/data-schemas/src/types/triggerDelivery.ts b/packages/data-schemas/src/types/triggerDelivery.ts index e29fd2b7e6..512278bb07 100644 --- a/packages/data-schemas/src/types/triggerDelivery.ts +++ b/packages/data-schemas/src/types/triggerDelivery.ts @@ -60,6 +60,19 @@ export interface AgentTriggerDeliveryRecord createdAt: Date; } +/** Owner-scoped projection safe for public delivery-status reads. */ +export type AgentTriggerDeliveryStatusRecord = Pick< + AgentTriggerDeliveryRecord, + | 'deliveryKey' + | 'status' + | 'attempts' + | 'availableAt' + | 'createdAt' + | 'settledAt' + | 'result' + | 'lastError' +>; + export interface IAgentTriggerLaneSequence { _id: string; value: number;