diff --git a/api/server/controllers/agents/__tests__/steer.spec.js b/api/server/controllers/agents/__tests__/steer.spec.js index 1ff636e1d1..0da95294f4 100644 --- a/api/server/controllers/agents/__tests__/steer.spec.js +++ b/api/server/controllers/agents/__tests__/steer.spec.js @@ -41,7 +41,7 @@ jest.mock('~/models', () => ({ const { Permissions, PermissionTypes, PermissionBits } = require('librechat-data-provider'); const SteerController = require('~/server/controllers/agents/steer'); -const { SteerCancelController, SteerArmController } = SteerController; +const { SteerDeliveryController, SteerCancelController, SteerArmController } = SteerController; const GENERATION_PROTOCOL_HEADER = 'x-librechat-generation-protocol'; @@ -60,6 +60,7 @@ function buildApp(user = { id: 'user-1', tenantId: 'tenant-1' }) { next(); }); app.post('/chat/steer', SteerController); + app.post('/chat/steer/deliver', SteerDeliveryController); app.post('/chat/steer/cancel', SteerCancelController); app.post('/chat/steer/arm', SteerArmController); return app; @@ -100,6 +101,7 @@ describe('SteerController (wrapper)', () => { { conversationId: 'c1', text: 'hello', files: [{ file_id: 'f1' }] }, { generationProtocolVersion: 1, + signal: expect.any(AbortSignal), getFiles: expect.any(Function), updateFilesUsage: expect.any(Function), checkAgentAccess: expect.any(Function), @@ -121,6 +123,44 @@ describe('SteerController (wrapper)', () => { expect(res.headers[GENERATION_PROTOCOL_HEADER]).toBe('1'); }); + it('makes trigger delivery strict and fences it to the declared agent', async () => { + mockHandleSteerRequest.mockResolvedValue({ + status: 202, + body: { status: 'queued', generationProtocolVersion: 2 }, + }); + mockCheckAccess.mockResolvedValue(true); + mockHasCapability.mockResolvedValue(true); + + await request(buildApp({ id: 'user-1', role: 'USER' })) + .post('/chat/steer/deliver') + .set('X-LibreChat-Generation-Protocol', '2') + .send({ + agentId: 'agent-1', + conversationId: 'c1', + clientSteerId: 'delivery-1', + text: 'move now', + generationProtocolVersion: 2, + }); + + const options = mockHandleSteerRequest.mock.calls[0][2]; + expect(options).toEqual( + expect.objectContaining({ + generationProtocolVersion: 2, + requireIdempotentDelivery: true, + signal: expect.any(AbortSignal), + }), + ); + await expect( + options.checkAgentAccess({ agentId: 'agent-1', endpoint: 'agents' }), + ).resolves.toBe(true); + await expect( + options.checkAgentAccess({ agentId: 'agent-other', endpoint: 'agents' }), + ).resolves.toBe(false); + await expect( + options.checkAgentAccess({ agentId: 'agent-1', endpoint: 'openAI' }), + ).resolves.toBe(false); + }); + it('500s with STEER_FAILED when the handler throws', async () => { mockHandleSteerRequest.mockRejectedValue(new Error('store down')); diff --git a/api/server/controllers/agents/steer.js b/api/server/controllers/agents/steer.js index 4ee163c3a7..5711154160 100644 --- a/api/server/controllers/agents/steer.js +++ b/api/server/controllers/agents/steer.js @@ -103,22 +103,60 @@ const createAgentAccessCheck = * (`handleSteerRequest`), which returns the HTTP status + JSON body to * serialize verbatim. DB access and permission services are injected here. */ -const SteerController = async (req, res) => { +const runSteerController = async (req, res, requireIdempotentDelivery) => { + const abortController = new AbortController(); + const abort = () => { + if (!res.writableEnded) { + abortController.abort(); + } + }; + req.once('aborted', abort); + res.once('close', abort); try { const generationProtocolVersion = getHostGenerationProtocol(req); + const checkAgentAccess = createAgentAccessCheck(req); + const expectedAgentId = requireIdempotentDelivery ? req.body?.agentId : undefined; const { status, body } = await handleSteerRequest(req.user ?? {}, req.body ?? {}, { generationProtocolVersion, + signal: abortController.signal, + ...(requireIdempotentDelivery && { requireIdempotentDelivery: true }), getFiles: db.getFiles, updateFilesUsage: db.updateFilesUsage, - checkAgentAccess: createAgentAccessCheck(req), + checkAgentAccess: requireIdempotentDelivery + ? async (run) => + typeof expectedAgentId === 'string' && + run.agentId === expectedAgentId && + isAgentsEndpoint(run.endpoint) && + (await checkAgentAccess(run)) + : checkAgentAccess, }); + if (res.destroyed || res.writableEnded) { + return; + } return sendProtocolResult(res, status, body); } catch (error) { logger.error('[SteerController] Failed to queue steer', error); + if (res.destroyed || res.headersSent) { + return; + } return sendProtocolFailure(res, 500, 'STEER_FAILED'); + } finally { + req.off('aborted', abort); + res.off('close', abort); } }; +const SteerController = (req, res) => runSteerController(req, res, false); + +/** + * Strict trigger-delivery endpoint. It shares the ordinary steer route's + * authentication, limiters, PII filter, moderation, owner/tenant checks, and + * agent ACL, while refusing any job/store path that cannot persist a durable + * clientSteerId receipt. The dedicated path also fails closed on old replicas + * during a rolling deploy: they return 404 instead of accepting a legacy steer. + */ +const SteerDeliveryController = (req, res) => runSteerController(req, res, true); + /** * POST /api/agents/chat/steer/cancel * @@ -163,5 +201,6 @@ const SteerArmController = async (req, res) => { }; module.exports = SteerController; +module.exports.SteerDeliveryController = SteerDeliveryController; module.exports.SteerCancelController = SteerCancelController; module.exports.SteerArmController = SteerArmController; diff --git a/api/server/routes/agents/index.js b/api/server/routes/agents/index.js index 1f53181e6d..987fba06df 100644 --- a/api/server/routes/agents/index.js +++ b/api/server/routes/agents/index.js @@ -913,6 +913,22 @@ router.post( SteerController, ); +/** + * @route POST /chat/steer/deliver + * @desc Strict, idempotent steer admission for trusted event-delivery hosts + * @access Private + * @description Uses the same text-admission chain as an interactive steer, + * then requires a v2 durable receipt and exact originating-agent identity. + */ +router.post( + '/chat/steer/deliver', + configMiddleware, + ...steerLimiters, + createMessageFilterPii({ getConfig: (req) => req.config?.messageFilter?.pii }), + moderateText, + SteerController.SteerDeliveryController, +); + /** * @route POST /chat/steer/cancel * @desc Remove a still-queued steer before injection (no model-bound content, diff --git a/packages/api/src/agents/envelope.ts b/packages/api/src/agents/envelope.ts index ba544138ee..cb56cdfe42 100644 --- a/packages/api/src/agents/envelope.ts +++ b/packages/api/src/agents/envelope.ts @@ -1,14 +1,15 @@ import type { TEphemeralAgent } from 'librechat-data-provider'; +import type { JsonPrimitive, JsonValue } from './json'; import type { ChatCompletionRequest } from './openai'; import type { ResponseRequest } from './responses'; +import { AGENT_ENVELOPE_MAX_NESTING_DEPTH, cloneJsonValue } from './json'; export const AGENT_RUN_ENVELOPE_VERSION = 1 as const; -export const AGENT_RUN_ENVELOPE_MAX_NESTING_DEPTH = 64; +export const AGENT_RUN_ENVELOPE_MAX_NESTING_DEPTH: number = AGENT_ENVELOPE_MAX_NESTING_DEPTH; export type AgentRunProtocol = 'chat.completions' | 'responses'; -export type JsonPrimitive = string | number | boolean | null; -export type JsonValue = JsonPrimitive | JsonValue[] | { [key: string]: JsonValue }; +export type { JsonPrimitive, JsonValue }; export interface AgentRunPrincipal { userId: string; @@ -91,105 +92,6 @@ function assertNonEmptyString(value: string | undefined, path: string): string { return value; } -function cloneJsonValue(value: T, path: string, ancestors: WeakSet, depth: number): T; -function cloneJsonValue( - value: unknown, - path: string, - ancestors: WeakSet, - depth: number, -): unknown { - if (depth > AGENT_RUN_ENVELOPE_MAX_NESTING_DEPTH) { - throw new AgentRunEnvelopeError( - `${path} exceeds the maximum nesting depth of ${AGENT_RUN_ENVELOPE_MAX_NESTING_DEPTH}`, - ); - } - - if (value === null || typeof value === 'string' || typeof value === 'boolean') { - return value; - } - - if (typeof value === 'number') { - if (!Number.isFinite(value)) { - throw new AgentRunEnvelopeError(`${path} must contain only finite numbers`); - } - return value; - } - - if (typeof value !== 'object') { - throw new AgentRunEnvelopeError(`${path} contains a non-JSON ${typeof value} value`); - } - - if (ancestors.has(value)) { - throw new AgentRunEnvelopeError(`${path} contains a circular reference`); - } - - ancestors.add(value); - - try { - const symbolKeys = Object.getOwnPropertySymbols(value); - if (symbolKeys.length > 0) { - throw new AgentRunEnvelopeError(`${path} contains symbol keys`); - } - - if (Array.isArray(value)) { - const cloned: unknown[] = new Array(value.length); - let clonedItemCount = 0; - for (const key of Object.getOwnPropertyNames(value)) { - if (key === 'length') { - continue; - } - const index = Number(key); - if ( - !Number.isSafeInteger(index) || - index < 0 || - index >= value.length || - String(index) !== key - ) { - throw new AgentRunEnvelopeError(`${path} contains non-index array properties`); - } - const descriptor = Object.getOwnPropertyDescriptor(value, key); - if (!descriptor || !Object.prototype.hasOwnProperty.call(descriptor, 'value')) { - throw new AgentRunEnvelopeError(`${path}[${index}] must not be an accessor property`); - } - const itemValue: unknown = descriptor.value; - cloned[index] = cloneJsonValue(itemValue, `${path}[${index}]`, ancestors, depth + 1); - clonedItemCount++; - } - if (clonedItemCount !== value.length) { - throw new AgentRunEnvelopeError(`${path} contains sparse array entries`); - } - return cloned; - } - - const prototype = Object.getPrototypeOf(value); - if (prototype !== Object.prototype && prototype !== null) { - const typeName = value.constructor?.name ?? 'object'; - throw new AgentRunEnvelopeError(`${path} contains a non-plain ${typeName} value`); - } - - const cloned: { [key: string]: unknown } = {}; - for (const key of Object.getOwnPropertyNames(value)) { - const descriptor = Object.getOwnPropertyDescriptor(value, key); - if (descriptor?.enumerable !== true) { - throw new AgentRunEnvelopeError(`${path}.${key} must be an enumerable property`); - } - if (!Object.prototype.hasOwnProperty.call(descriptor, 'value')) { - throw new AgentRunEnvelopeError(`${path}.${key} must not be an accessor property`); - } - const propertyValue: unknown = descriptor.value; - Object.defineProperty(cloned, key, { - configurable: true, - enumerable: true, - writable: true, - value: cloneJsonValue(propertyValue, `${path}.${key}`, ancestors, depth + 1), - }); - } - return cloned; - } finally { - ancestors.delete(value); - } -} - function createPrincipal(input: AgentRunPrincipalInput | null | undefined): AgentRunPrincipal { const userId = assertNonEmptyString(input?.id, 'principal.id'); const principal: AgentRunPrincipal = { userId }; @@ -233,7 +135,11 @@ export function createAgentRunEnvelope(input: CreateAgentRunEnvelopeInput): Agen return { ...base, protocol: input.protocol, - payload: cloneJsonValue(input.payload, 'payload', new WeakSet(), 0), + payload: cloneJsonValue( + input.payload, + 'payload', + (message) => new AgentRunEnvelopeError(message), + ), }; } @@ -241,7 +147,11 @@ export function createAgentRunEnvelope(input: CreateAgentRunEnvelopeInput): Agen return { ...base, protocol: input.protocol, - payload: cloneJsonValue(input.payload, 'payload', new WeakSet(), 0), + payload: cloneJsonValue( + input.payload, + 'payload', + (message) => new AgentRunEnvelopeError(message), + ), }; } diff --git a/packages/api/src/agents/index.ts b/packages/api/src/agents/index.ts index fb78376e8a..a91ec8fee0 100644 --- a/packages/api/src/agents/index.ts +++ b/packages/api/src/agents/index.ts @@ -43,6 +43,7 @@ export * from './sender'; export * from './hitl'; export * from './hooks'; export * from './steering'; +export * from './triggers'; export * from './activityLabels'; export * from './activityPhases'; export * from './reasoningLabels'; diff --git a/packages/api/src/agents/json.ts b/packages/api/src/agents/json.ts new file mode 100644 index 0000000000..2f4846a888 --- /dev/null +++ b/packages/api/src/agents/json.ts @@ -0,0 +1,109 @@ +export const AGENT_ENVELOPE_MAX_NESTING_DEPTH = 64; + +export type JsonPrimitive = string | number | boolean | null; +export type JsonValue = JsonPrimitive | JsonValue[] | { [key: string]: JsonValue }; + +type ErrorFactory = (message: string) => Error; + +export function cloneJsonValue( + value: T, + path: string, + createError: ErrorFactory, + ancestors: WeakSet = new WeakSet(), + depth = 0, +): T { + if (depth > AGENT_ENVELOPE_MAX_NESTING_DEPTH) { + throw createError( + `${path} exceeds the maximum nesting depth of ${AGENT_ENVELOPE_MAX_NESTING_DEPTH}`, + ); + } + + if (value === null || typeof value === 'string' || typeof value === 'boolean') { + return value; + } + + if (typeof value === 'number') { + if (!Number.isFinite(value)) { + throw createError(`${path} must contain only finite numbers`); + } + return value; + } + + if (typeof value !== 'object') { + throw createError(`${path} contains a non-JSON ${typeof value} value`); + } + + if (ancestors.has(value)) { + throw createError(`${path} contains a circular reference`); + } + + ancestors.add(value); + + try { + if (Object.getOwnPropertySymbols(value).length > 0) { + throw createError(`${path} contains symbol keys`); + } + + if (Array.isArray(value)) { + const cloned: unknown[] = new Array(value.length); + let clonedItemCount = 0; + for (const key of Object.getOwnPropertyNames(value)) { + if (key === 'length') { + continue; + } + const index = Number(key); + if ( + !Number.isSafeInteger(index) || + index < 0 || + index >= value.length || + String(index) !== key + ) { + throw createError(`${path} contains non-index array properties`); + } + const descriptor = Object.getOwnPropertyDescriptor(value, key); + if (!descriptor || !Object.prototype.hasOwnProperty.call(descriptor, 'value')) { + throw createError(`${path}[${index}] must not be an accessor property`); + } + const itemValue: unknown = descriptor.value; + cloned[index] = cloneJsonValue( + itemValue, + `${path}[${index}]`, + createError, + ancestors, + depth + 1, + ); + clonedItemCount++; + } + if (clonedItemCount !== value.length) { + throw createError(`${path} contains sparse array entries`); + } + return cloned as T; + } + + const prototype = Object.getPrototypeOf(value); + if (prototype !== Object.prototype && prototype !== null) { + throw createError(`${path} contains a non-plain object value`); + } + + const cloned: { [key: string]: unknown } = {}; + for (const key of Object.getOwnPropertyNames(value)) { + const descriptor = Object.getOwnPropertyDescriptor(value, key); + if (descriptor?.enumerable !== true) { + throw createError(`${path}.${key} must be an enumerable property`); + } + if (!Object.prototype.hasOwnProperty.call(descriptor, 'value')) { + throw createError(`${path}.${key} must not be an accessor property`); + } + const propertyValue: unknown = descriptor.value; + Object.defineProperty(cloned, key, { + configurable: true, + enumerable: true, + writable: true, + value: cloneJsonValue(propertyValue, `${path}.${key}`, createError, ancestors, depth + 1), + }); + } + return cloned as T; + } finally { + ancestors.delete(value); + } +} diff --git a/packages/api/src/agents/steering/__tests__/request.spec.ts b/packages/api/src/agents/steering/__tests__/request.spec.ts index 292728fe70..646e521ce6 100644 --- a/packages/api/src/agents/steering/__tests__/request.spec.ts +++ b/packages/api/src/agents/steering/__tests__/request.spec.ts @@ -66,6 +66,19 @@ describe('handleSteerRequest (real in-memory job manager)', () => { expect(result.body.code).toBe('INVALID_CLIENT_STEER_ID'); }); + it('requires a client correlation id when idempotent delivery is mandatory', async () => { + const result = await handleSteerRequest( + user, + { conversationId: 'c1', text: 'valid text' }, + { requireIdempotentDelivery: true }, + ); + + expect(result).toEqual({ + status: 400, + body: { code: 'CLIENT_STEER_ID_REQUIRED' }, + }); + }); + it('400s an invalid generation identity', async () => { const result = await handleSteerRequest(user, { conversationId: 'c1', @@ -669,6 +682,94 @@ describe('generation protocol bridge for steering mutations', () => { ]); }); + it('rejects a v1 job before enqueue when the caller requires idempotent delivery', async () => { + const streamId = 'steer-protocol-v1-idempotency-required'; + await GenerationJobManager.createJob(streamId, user.id, undefined, { + initialMetadata: { generationProtocolVersion: 1 }, + }); + const receiptEnqueue = jest.spyOn(GenerationJobManager.steering, 'enqueueWithReceipt'); + const legacyEnqueue = jest.spyOn(GenerationJobManager.steering, 'enqueueVersioned'); + + const result = await handleSteerRequest( + user, + { + conversationId: streamId, + clientSteerId: 'client-v1-requires-idempotency', + text: 'must not be injected without a receipt', + }, + { generationProtocolVersion: 2, requireIdempotentDelivery: true }, + ); + + expect(result).toEqual({ + status: 409, + body: { code: 'STEER_IDEMPOTENCY_UNAVAILABLE', generationProtocolVersion: 1 }, + }); + expect(receiptEnqueue).not.toHaveBeenCalled(); + expect(legacyEnqueue).not.toHaveBeenCalled(); + await expect(GenerationJobManager.steering.peek(streamId)).resolves.toEqual([]); + }); + + it('keeps ownership rejection ahead of the strict idempotency capability gate', async () => { + const streamId = 'steer-protocol-v1-idempotency-foreign-owner'; + await GenerationJobManager.createJob(streamId, 'someone-else', undefined, { + initialMetadata: { generationProtocolVersion: 1 }, + }); + + const result = await handleSteerRequest( + user, + { + conversationId: streamId, + clientSteerId: 'client-v1-foreign-owner', + text: 'must remain unauthorized', + }, + { generationProtocolVersion: 2, requireIdempotentDelivery: true }, + ); + + expect(result).toEqual({ + status: 403, + body: { code: 'UNAUTHORIZED', generationProtocolVersion: 1 }, + }); + await expect(GenerationJobManager.steering.peek(streamId)).resolves.toEqual([]); + }); + + it('observes delivery cancellation after async admission and before enqueue', async () => { + const streamId = 'steer-protocol-v2-aborted-admission'; + await GenerationJobManager.createJob(streamId, user.id, undefined, { + initialMetadata: { + agent_id: 'agent-1', + endpoint: 'agents', + generationProtocolVersion: 2, + }, + }); + const controller = new AbortController(); + const receiptEnqueue = jest.spyOn(GenerationJobManager.steering, 'enqueueWithReceipt'); + + const result = await handleSteerRequest( + user, + { + conversationId: streamId, + clientSteerId: 'client-v2-aborted-admission', + text: 'must not land after the worker lease is cancelled', + }, + { + generationProtocolVersion: 2, + requireIdempotentDelivery: true, + signal: controller.signal, + checkAgentAccess: async () => { + controller.abort(); + return true; + }, + }, + ); + + expect(result).toEqual({ + status: 499, + body: { code: 'STEER_ABORTED', generationProtocolVersion: 2 }, + }); + expect(receiptEnqueue).not.toHaveBeenCalled(); + await expect(GenerationJobManager.steering.peek(streamId)).resolves.toEqual([]); + }); + it('keeps v2 receipt replay and correlation broadcasts behind an exact v2 job marker', async () => { const streamId = 'steer-protocol-v2-job'; await GenerationJobManager.createJob(streamId, user.id, undefined, { diff --git a/packages/api/src/agents/steering/request.ts b/packages/api/src/agents/steering/request.ts index ab3b471a52..632a48799a 100644 --- a/packages/api/src/agents/steering/request.ts +++ b/packages/api/src/agents/steering/request.ts @@ -125,6 +125,12 @@ export interface SteerRequestDeps { * and its rollout gate. Direct package callers that omit it retain the * current-package (v2) behavior. */ generationProtocolVersion?: GenerationProtocolVersion; + /** Refuse a legacy v1 enqueue that cannot persist `clientSteerId` receipts. + * Background/event deliveries use this to make retrying an ambiguous + * outcome safe instead of potentially injecting the same instruction twice. */ + requireIdempotentDelivery?: boolean; + /** Best-effort cancellation observed immediately before durable mutation. */ + signal?: AbortSignal; /** Owner-scoped file fetch (`db.getFiles`-shaped). When present, every * client-supplied ref must resolve to an owned DB doc at enqueue and the * queued refs are replaced with DB-derived ones. */ @@ -156,6 +162,10 @@ function parseExpectedGenerationCreatedAt(value: unknown): { value?: number; inv return { value }; } +function isAborted(signal: AbortSignal | undefined): boolean { + return signal?.aborted === true; +} + /** Sanitizes client-supplied attachment refs via the shared ref picker; * a single unusable entry rejects the whole list (fail-loud for the client). */ function sanitizeSteerFiles(rawFiles: unknown): SanitizedFilesResult { @@ -382,6 +392,9 @@ async function handleSteerRequestInternal( ) { return { status: 400, body: { code: 'INVALID_CLIENT_STEER_ID' } }; } + if (deps.requireIdempotentDelivery === true && typeof clientSteerId !== 'string') { + return { status: 400, body: { code: 'CLIENT_STEER_ID_REQUIRED' } }; + } const { files, error: filesError } = sanitizeSteerFiles(body.files); if (filesError) { @@ -483,6 +496,9 @@ async function handleSteerRequestInternal( if (!isSteeringSupported()) { return { status: 501, body: { code: 'STEER_UNSUPPORTED' } }; } + if (deps.requireIdempotentDelivery === true && protocol.value !== 2) { + return { status: 409, body: { code: 'STEER_IDEMPOTENCY_UNAVAILABLE' } }; + } let queuedFiles = files; if (files && deps.getFiles) { @@ -531,6 +547,9 @@ async function handleSteerRequestInternal( ? { status: 409, body: { code: 'RUN_REPLACED' } } : { status: 404, body: { code: 'NO_ACTIVE_RUN' } }; } + if (isAborted(deps.signal)) { + return { status: 499, body: { code: 'STEER_ABORTED' } }; + } /** Normal sends make resolved uploads durable before model execution. Do * the same before enqueue: once a 202 is visible, neither an upload-window * sweep nor a process crash may turn the accepted steer into text-only @@ -538,6 +557,9 @@ async function handleSteerRequestInternal( if (!(await markSteerFilesUsed(streamId, { files: queuedFiles }, user, deps))) { return { status: 503, body: { code: 'STEER_FILE_RETENTION_FAILED' } }; } + if (isAborted(deps.signal)) { + return { status: 499, body: { code: 'STEER_ABORTED' } }; + } const item: SteerQueueItem = { steerId: randomUUID(), ...(protocol.value === 2 && typeof clientSteerId === 'string' && { clientSteerId }), diff --git a/packages/api/src/agents/triggers/dispatch.spec.ts b/packages/api/src/agents/triggers/dispatch.spec.ts new file mode 100644 index 0000000000..b86ae40b20 --- /dev/null +++ b/packages/api/src/agents/triggers/dispatch.spec.ts @@ -0,0 +1,114 @@ +import type { AgentTriggerEnvelope } from './envelope'; +import { AgentTriggerDispatchError, dispatchAgentTrigger } from './dispatch'; +import { createAgentTriggerEnvelope } from './envelope'; + +describe('dispatchAgentTrigger', () => { + const createFireInput = () => ({ + mode: 'fire' as const, + requestId: 'request-1', + deliveryId: 'delivery-1', + receivedAt: 20, + principal: { id: 'user-1' }, + target: { agentId: 'agent-1' }, + event: { + id: 'event-1', + type: 'resource.ready', + occurredAt: 10, + source: { id: 'source-1', type: 'webhook' }, + }, + input: 'Handle the ready resource.', + }); + + const fireEnvelope = () => createAgentTriggerEnvelope(createFireInput()); + + it('routes fire deliveries with their stable idempotency identity and abort signal', async () => { + const controller = new AbortController(); + const fire = jest.fn(async () => ({ status: 'accepted' as const })); + const steer = jest.fn(async () => ({ status: 'accepted' as const })); + const envelope = fireEnvelope(); + + await expect( + dispatchAgentTrigger(envelope, { fire, steer }, { signal: controller.signal }), + ).resolves.toEqual({ status: 'accepted' }); + + expect(fire).toHaveBeenCalledWith(envelope, { + idempotencyKey: expect.stringMatching(/^trigger_[a-f0-9]{64}$/), + signal: controller.signal, + }); + expect(steer).not.toHaveBeenCalled(); + }); + + it('routes steer deliveries without requiring a fire implementation detail', async () => { + const fire = jest.fn(async () => 'fire'); + const steer = jest.fn(async () => 'steer'); + const envelope = createAgentTriggerEnvelope({ + ...createFireInput(), + mode: 'steer', + target: { + agentId: 'agent-1', + conversationId: 'conversation-1', + generationCreatedAt: 15, + }, + }); + + await expect(dispatchAgentTrigger(envelope, { fire, steer })).resolves.toBe('steer'); + expect(steer).toHaveBeenCalledWith(envelope, { + idempotencyKey: expect.stringMatching(/^trigger_[a-f0-9]{64}$/), + }); + expect(fire).not.toHaveBeenCalled(); + }); + + it('propagates handler failures without falling back to another mode', async () => { + const error = new Error('fire rejected'); + const fire = jest.fn(async () => Promise.reject(error)); + const steer = jest.fn(async () => 'steer'); + + await expect(dispatchAgentTrigger(fireEnvelope(), { fire, steer })).rejects.toBe(error); + expect(fire).toHaveBeenCalledTimes(1); + expect(steer).not.toHaveBeenCalled(); + }); + + it('rejects unknown modes before deriving identity or calling a handler', () => { + const fire = jest.fn(async () => 'fire'); + const steer = jest.fn(async () => 'steer'); + const envelope = { + ...fireEnvelope(), + mode: 'resume', + } as unknown as AgentTriggerEnvelope; + + expect(() => dispatchAgentTrigger(envelope, { fire, steer })).toThrow( + new AgentTriggerDispatchError('Unsupported agent trigger mode: resume'), + ); + expect(fire).not.toHaveBeenCalled(); + expect(steer).not.toHaveBeenCalled(); + }); + + it('rejects unknown envelope versions before deriving identity or calling a handler', () => { + const fire = jest.fn(async () => 'fire'); + const steer = jest.fn(async () => 'steer'); + const envelope = { + ...fireEnvelope(), + version: 2, + } as unknown as AgentTriggerEnvelope; + + expect(() => dispatchAgentTrigger(envelope, { fire, steer })).toThrow( + new AgentTriggerDispatchError('Unsupported agent trigger envelope version: 2'), + ); + expect(fire).not.toHaveBeenCalled(); + expect(steer).not.toHaveBeenCalled(); + }); + + it('rejects malformed v1 envelopes before deriving identity or calling a handler', () => { + const fire = jest.fn(async () => 'fire'); + const steer = jest.fn(async () => 'steer'); + const malformed = { ...fireEnvelope() }; + Reflect.deleteProperty(malformed, 'target'); + const envelope = malformed as unknown as AgentTriggerEnvelope; + + expect(() => dispatchAgentTrigger(envelope, { fire, steer })).toThrow( + new AgentTriggerDispatchError('target must be an object'), + ); + expect(fire).not.toHaveBeenCalled(); + expect(steer).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/api/src/agents/triggers/dispatch.ts b/packages/api/src/agents/triggers/dispatch.ts new file mode 100644 index 0000000000..39b7a75107 --- /dev/null +++ b/packages/api/src/agents/triggers/dispatch.ts @@ -0,0 +1,55 @@ +import type { + AgentFireTriggerEnvelope, + AgentSteerTriggerEnvelope, + AgentTriggerEnvelope, +} from './envelope'; +import { getAgentTriggerIdempotencyKey, parseAgentTriggerEnvelope } from './envelope'; + +export interface AgentTriggerDispatchContext { + idempotencyKey: string; + signal?: AbortSignal; +} + +export class AgentTriggerDispatchError extends TypeError { + constructor(message: string) { + super(message); + this.name = 'AgentTriggerDispatchError'; + } +} + +/** + * Host-owned execution adapters. Each handler must enforce current authorization, + * limits, persistence, and the supplied idempotency identity before accepting work. + */ +export interface AgentTriggerDispatchHandlers { + fire: ( + envelope: AgentFireTriggerEnvelope, + context: AgentTriggerDispatchContext, + ) => Promise; + steer: ( + envelope: AgentSteerTriggerEnvelope, + context: AgentTriggerDispatchContext, + ) => Promise; +} + +/** Routes a normalized trigger without coupling its source to an execution transport. */ +export function dispatchAgentTrigger( + envelope: unknown, + handlers: AgentTriggerDispatchHandlers, + options?: { signal?: AbortSignal }, +): Promise { + let normalized: AgentTriggerEnvelope; + try { + normalized = parseAgentTriggerEnvelope(envelope); + } catch (error) { + throw new AgentTriggerDispatchError(error instanceof Error ? error.message : String(error)); + } + const context: AgentTriggerDispatchContext = { + idempotencyKey: getAgentTriggerIdempotencyKey(normalized), + ...(options?.signal != null && { signal: options.signal }), + }; + if (normalized.mode === 'fire') { + return handlers.fire(normalized, context); + } + return handlers.steer(normalized, context); +} diff --git a/packages/api/src/agents/triggers/envelope.spec.ts b/packages/api/src/agents/triggers/envelope.spec.ts new file mode 100644 index 0000000000..ad93e63230 --- /dev/null +++ b/packages/api/src/agents/triggers/envelope.spec.ts @@ -0,0 +1,217 @@ +import type { CreateAgentTriggerEnvelopeInput } from './envelope'; +import { + AGENT_TRIGGER_ENVELOPE_VERSION, + AgentTriggerEnvelopeError, + createAgentTriggerEnvelope, + getAgentTriggerIdempotencyKey, + parseAgentTriggerEnvelope, +} from './envelope'; + +describe('createAgentTriggerEnvelope', () => { + const createFireInput = (): CreateAgentTriggerEnvelopeInput => ({ + mode: 'fire', + requestId: 'request-1', + deliveryId: 'subscription-1:event-1', + receivedAt: 1_725_000_000_010, + principal: { + id: 'user-1', + role: 'USER', + tenantId: 'tenant-1', + }, + target: { agentId: 'agent-1' }, + event: { + id: 'event-1', + type: 'work.ready', + occurredAt: 1_725_000_000_000, + source: { id: 'source-1', type: 'mcp' }, + payload: { resourceId: 'resource-1', attempt: 1 }, + }, + input: 'Inspect the ready work item.', + }); + + it('creates a detached, versioned fire envelope with a projected principal', () => { + const input = createFireInput(); + const principal = input.principal as NonNullable< + CreateAgentTriggerEnvelopeInput['principal'] + > & { + password?: string; + }; + principal.password = 'must-not-cross'; + const envelope = createAgentTriggerEnvelope(input); + const payload = input.event.payload as { resourceId: string }; + payload.resourceId = 'changed-after-dispatch'; + + expect(envelope).toEqual({ + version: AGENT_TRIGGER_ENVELOPE_VERSION, + mode: 'fire', + requestId: 'request-1', + deliveryId: 'subscription-1:event-1', + receivedAt: 1_725_000_000_010, + principal: { userId: 'user-1', role: 'USER', tenantId: 'tenant-1' }, + target: { agentId: 'agent-1' }, + event: { + id: 'event-1', + type: 'work.ready', + occurredAt: 1_725_000_000_000, + source: { id: 'source-1', type: 'mcp' }, + payload: { resourceId: 'resource-1', attempt: 1 }, + }, + input: 'Inspect the ready work item.', + }); + expect(JSON.parse(JSON.stringify(envelope))).toEqual(envelope); + expect(JSON.stringify(envelope)).not.toContain('must-not-cross'); + }); + + it('parses and detaches a serialized envelope', () => { + const created = createAgentTriggerEnvelope(createFireInput()); + const serialized = JSON.parse(JSON.stringify(created)) as unknown; + const parsed = parseAgentTriggerEnvelope(serialized); + + expect(parsed).toEqual(created); + expect(parsed).not.toBe(serialized); + expect(parsed.event).not.toBe((serialized as { event: object }).event); + }); + + it('requires a generation fence for steer deliveries', () => { + const envelope = createAgentTriggerEnvelope({ + ...createFireInput(), + mode: 'steer', + target: { + agentId: 'agent-1', + conversationId: 'conversation-1', + generationCreatedAt: 1_725_000_000_005, + preempt: true, + }, + }); + + expect(envelope.mode).toBe('steer'); + expect(envelope.target).toEqual({ + agentId: 'agent-1', + conversationId: 'conversation-1', + generationCreatedAt: 1_725_000_000_005, + preempt: true, + }); + }); + + it('builds a stable generation-compatible idempotency key per delivery target', () => { + const first = createAgentTriggerEnvelope(createFireInput()); + const retry = createAgentTriggerEnvelope({ + ...createFireInput(), + requestId: 'request-2', + receivedAt: 1_725_000_000_020, + }); + const anotherDelivery = createAgentTriggerEnvelope({ + ...createFireInput(), + deliveryId: 'subscription-2:event-1', + }); + const anotherSource = createAgentTriggerEnvelope({ + ...createFireInput(), + event: { + ...createFireInput().event, + source: { id: 'source-2', type: 'webhook' }, + }, + }); + const anotherEvent = createAgentTriggerEnvelope({ + ...createFireInput(), + event: { ...createFireInput().event, id: 'event-2' }, + }); + + const firstKey = getAgentTriggerIdempotencyKey(first); + expect(getAgentTriggerIdempotencyKey(retry)).toBe(firstKey); + expect(getAgentTriggerIdempotencyKey(anotherDelivery)).not.toBe(firstKey); + expect(getAgentTriggerIdempotencyKey(anotherSource)).not.toBe(firstKey); + expect(getAgentTriggerIdempotencyKey(anotherEvent)).not.toBe(firstKey); + expect(firstKey).toMatch(/^trigger_[a-f0-9]{64}$/); + expect(firstKey).toMatch(/^[A-Za-z0-9_-]+$/); + expect(firstKey.length).toBeLessThanOrEqual(128); + }); + + it.each([ + ['requestId', { requestId: ' ' }], + ['deliveryId', { deliveryId: '' }], + ['principal.id', { principal: undefined }], + ['target.agentId', { target: { agentId: '' } }], + ['event', { event: undefined }], + ['event.id', { event: { ...createFireInput().event, id: '' } }], + ['event.type', { event: { ...createFireInput().event, type: '' } }], + ['event.source.id', { event: { ...createFireInput().event, source: { id: '', type: 'mcp' } } }], + ['input', { input: '' }], + ])('rejects an invalid %s', (_path, override) => { + expect(() => + createAgentTriggerEnvelope({ + ...createFireInput(), + ...override, + } as CreateAgentTriggerEnvelopeInput), + ).toThrow(AgentTriggerEnvelopeError); + }); + + it('rejects invalid timestamps and malformed steer controls', () => { + expect(() => + createAgentTriggerEnvelope({ + ...createFireInput(), + event: { ...createFireInput().event, occurredAt: Number.NaN }, + }), + ).toThrow('event.occurredAt must be a non-negative integer timestamp'); + + expect(() => + createAgentTriggerEnvelope({ + ...createFireInput(), + mode: 'steer', + target: { + agentId: 'agent-1', + conversationId: 'conversation-1', + generationCreatedAt: -1, + preempt: true, + }, + }), + ).toThrow('target.generationCreatedAt must be a non-negative integer timestamp'); + + expect(() => + createAgentTriggerEnvelope({ + ...createFireInput(), + mode: 'steer', + target: { + agentId: 'agent-1', + conversationId: 'conversation-1', + generationCreatedAt: 1, + preempt: 'yes', + }, + } as unknown as CreateAgentTriggerEnvelopeInput), + ).toThrow('target.preempt must be a boolean'); + + expect(() => + createAgentTriggerEnvelope({ + ...createFireInput(), + mode: 'steer', + target: undefined, + } as unknown as CreateAgentTriggerEnvelopeInput), + ).toThrow('target.agentId must be a non-empty string'); + }); + + it('rejects non-JSON and circular event payloads', () => { + expect(() => + createAgentTriggerEnvelope({ + ...createFireInput(), + event: { ...createFireInput().event, payload: { callback: () => undefined } }, + } as unknown as CreateAgentTriggerEnvelopeInput), + ).toThrow('event.payload.callback contains a non-JSON function value'); + + const payload: { self?: object } = {}; + payload.self = payload; + expect(() => + createAgentTriggerEnvelope({ + ...createFireInput(), + event: { ...createFireInput().event, payload }, + } as unknown as CreateAgentTriggerEnvelopeInput), + ).toThrow('event.payload.self contains a circular reference'); + }); + + it('rejects unknown trigger modes at runtime', () => { + expect(() => + createAgentTriggerEnvelope({ + ...createFireInput(), + mode: 'resume', + } as unknown as CreateAgentTriggerEnvelopeInput), + ).toThrow('Unsupported agent trigger mode: resume'); + }); +}); diff --git a/packages/api/src/agents/triggers/envelope.ts b/packages/api/src/agents/triggers/envelope.ts new file mode 100644 index 0000000000..bdc761e604 --- /dev/null +++ b/packages/api/src/agents/triggers/envelope.ts @@ -0,0 +1,296 @@ +import { createHash } from 'node:crypto'; +import type { AgentRunPrincipal, AgentRunPrincipalInput } from '../envelope'; +import type { JsonValue } from '../json'; +import { cloneJsonValue } from '../json'; + +export const AGENT_TRIGGER_ENVELOPE_VERSION = 1 as const; +export const AGENT_TRIGGER_IDEMPOTENCY_PREFIX = 'trigger_'; + +export type AgentTriggerMode = 'fire' | 'steer'; + +export interface AgentTriggerSource { + /** Stable identity of the configured source, such as a webhook or schedule id. */ + id: string; + /** Source adapter kind. Core dispatch does not interpret this value. */ + type: string; +} + +export interface AgentTriggerEvent { + /** Stable identity assigned by the source for this event occurrence. */ + id: string; + type: string; + /** Unix time in milliseconds recorded by the source. */ + occurredAt: number; + /** Sanitized source data; adapters must omit credentials and transport secrets. */ + payload?: JsonValue; + source: AgentTriggerSource; +} + +interface AgentTriggerTarget { + agentId: string; +} + +/** + * One fire delivery represents one new conversation for this agent; retries + * reuse the delivery id instead of starting another conversation. + */ +export type AgentFireTarget = AgentTriggerTarget; + +export interface AgentSteerTarget extends AgentTriggerTarget { + /** Existing conversation whose active generation receives the input. */ + conversationId: string; + /** Fences delivery to the generation observed by the event adapter. */ + generationCreatedAt: number; + /** Request an interrupt when supported; otherwise delivery stays queued. */ + preempt?: boolean; +} + +interface AgentTriggerEnvelopeBase { + version: typeof AGENT_TRIGGER_ENVELOPE_VERSION; + requestId: string; + /** Stable across retries of one source-event-to-target delivery. */ + deliveryId: string; + /** Unix time in milliseconds recorded by the receiving host. */ + receivedAt: number; + principal: AgentRunPrincipal; + event: AgentTriggerEvent; + /** Host-rendered agent input; source payload remains available in `event.payload`. */ + input: string; +} + +export interface AgentFireTriggerEnvelope extends AgentTriggerEnvelopeBase { + mode: 'fire'; + target: AgentFireTarget; +} + +export interface AgentSteerTriggerEnvelope extends AgentTriggerEnvelopeBase { + mode: 'steer'; + target: AgentSteerTarget; +} + +export type AgentTriggerEnvelope = AgentFireTriggerEnvelope | AgentSteerTriggerEnvelope; + +interface CreateAgentTriggerEnvelopeBase { + requestId: string; + deliveryId: string; + receivedAt: number; + principal: AgentRunPrincipalInput | null | undefined; + event: AgentTriggerEvent; + input: string; +} + +export type CreateAgentTriggerEnvelopeInput = + | (CreateAgentTriggerEnvelopeBase & { + mode: 'fire'; + target: AgentFireTarget; + }) + | (CreateAgentTriggerEnvelopeBase & { + mode: 'steer'; + target: AgentSteerTarget; + }); + +export class AgentTriggerEnvelopeError extends TypeError { + constructor(message: string) { + super(message); + this.name = 'AgentTriggerEnvelopeError'; + } +} + +function error(message: string): AgentTriggerEnvelopeError { + return new AgentTriggerEnvelopeError(message); +} + +function requireString(value: unknown, path: string): string { + if (typeof value !== 'string' || value.trim().length === 0) { + throw error(`${path} must be a non-empty string`); + } + return value; +} + +function requireTimestamp(value: unknown, path: string): number { + if (typeof value !== 'number' || !Number.isSafeInteger(value) || value < 0) { + throw error(`${path} must be a non-negative integer timestamp`); + } + return value; +} + +function requireRecord(value: unknown, path: string): Record { + if (value == null || typeof value !== 'object' || Array.isArray(value)) { + throw error(`${path} must be an object`); + } + return value as Record; +} + +function createPrincipal(input: AgentRunPrincipalInput | null | undefined): AgentRunPrincipal { + const principal: AgentRunPrincipal = { + userId: requireString(input?.id, 'principal.id'), + }; + if (input?.role != null) { + principal.role = requireString(input.role, 'principal.role'); + } + if (input?.tenantId != null) { + principal.tenantId = requireString(input.tenantId, 'principal.tenantId'); + } + return principal; +} + +function createEvent(input: AgentTriggerEvent | null | undefined): AgentTriggerEvent { + return { + id: requireString(input?.id, 'event.id'), + type: requireString(input?.type, 'event.type'), + occurredAt: requireTimestamp(input?.occurredAt, 'event.occurredAt'), + source: { + id: requireString(input?.source?.id, 'event.source.id'), + type: requireString(input?.source?.type, 'event.source.type'), + }, + ...(input?.payload !== undefined && { + payload: cloneJsonValue(input.payload, 'event.payload', error), + }), + }; +} + +export function createAgentTriggerEnvelope( + input: CreateAgentTriggerEnvelopeInput, +): AgentTriggerEnvelope { + const receivedMode: string = input.mode; + const base: AgentTriggerEnvelopeBase = { + version: AGENT_TRIGGER_ENVELOPE_VERSION, + requestId: requireString(input.requestId, 'requestId'), + deliveryId: requireString(input.deliveryId, 'deliveryId'), + receivedAt: requireTimestamp(input.receivedAt, 'receivedAt'), + principal: createPrincipal(input.principal), + event: createEvent(input.event), + input: requireString(input.input, 'input'), + }; + + if (input.mode === 'fire') { + return { + ...base, + mode: input.mode, + target: { agentId: requireString(input.target?.agentId, 'target.agentId') }, + }; + } + + if (input.mode === 'steer') { + const target = input.target; + if (target?.preempt != null && typeof target.preempt !== 'boolean') { + throw error('target.preempt must be a boolean'); + } + return { + ...base, + mode: input.mode, + target: { + agentId: requireString(target?.agentId, 'target.agentId'), + conversationId: requireString(target?.conversationId, 'target.conversationId'), + generationCreatedAt: requireTimestamp( + target?.generationCreatedAt, + 'target.generationCreatedAt', + ), + ...(target?.preempt != null && { preempt: target.preempt }), + }, + }; + } + + throw error(`Unsupported agent trigger mode: ${receivedMode}`); +} + +/** + * Validates and detaches an envelope received from a queue or another process. + * Unknown versions, modes, and malformed v1 fields fail closed before dispatch. + */ +export function parseAgentTriggerEnvelope(input: unknown): AgentTriggerEnvelope { + const envelope = requireRecord(cloneJsonValue(input, 'envelope', error), 'envelope'); + if (envelope.version !== AGENT_TRIGGER_ENVELOPE_VERSION) { + throw error(`Unsupported agent trigger envelope version: ${String(envelope.version)}`); + } + + const mode = envelope.mode; + if (mode !== 'fire' && mode !== 'steer') { + throw error(`Unsupported agent trigger mode: ${String(mode)}`); + } + + const principalInput = requireRecord(envelope.principal, 'principal'); + const principal: AgentRunPrincipal = { + userId: requireString(principalInput.userId, 'principal.userId'), + }; + if (principalInput.role != null) { + principal.role = requireString(principalInput.role, 'principal.role'); + } + if (principalInput.tenantId != null) { + principal.tenantId = requireString(principalInput.tenantId, 'principal.tenantId'); + } + + const eventInput = requireRecord(envelope.event, 'event'); + const sourceInput = requireRecord(eventInput.source, 'event.source'); + const event: AgentTriggerEvent = { + id: requireString(eventInput.id, 'event.id'), + type: requireString(eventInput.type, 'event.type'), + occurredAt: requireTimestamp(eventInput.occurredAt, 'event.occurredAt'), + source: { + id: requireString(sourceInput.id, 'event.source.id'), + type: requireString(sourceInput.type, 'event.source.type'), + }, + ...(eventInput.payload !== undefined && { payload: eventInput.payload as JsonValue }), + }; + + const base: AgentTriggerEnvelopeBase = { + version: AGENT_TRIGGER_ENVELOPE_VERSION, + requestId: requireString(envelope.requestId, 'requestId'), + deliveryId: requireString(envelope.deliveryId, 'deliveryId'), + receivedAt: requireTimestamp(envelope.receivedAt, 'receivedAt'), + principal, + event, + input: requireString(envelope.input, 'input'), + }; + const target = requireRecord(envelope.target, 'target'); + + if (mode === 'fire') { + return { + ...base, + mode, + target: { agentId: requireString(target.agentId, 'target.agentId') }, + }; + } + + if (target.preempt != null && typeof target.preempt !== 'boolean') { + throw error('target.preempt must be a boolean'); + } + return { + ...base, + mode, + target: { + agentId: requireString(target.agentId, 'target.agentId'), + conversationId: requireString(target.conversationId, 'target.conversationId'), + generationCreatedAt: requireTimestamp( + target.generationCreatedAt, + 'target.generationCreatedAt', + ), + ...(target.preempt != null && { preempt: target.preempt }), + }, + }; +} + +/** + * Stable idempotency key for one target delivery. Its character set and length + * are accepted by both generation `clientRequestId` and steering `clientSteerId`. + */ +export function getAgentTriggerIdempotencyKey(envelope: AgentTriggerEnvelope): string { + const digest = createHash('sha256') + .update( + JSON.stringify([ + envelope.version, + envelope.principal.tenantId ?? '', + envelope.principal.userId, + envelope.event.source.type, + envelope.event.source.id, + envelope.event.type, + envelope.event.id, + envelope.deliveryId, + envelope.mode, + envelope.target.agentId, + envelope.mode === 'steer' ? envelope.target.conversationId : '', + ]), + ) + .digest('hex'); + return `${AGENT_TRIGGER_IDEMPOTENCY_PREFIX}${digest}`; +} diff --git a/packages/api/src/agents/triggers/host.spec.ts b/packages/api/src/agents/triggers/host.spec.ts new file mode 100644 index 0000000000..d94a935b05 --- /dev/null +++ b/packages/api/src/agents/triggers/host.spec.ts @@ -0,0 +1,624 @@ +import { Constants, EModelEndpoint } from 'librechat-data-provider'; +import { getRequestId, getTenantId, getUserId } from '@librechat/data-schemas'; +import type { AgentTriggerExecutionHostDeps, AgentTriggerFetch } from './host'; +import { createAgentTriggerEnvelope, getAgentTriggerIdempotencyKey } from './envelope'; +import { AgentTriggerExecutionError, createAgentTriggerExecutionHost } from './host'; + +const createFireEnvelope = () => + createAgentTriggerEnvelope({ + mode: 'fire', + requestId: 'request-1', + deliveryId: 'delivery-1', + receivedAt: 20, + principal: { id: 'user-1', role: 'member', tenantId: 'tenant-1' }, + target: { agentId: 'agent-1' }, + event: { + id: 'event-1', + type: 'resource.ready', + occurredAt: 10, + source: { id: 'source-1', type: 'webhook' }, + payload: { resourceId: 'resource-1' }, + }, + input: 'Handle the ready resource.', + }); + +const createSteerEnvelope = () => + createAgentTriggerEnvelope({ + mode: 'steer', + requestId: 'request-2', + deliveryId: 'delivery-2', + receivedAt: 30, + principal: { id: 'user-1', role: 'member', tenantId: 'tenant-1' }, + target: { + agentId: 'agent-1', + conversationId: 'conversation-1', + generationCreatedAt: 25, + preempt: true, + }, + event: { + id: 'event-2', + type: 'opponent.moved', + occurredAt: 21, + source: { id: 'game-1', type: 'mcp' }, + }, + input: 'The opponent moved. Take your turn.', + }); + +function response(payload: unknown, init?: ResponseInit): Response { + return new Response(JSON.stringify(payload), { + status: 200, + headers: { 'content-type': 'application/json' }, + ...init, + }); +} + +function fetchMock(implementation: AgentTriggerFetch): jest.MockedFunction { + return jest.fn(implementation); +} + +function deps( + fetcher: jest.MockedFunction, + overrides: Partial = {}, +): AgentTriggerExecutionHostDeps { + return { + getBaseUrl: () => 'http://127.0.0.1:3080', + mintToken: () => 'signed-token', + fetch: fetcher, + ...overrides, + }; +} + +function expectExecutionError(error: unknown, expected: Partial): void { + expect(error).toBeInstanceOf(AgentTriggerExecutionError); + expect(error).toMatchObject(expected); +} + +describe('createAgentTriggerExecutionHost fire adapter', () => { + it('rejects malformed serialized envelopes through its promise contract', async () => { + const host = createAgentTriggerExecutionHost(deps(fetchMock(async () => response({})))); + const malformed = { ...createFireEnvelope() }; + Reflect.deleteProperty(malformed, 'requestId'); + + await expect(host.dispatch(malformed)).rejects.toThrow('requestId must be a non-empty string'); + }); + + it('starts a new run with stable identity, principal context, and no source coupling', async () => { + const envelope = createFireEnvelope(); + const idempotencyKey = getAgentTriggerIdempotencyKey(envelope); + const fetcher = fetchMock(async () => + response({ + streamId: 'stream-1', + conversationId: 'conversation-1', + generationCreatedAt: 40, + status: 'started', + }), + ); + const mintToken = jest.fn(() => { + expect(getUserId()).toBe('user-1'); + expect(getTenantId()).toBe('tenant-1'); + expect(getRequestId()).toBe(idempotencyKey); + return 'signed-token'; + }); + const host = createAgentTriggerExecutionHost( + deps(fetcher, { + getBaseUrl: () => 'https://chat.example.test/base/?ignored=true', + mintToken, + getTimezone: async () => ' America/New_York ', + }), + ); + + await expect(host.dispatch(envelope)).resolves.toEqual({ + mode: 'fire', + streamId: 'stream-1', + conversationId: 'conversation-1', + generationCreatedAt: 40, + status: 'started', + }); + + expect(fetcher).toHaveBeenCalledTimes(1); + const [input, init] = fetcher.mock.calls[0]; + expect(String(input)).toBe('https://chat.example.test/base/api/agents/chat/agents'); + const headers = new Headers(init?.headers); + expect(headers.get('authorization')).toBe('Bearer signed-token'); + expect(headers.get('x-request-id')).toBe(idempotencyKey); + expect(headers.get('x-librechat-generation-protocol')).toBe('2'); + expect(headers.get('user-agent')).toContain('LibreChat-Agent-Trigger/1'); + expect(JSON.parse(String(init?.body))).toEqual({ + text: envelope.input, + endpoint: EModelEndpoint.agents, + agent_id: envelope.target.agentId, + parentMessageId: Constants.NO_PARENT, + isContinued: false, + isRegenerate: false, + clientRequestId: idempotencyKey, + generationProtocolVersion: 2, + timezone: 'America/New_York', + }); + expect(mintToken).toHaveBeenCalledWith(envelope.principal, envelope); + }); + + it('reuses the same generation identity when an ambiguous delivery is retried', async () => { + const envelope = createFireEnvelope(); + const fetcher = fetchMock( + jest + .fn() + .mockResolvedValueOnce( + response({ + streamId: 'stream-1', + conversationId: 'conversation-1', + generationCreatedAt: 40, + status: 'started', + }), + ) + .mockResolvedValueOnce( + response({ + streamId: 'stream-1', + conversationId: 'conversation-1', + generationCreatedAt: 40, + status: 'resumed', + }), + ), + ); + const host = createAgentTriggerExecutionHost(deps(fetcher)); + + await host.dispatch(envelope); + await expect(host.dispatch(envelope)).resolves.toMatchObject({ + status: 'resumed', + conversationId: 'conversation-1', + }); + + const bodies = fetcher.mock.calls.map(([, init]) => JSON.parse(String(init?.body)) as unknown); + expect(bodies).toEqual([ + expect.objectContaining({ clientRequestId: getAgentTriggerIdempotencyKey(envelope) }), + expect.objectContaining({ clientRequestId: getAgentTriggerIdempotencyKey(envelope) }), + ]); + }); + + it('classifies a server rejection as definite and preserves retry guidance', async () => { + expect.hasAssertions(); + const fetcher = fetchMock(async () => + response( + { code: 'SERVER_NOT_READY', error: 'Try shortly.' }, + { status: 503, headers: { 'retry-after': '2' } }, + ), + ); + const host = createAgentTriggerExecutionHost(deps(fetcher)); + + await host.dispatch(createFireEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + mode: 'fire', + certainty: 'definite', + retryable: true, + code: 'SERVER_NOT_READY', + status: 503, + retryAfter: '2', + }); + }); + }); + + it('does not mark an authorization rejection retryable', async () => { + expect.hasAssertions(); + const fetcher = fetchMock(async () => response({ code: 'FORBIDDEN' }, { status: 403 })); + const host = createAgentTriggerExecutionHost(deps(fetcher)); + + await host.dispatch(createFireEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + certainty: 'definite', + retryable: false, + code: 'FORBIDDEN', + status: 403, + }); + }); + }); + + it('retries an expired delivery token so the next attempt can remint it', async () => { + expect.hasAssertions(); + const fetcher = fetchMock(async () => response({ code: 'UNAUTHORIZED' }, { status: 401 })); + const host = createAgentTriggerExecutionHost(deps(fetcher)); + + await host.dispatch(createFireEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + mode: 'fire', + certainty: 'definite', + retryable: true, + code: 'UNAUTHORIZED', + status: 401, + }); + }); + }); + + it('treats malformed success as ambiguous because the generation may have started', async () => { + expect.hasAssertions(); + const fetcher = fetchMock(async () => response({ status: 'started' })); + const host = createAgentTriggerExecutionHost(deps(fetcher)); + + await host.dispatch(createFireEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + mode: 'fire', + certainty: 'ambiguous', + retryable: true, + code: 'INVALID_RESPONSE', + status: 200, + }); + }); + }); + + it('keeps the timeout active while reading the accepted response body', async () => { + expect.hasAssertions(); + const fetcher = fetchMock(async (_input, init) => { + const signal = init?.signal; + return new Response( + new ReadableStream({ + start(controller) { + const abort = () => controller.error(new Error('body aborted')); + if (signal?.aborted === true) { + abort(); + } else { + signal?.addEventListener('abort', abort, { once: true }); + } + }, + }), + ); + }); + const host = createAgentTriggerExecutionHost(deps(fetcher, { timeoutMs: 10 })); + + await host.dispatch(createFireEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + certainty: 'ambiguous', + retryable: true, + code: 'TIMEOUT', + status: 200, + }); + }); + }); + + it('applies the timeout while asynchronous setup is still pending', async () => { + expect.hasAssertions(); + const fetcher = fetchMock(async () => response({})); + const host = createAgentTriggerExecutionHost( + deps(fetcher, { + mintToken: () => new Promise(() => undefined), + timeoutMs: 10, + }), + ); + + await host.dispatch(createFireEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + mode: 'fire', + certainty: 'definite', + retryable: true, + code: 'TIMEOUT', + }); + }); + expect(fetcher).not.toHaveBeenCalled(); + }); + + it('starts independent token, timezone, and origin setup concurrently', async () => { + let resolveToken!: (value: string) => void; + let resolveTimezone!: (value: string) => void; + const token = new Promise((resolve) => { + resolveToken = resolve; + }); + const timezone = new Promise((resolve) => { + resolveTimezone = resolve; + }); + const mintToken = jest.fn(() => token); + const getTimezone = jest.fn(() => timezone); + const getBaseUrl = jest.fn(() => 'http://127.0.0.1:3080'); + const fetcher = fetchMock(async () => + response({ + status: 'started', + streamId: 'stream-1', + conversationId: 'conversation-1', + }), + ); + const host = createAgentTriggerExecutionHost({ + mintToken, + getTimezone, + getBaseUrl, + fetch: fetcher, + }); + + const pending = host.dispatch(createFireEnvelope()); + await Promise.resolve(); + await Promise.resolve(); + + expect(mintToken).toHaveBeenCalledTimes(1); + expect(getTimezone).toHaveBeenCalledTimes(1); + expect(getBaseUrl).toHaveBeenCalledTimes(1); + expect(fetcher).not.toHaveBeenCalled(); + + resolveToken('signed-token'); + resolveTimezone('UTC'); + await expect(pending).resolves.toMatchObject({ mode: 'fire', status: 'started' }); + }); + + it('observes caller cancellation while asynchronous setup is pending', async () => { + expect.hasAssertions(); + const fetcher = fetchMock(async () => response({})); + const host = createAgentTriggerExecutionHost( + deps(fetcher, { mintToken: () => new Promise(() => undefined) }), + ); + const controller = new AbortController(); + const pending = host.dispatch(createFireEnvelope(), { signal: controller.signal }); + controller.abort(); + + await pending.catch((error: unknown) => { + expectExecutionError(error, { + mode: 'fire', + certainty: 'definite', + retryable: true, + code: 'ABORTED', + }); + }); + expect(fetcher).not.toHaveBeenCalled(); + }); + + it('marks thrown setup dependency failures retryable without sending', async () => { + expect.hasAssertions(); + const fetcher = fetchMock(async () => response({})); + const host = createAgentTriggerExecutionHost( + deps(fetcher, { + mintToken: () => new Promise(() => undefined), + getTimezone: async () => Promise.reject(new Error('timezone store unavailable')), + }), + ); + + await host.dispatch(createFireEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + mode: 'fire', + certainty: 'definite', + retryable: true, + code: 'SETUP_FAILED', + }); + }); + expect(fetcher).not.toHaveBeenCalled(); + }); + + it('cancels an oversized success body instead of buffering it without a bound', async () => { + expect.hasAssertions(); + const host = createAgentTriggerExecutionHost( + deps(fetchMock(async () => response({ padding: 'x'.repeat(70 * 1024) }))), + ); + + await host.dispatch(createFireEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + mode: 'fire', + certainty: 'ambiguous', + retryable: true, + code: 'RESPONSE_TOO_LARGE', + status: 200, + }); + }); + }); + + it('rejects invalid host configuration before making a network request', async () => { + const fetcher = fetchMock(async () => response({})); + const host = createAgentTriggerExecutionHost( + deps(fetcher, { getBaseUrl: () => 'file:///tmp/not-http' }), + ); + + await host.dispatch(createFireEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + certainty: 'definite', + retryable: false, + code: 'INVALID_BASE_URL', + }); + }); + expect(fetcher).not.toHaveBeenCalled(); + }); + + it('distinguishes pre-connect failures from unknown network outcomes', async () => { + expect.hasAssertions(); + const refused = new TypeError('fetch failed'); + Object.assign(refused, { cause: { code: 'ECONNREFUSED' } }); + const preconnect = createAgentTriggerExecutionHost( + deps(fetchMock(async () => Promise.reject(refused))), + ); + const reset = createAgentTriggerExecutionHost( + deps(fetchMock(async () => Promise.reject(new TypeError('connection reset')))), + ); + + await preconnect.dispatch(createFireEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + certainty: 'definite', + retryable: true, + code: 'NETWORK_ERROR', + }); + }); + await reset.dispatch(createFireEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + certainty: 'ambiguous', + retryable: true, + code: 'NETWORK_ERROR', + }); + }); + }); + + it('does not send when the delivery was already aborted', async () => { + const fetcher = fetchMock(async () => response({})); + const host = createAgentTriggerExecutionHost(deps(fetcher)); + const controller = new AbortController(); + controller.abort(); + + await host + .dispatch(createFireEnvelope(), { signal: controller.signal }) + .catch((error: unknown) => { + expectExecutionError(error, { + certainty: 'definite', + retryable: true, + code: 'ABORTED', + }); + }); + expect(fetcher).not.toHaveBeenCalled(); + }); +}); + +describe('createAgentTriggerExecutionHost steer adapter', () => { + it('steers through the authenticated admission route with a strict v2 receipt', async () => { + const envelope = createSteerEnvelope(); + const idempotencyKey = getAgentTriggerIdempotencyKey(envelope); + const mintToken = jest.fn(() => { + expect(getUserId()).toBe('user-1'); + expect(getTenantId()).toBe('tenant-1'); + expect(getRequestId()).toBe(idempotencyKey); + return 'signed-token'; + }); + const fetcher = fetchMock(async () => + response( + { + status: 'queued', + conversationId: 'conversation-1', + steerId: 'steer-1', + position: 1, + preempt: true, + preemptRevision: 4, + generationProtocolVersion: 2, + }, + { status: 202 }, + ), + ); + const host = createAgentTriggerExecutionHost( + deps(fetcher, { + getBaseUrl: () => 'https://chat.example.test/base/', + mintToken, + }), + ); + + await expect(host.dispatch(envelope)).resolves.toEqual({ + mode: 'steer', + status: 'queued', + conversationId: 'conversation-1', + steerId: 'steer-1', + position: 1, + preempt: true, + preemptRevision: 4, + }); + expect(fetcher).toHaveBeenCalledTimes(1); + const [input, init] = fetcher.mock.calls[0]; + expect(String(input)).toBe('https://chat.example.test/base/api/agents/chat/steer/deliver'); + const headers = new Headers(init?.headers); + expect(headers.get('authorization')).toBe('Bearer signed-token'); + expect(headers.get('x-request-id')).toBe(idempotencyKey); + expect(headers.get('x-librechat-generation-protocol')).toBe('2'); + expect(JSON.parse(String(init?.body))).toEqual({ + agentId: 'agent-1', + conversationId: 'conversation-1', + generationCreatedAt: 25, + text: envelope.input, + clientSteerId: idempotencyKey, + preempt: true, + generationProtocolVersion: 2, + }); + expect(mintToken).toHaveBeenCalledWith(envelope.principal, envelope); + }); + + it('surfaces temporary steer backpressure as a retryable definite rejection', async () => { + expect.hasAssertions(); + const host = createAgentTriggerExecutionHost( + deps( + fetchMock(async () => + response( + { code: 'STEER_QUEUE_FULL', generationProtocolVersion: 2 }, + { status: 429, headers: { 'retry-after': '3' } }, + ), + ), + ), + ); + + await host.dispatch(createSteerEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + certainty: 'definite', + retryable: true, + code: 'STEER_QUEUE_FULL', + status: 429, + retryAfter: '3', + }); + }); + }); + + it('retries a steer while the same generation is paused for human review', async () => { + expect.hasAssertions(); + const host = createAgentTriggerExecutionHost( + deps( + fetchMock(async () => + response({ code: 'RUN_PAUSED', generationProtocolVersion: 2 }, { status: 409 }), + ), + ), + ); + + await host.dispatch(createSteerEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + mode: 'steer', + certainty: 'definite', + retryable: true, + code: 'RUN_PAUSED', + status: 409, + }); + }); + }); + + it('retries a missing strict route during a rolling deployment without mutating legacy state', async () => { + expect.hasAssertions(); + const host = createAgentTriggerExecutionHost( + deps(fetchMock(async () => response('Not Found', { status: 404 }))), + ); + + await host.dispatch(createSteerEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + mode: 'steer', + certainty: 'definite', + retryable: true, + code: 'STEER_ADMISSION_UNAVAILABLE', + status: 404, + }); + }); + }); + + it('classifies a reset during steer admission as ambiguous and safe to retry', async () => { + expect.hasAssertions(); + const host = createAgentTriggerExecutionHost( + deps(fetchMock(async () => Promise.reject(new Error('connection reset')))), + ); + + await host.dispatch(createSteerEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + mode: 'steer', + certainty: 'ambiguous', + retryable: true, + code: 'NETWORK_ERROR', + }); + }); + }); + + it('refuses to retry a success that lacks the v2 receipt guarantee', async () => { + expect.hasAssertions(); + const host = createAgentTriggerExecutionHost( + deps( + fetchMock(async () => + response( + { + status: 'queued', + conversationId: 'conversation-1', + steerId: 'steer-1', + position: 1, + preempt: false, + generationProtocolVersion: 1, + }, + { status: 202 }, + ), + ), + ), + ); + + await host.dispatch(createSteerEnvelope()).catch((error: unknown) => { + expectExecutionError(error, { + mode: 'steer', + certainty: 'ambiguous', + retryable: false, + code: 'STEER_IDEMPOTENCY_UNAVAILABLE', + status: 202, + }); + }); + }); +}); diff --git a/packages/api/src/agents/triggers/host.ts b/packages/api/src/agents/triggers/host.ts new file mode 100644 index 0000000000..a59be869b2 --- /dev/null +++ b/packages/api/src/agents/triggers/host.ts @@ -0,0 +1,811 @@ +import { tenantStorage } from '@librechat/data-schemas'; +import { Constants, EModelEndpoint } from 'librechat-data-provider'; +import type { + AgentFireTriggerEnvelope, + AgentSteerTriggerEnvelope, + AgentTriggerEnvelope, +} from './envelope'; +import type { AgentTriggerDispatchContext } from './dispatch'; +import type { AgentRunPrincipal } from '../envelope'; +import { dispatchAgentTrigger } from './dispatch'; + +const DEFAULT_FIRE_TIMEOUT_MS = 30_000; +const MAX_RESPONSE_BODY_BYTES = 64 * 1024; +const GENERATION_PROTOCOL_HEADER = 'x-librechat-generation-protocol'; +const TRIGGER_USER_AGENT = + 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 ' + + '(KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36 LibreChat-Agent-Trigger/1'; + +const PRE_CONNECT_ERROR_CODES = new Set([ + 'ECONNREFUSED', + 'ENOTFOUND', + 'EAI_AGAIN', + 'EHOSTUNREACH', + 'ENETUNREACH', + 'UND_ERR_CONNECT_TIMEOUT', +]); + +type MaybePromise = T | Promise; +type FireStatus = 'started' | 'resumed' | 'replaced' | 'settled'; + +export type AgentTriggerFetch = (input: string | URL, init?: RequestInit) => Promise; + +export type AgentTriggerFailureCertainty = 'definite' | 'ambiguous'; + +export interface AgentTriggerExecutionErrorOptions { + mode: 'fire' | 'steer'; + certainty: AgentTriggerFailureCertainty; + retryable: boolean; + code?: string; + status?: number; + retryAfter?: string; +} + +/** + * A typed delivery failure. `ambiguous` means the action may already be + * committed; callers may retry only the same envelope so its idempotency key + * remains unchanged. + */ +export class AgentTriggerExecutionError extends Error { + readonly mode: 'fire' | 'steer'; + readonly certainty: AgentTriggerFailureCertainty; + readonly retryable: boolean; + readonly code?: string; + readonly status?: number; + readonly retryAfter?: string; + + constructor(message: string, options: AgentTriggerExecutionErrorOptions) { + super(message); + this.name = 'AgentTriggerExecutionError'; + this.mode = options.mode; + this.certainty = options.certainty; + this.retryable = options.retryable; + this.code = options.code; + this.status = options.status; + this.retryAfter = options.retryAfter; + } +} + +export interface AgentTriggerFireResult { + mode: 'fire'; + status: FireStatus; + conversationId: string; + streamId?: string; + generationCreatedAt?: number; +} + +export interface AgentTriggerSteerResult { + mode: 'steer'; + status: 'queued'; + conversationId: string; + steerId: string; + position: number; + preempt: boolean; + preemptRevision?: number; + replayed?: boolean; + settled?: boolean; + leftover?: boolean; +} + +export type AgentTriggerExecutionResult = AgentTriggerFireResult | AgentTriggerSteerResult; + +export interface AgentTriggerExecutionHostDeps { + /** Trusted root URL for this LibreChat server. */ + getBaseUrl: () => string; + /** Mint a short-lived token for the envelope's already-authenticated principal. */ + mintToken: (principal: AgentRunPrincipal, envelope: AgentTriggerEnvelope) => MaybePromise; + /** Optional user-timezone resolver for dynamic date variables in a new run. */ + getTimezone?: ( + principal: AgentRunPrincipal, + envelope: AgentFireTriggerEnvelope, + ) => MaybePromise; + fetch?: AgentTriggerFetch; + /** Total bound for setup, admission, and the bounded response read. */ + timeoutMs?: number; +} + +export interface AgentTriggerExecutionHost { + dispatch: ( + envelope: unknown, + options?: { signal?: AbortSignal }, + ) => Promise; +} + +interface AbortScope { + signal: AbortSignal; + abort: () => void; + cleanup: () => void; + timedOut: () => boolean; +} + +interface BoundedResponseBody { + text: string; + truncated: boolean; +} + +function executionError( + message: string, + options: AgentTriggerExecutionErrorOptions, +): AgentTriggerExecutionError { + return new AgentTriggerExecutionError(message, options); +} + +function requireTimeout(timeoutMs: number | undefined): number { + const value = timeoutMs ?? DEFAULT_FIRE_TIMEOUT_MS; + if (!Number.isSafeInteger(value) || value <= 0) { + throw new TypeError('Agent trigger timeout must be a positive integer'); + } + return value; +} + +function abortScope(parent: AbortSignal | undefined, timeoutMs: number): AbortScope { + const controller = new AbortController(); + let timeoutReached = false; + const onAbort = () => controller.abort(parent?.reason); + if (parent?.aborted === true) { + onAbort(); + } else { + parent?.addEventListener('abort', onAbort, { once: true }); + } + const timeout = setTimeout(() => { + timeoutReached = true; + controller.abort(); + }, timeoutMs); + return { + signal: controller.signal, + abort: () => controller.abort(), + timedOut: () => timeoutReached, + cleanup: () => { + clearTimeout(timeout); + parent?.removeEventListener('abort', onAbort); + }, + }; +} + +function abortError( + mode: 'fire' | 'steer', + scope: AbortScope, + parent: AbortSignal | undefined, + stage: string, + certainty: AgentTriggerFailureCertainty, +): AgentTriggerExecutionError { + const code = abortCode(scope, parent, 'ABORTED'); + return executionError( + `Agent trigger ${mode} was ${code === 'TIMEOUT' ? 'timed out' : 'aborted'} ${stage}`, + { + mode, + certainty, + retryable: true, + code, + }, + ); +} + +function observeAbort( + operation: () => MaybePromise, + mode: 'fire' | 'steer', + scope: AbortScope, + parent: AbortSignal | undefined, +): Promise { + if (scope.signal.aborted) { + return Promise.reject(abortError(mode, scope, parent, 'before dispatch', 'definite')); + } + return new Promise((resolve, reject) => { + const onAbort = () => { + scope.signal.removeEventListener('abort', onAbort); + reject(abortError(mode, scope, parent, 'during setup', 'definite')); + }; + scope.signal.addEventListener('abort', onAbort, { once: true }); + Promise.resolve() + .then(operation) + .then( + (value) => { + scope.signal.removeEventListener('abort', onAbort); + resolve(value); + }, + (error: unknown) => { + scope.signal.removeEventListener('abort', onAbort); + reject(error); + }, + ); + }); +} + +async function setupValue( + operation: () => MaybePromise, + mode: 'fire' | 'steer', + scope: AbortScope, + parent: AbortSignal | undefined, +): Promise { + try { + return await observeAbort(operation, mode, scope, parent); + } catch (error) { + if (error instanceof AgentTriggerExecutionError) { + throw error; + } + throw executionError( + `Agent trigger ${mode} setup failed: ${error instanceof Error ? error.message : String(error)}`, + { + mode, + certainty: 'definite', + retryable: true, + code: 'SETUP_FAILED', + }, + ); + } +} + +async function readResponseBody(response: Response): Promise { + if (response.body == null) { + return { text: '', truncated: false }; + } + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let size = 0; + const parts: string[] = []; + try { + while (true) { + const chunk = await reader.read(); + if (chunk.done) { + parts.push(decoder.decode()); + return { text: parts.join(''), truncated: false }; + } + const remaining = MAX_RESPONSE_BODY_BYTES - size; + if (chunk.value.byteLength > remaining) { + if (remaining > 0) { + parts.push(decoder.decode(chunk.value.subarray(0, remaining), { stream: true })); + } + parts.push(decoder.decode()); + await reader.cancel().catch(() => undefined); + return { text: parts.join(''), truncated: true }; + } + size += chunk.value.byteLength; + parts.push(decoder.decode(chunk.value, { stream: true })); + } + } finally { + reader.releaseLock(); + } +} + +function errorCode(value: unknown): string | undefined { + if (value == null || typeof value !== 'object' || !('code' in value)) { + return undefined; + } + const code = value.code; + return typeof code === 'string' && code.length > 0 ? code : undefined; +} + +function errorMessage(value: unknown): string | undefined { + if (value == null || typeof value !== 'object') { + return undefined; + } + const record = value as Record; + for (const key of ['error', 'message', 'code']) { + if (key in record) { + const message = record[key]; + if (typeof message === 'string' && message.length > 0) { + return message; + } + } + } + return undefined; +} + +function parseJson(value: string): unknown { + if (value.length === 0) { + return undefined; + } + try { + return JSON.parse(value) as unknown; + } catch { + return undefined; + } +} + +function fetchErrorCode(error: unknown): string | undefined { + const read = (value: unknown): string | undefined => { + if (value == null || typeof value !== 'object' || !('code' in value)) { + return undefined; + } + const code = value.code; + return typeof code === 'string' ? code : undefined; + }; + if (error == null || typeof error !== 'object') { + return undefined; + } + return ('cause' in error ? read(error.cause) : undefined) ?? read(error); +} + +function isDefiniteConnectFailure(error: unknown): boolean { + const code = fetchErrorCode(error); + return ( + code != null && + (PRE_CONNECT_ERROR_CODES.has(code) || code.startsWith('ERR_TLS') || code.includes('CERT')) + ); +} + +function isRetryableStatus(status: number): boolean { + return status === 401 || status === 408 || status === 425 || status === 429 || status >= 500; +} + +function isRetryableSteerRejection(status: number, code: string): boolean { + return code === 'RUN_PAUSED' || isRetryableStatus(status); +} + +function abortCode(scope: AbortScope, parent: AbortSignal | undefined, fallback: string): string { + if (scope.timedOut()) { + return 'TIMEOUT'; + } + if (parent?.aborted === true) { + return 'ABORTED'; + } + return fallback; +} + +function requireToken(value: unknown, mode: 'fire' | 'steer'): string { + if (typeof value !== 'string' || value.length === 0 || /\s/.test(value)) { + throw executionError('Agent trigger token mint returned an invalid token', { + mode, + certainty: 'definite', + retryable: false, + code: 'INVALID_TOKEN', + }); + } + return value; +} + +function triggerUrl(baseUrl: string, path: string, mode: 'fire' | 'steer'): string { + let url: URL; + try { + url = new URL(baseUrl); + } catch { + throw executionError('Agent trigger base URL is invalid', { + mode, + certainty: 'definite', + retryable: false, + code: 'INVALID_BASE_URL', + }); + } + if ((url.protocol !== 'http:' && url.protocol !== 'https:') || url.username || url.password) { + throw executionError('Agent trigger base URL must be an HTTP(S) URL without credentials', { + mode, + certainty: 'definite', + retryable: false, + code: 'INVALID_BASE_URL', + }); + } + url.pathname = `${url.pathname.replace(/\/+$/, '')}${path}`; + url.search = ''; + url.hash = ''; + return url.toString(); +} + +function fireUrl(baseUrl: string): string { + return triggerUrl(baseUrl, `/api/agents/chat/${EModelEndpoint.agents}`, 'fire'); +} + +function steerUrl(baseUrl: string): string { + return triggerUrl(baseUrl, '/api/agents/chat/steer/deliver', 'steer'); +} + +function requireString(value: unknown): string | undefined { + return typeof value === 'string' && value.length > 0 ? value : undefined; +} + +function requireSafeInteger(value: unknown): number | undefined { + return typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 ? value : undefined; +} + +function fireStatus(value: unknown): FireStatus | undefined { + return value === 'started' || value === 'resumed' || value === 'replaced' || value === 'settled' + ? value + : undefined; +} + +function parseFireResult(payload: unknown): AgentTriggerFireResult | undefined { + if (payload == null || typeof payload !== 'object') { + return undefined; + } + const status = fireStatus('status' in payload ? payload.status : undefined); + const conversationId = requireString( + 'conversationId' in payload ? payload.conversationId : undefined, + ); + if (status == null || conversationId == null) { + return undefined; + } + const streamId = requireString('streamId' in payload ? payload.streamId : undefined); + if (status !== 'settled' && streamId == null) { + return undefined; + } + const generationCreatedAt = requireSafeInteger( + 'generationCreatedAt' in payload ? payload.generationCreatedAt : undefined, + ); + return { + mode: 'fire', + status, + conversationId, + ...(streamId != null && { streamId }), + ...(generationCreatedAt != null && { generationCreatedAt }), + }; +} + +async function fire( + envelope: AgentFireTriggerEnvelope, + context: AgentTriggerDispatchContext, + deps: AgentTriggerExecutionHostDeps, + timeoutMs: number, +): Promise { + const scope = abortScope(context.signal, timeoutMs); + try { + const [token, timezone, baseUrl] = await Promise.all([ + setupValue( + () => deps.mintToken(envelope.principal, envelope), + 'fire', + scope, + context.signal, + ).then((value) => requireToken(value, 'fire')), + setupValue( + () => deps.getTimezone?.(envelope.principal, envelope), + 'fire', + scope, + context.signal, + ), + setupValue(() => deps.getBaseUrl(), 'fire', scope, context.signal), + ]).catch((error: unknown) => { + scope.abort(); + throw error; + }); + const url = fireUrl(baseUrl); + const fetcher: AgentTriggerFetch = deps.fetch ?? globalThis.fetch; + let response: Response; + try { + response = await fetcher(url, { + method: 'POST', + signal: scope.signal, + headers: { + Accept: 'application/json', + Authorization: `Bearer ${token}`, + 'Content-Type': 'application/json', + 'User-Agent': TRIGGER_USER_AGENT, + 'x-request-id': context.idempotencyKey, + [GENERATION_PROTOCOL_HEADER]: '2', + }, + body: JSON.stringify({ + text: envelope.input, + endpoint: EModelEndpoint.agents, + agent_id: envelope.target.agentId, + parentMessageId: Constants.NO_PARENT, + isContinued: false, + isRegenerate: false, + clientRequestId: context.idempotencyKey, + generationProtocolVersion: 2, + ...(typeof timezone === 'string' && timezone.trim().length > 0 + ? { timezone: timezone.trim() } + : {}), + }), + }); + } catch (error) { + const definite = isDefiniteConnectFailure(error); + const message = error instanceof Error ? error.message : String(error); + throw executionError( + `Agent trigger fire ${definite ? 'could not connect' : 'has an unknown outcome'}: ${message}`, + { + mode: 'fire', + certainty: definite ? 'definite' : 'ambiguous', + retryable: true, + code: abortCode(scope, context.signal, 'NETWORK_ERROR'), + }, + ); + } + + let boundedBody: BoundedResponseBody = { text: '', truncated: false }; + try { + boundedBody = await readResponseBody(response); + } catch (error) { + if (response.ok) { + throw executionError( + `Agent trigger fire response has an unknown outcome: ${ + error instanceof Error ? error.message : String(error) + }`, + { + mode: 'fire', + certainty: 'ambiguous', + retryable: true, + code: abortCode(scope, context.signal, 'INVALID_RESPONSE'), + status: response.status, + }, + ); + } + } + const payload = parseJson(boundedBody.text); + if (!response.ok) { + const message = + errorMessage(payload) ?? (boundedBody.text.slice(0, 300) || 'request rejected'); + throw executionError(`Agent trigger fire was rejected (${response.status}): ${message}`, { + mode: 'fire', + certainty: 'definite', + retryable: isRetryableStatus(response.status), + code: errorCode(payload) ?? 'FIRE_REJECTED', + status: response.status, + ...(response.headers.get('retry-after') != null && { + retryAfter: response.headers.get('retry-after') ?? undefined, + }), + }); + } + if (boundedBody.truncated) { + throw executionError('Agent trigger fire returned an oversized success response', { + mode: 'fire', + certainty: 'ambiguous', + retryable: true, + code: 'RESPONSE_TOO_LARGE', + status: response.status, + }); + } + if ( + payload != null && + typeof payload === 'object' && + 'status' in payload && + payload.status === 'aborted' + ) { + throw executionError('Agent trigger fire was aborted before generation started', { + mode: 'fire', + certainty: 'definite', + retryable: false, + code: 'START_ABORTED', + status: response.status, + }); + } + const result = parseFireResult(payload); + if (result == null) { + throw executionError('Agent trigger fire returned an invalid success response', { + mode: 'fire', + certainty: 'ambiguous', + retryable: true, + code: 'INVALID_RESPONSE', + status: response.status, + }); + } + return result; + } finally { + scope.cleanup(); + } +} + +function parseSteerResult(payload: unknown): AgentTriggerSteerResult | undefined { + if (payload == null || typeof payload !== 'object' || !('status' in payload)) { + return undefined; + } + if (payload.status !== 'queued') { + return undefined; + } + const conversationId = requireString( + 'conversationId' in payload ? payload.conversationId : undefined, + ); + const steerId = requireString('steerId' in payload ? payload.steerId : undefined); + const position = requireSafeInteger('position' in payload ? payload.position : undefined); + if ( + conversationId == null || + steerId == null || + position == null || + !('preempt' in payload) || + typeof payload.preempt !== 'boolean' + ) { + return undefined; + } + const preemptRevision = requireSafeInteger( + 'preemptRevision' in payload ? payload.preemptRevision : undefined, + ); + return { + mode: 'steer', + status: 'queued', + conversationId, + steerId, + position, + preempt: payload.preempt, + ...(preemptRevision != null && { preemptRevision }), + ...('replayed' in payload && + typeof payload.replayed === 'boolean' && { + replayed: payload.replayed, + }), + ...('settled' in payload && + typeof payload.settled === 'boolean' && { + settled: payload.settled, + }), + ...('leftover' in payload && + typeof payload.leftover === 'boolean' && { + leftover: payload.leftover, + }), + }; +} + +async function steer( + envelope: AgentSteerTriggerEnvelope, + context: AgentTriggerDispatchContext, + deps: AgentTriggerExecutionHostDeps, + timeoutMs: number, +): Promise { + const scope = abortScope(context.signal, timeoutMs); + try { + const [token, baseUrl] = await Promise.all([ + setupValue( + () => deps.mintToken(envelope.principal, envelope), + 'steer', + scope, + context.signal, + ).then((value) => requireToken(value, 'steer')), + setupValue(() => deps.getBaseUrl(), 'steer', scope, context.signal), + ]).catch((error: unknown) => { + scope.abort(); + throw error; + }); + const url = steerUrl(baseUrl); + const fetcher: AgentTriggerFetch = deps.fetch ?? globalThis.fetch; + let response: Response; + try { + response = await fetcher(url, { + method: 'POST', + signal: scope.signal, + headers: { + Accept: 'application/json', + Authorization: `Bearer ${token}`, + 'Content-Type': 'application/json', + 'User-Agent': TRIGGER_USER_AGENT, + 'x-request-id': context.idempotencyKey, + [GENERATION_PROTOCOL_HEADER]: '2', + }, + body: JSON.stringify({ + agentId: envelope.target.agentId, + conversationId: envelope.target.conversationId, + generationCreatedAt: envelope.target.generationCreatedAt, + text: envelope.input, + clientSteerId: context.idempotencyKey, + preempt: envelope.target.preempt === true, + generationProtocolVersion: 2, + }), + }); + } catch (error) { + const definite = isDefiniteConnectFailure(error); + throw executionError( + `Agent trigger steer ${definite ? 'could not connect' : 'has an unknown outcome'}: ${ + error instanceof Error ? error.message : String(error) + }`, + { + mode: 'steer', + certainty: definite ? 'definite' : 'ambiguous', + retryable: true, + code: abortCode(scope, context.signal, 'NETWORK_ERROR'), + }, + ); + } + + let boundedBody: BoundedResponseBody = { text: '', truncated: false }; + try { + boundedBody = await readResponseBody(response); + } catch (error) { + if (response.ok) { + throw executionError( + `Agent trigger steer response has an unknown outcome: ${ + error instanceof Error ? error.message : String(error) + }`, + { + mode: 'steer', + certainty: 'ambiguous', + retryable: true, + code: abortCode(scope, context.signal, 'INVALID_RESPONSE'), + status: response.status, + }, + ); + } + } + + const payload = parseJson(boundedBody.text); + if (response.status !== 202) { + const responseCode = errorCode(payload); + const routeUnavailable = response.status === 404 && responseCode == null; + const code = routeUnavailable + ? 'STEER_ADMISSION_UNAVAILABLE' + : (responseCode ?? 'STEER_REJECTED'); + throw executionError( + `Agent trigger steer was rejected (${response.status}): ${errorMessage(payload) ?? code}`, + { + mode: 'steer', + certainty: 'definite', + retryable: routeUnavailable || isRetryableSteerRejection(response.status, code), + code, + status: response.status, + ...(response.headers.get('retry-after') != null && { + retryAfter: response.headers.get('retry-after') ?? undefined, + }), + }, + ); + } + if (boundedBody.truncated) { + throw executionError('Agent trigger steer returned an oversized success response', { + mode: 'steer', + certainty: 'ambiguous', + retryable: true, + code: 'RESPONSE_TOO_LARGE', + status: response.status, + }); + } + if ( + payload == null || + typeof payload !== 'object' || + !('generationProtocolVersion' in payload) || + payload.generationProtocolVersion !== 2 + ) { + throw executionError('Agent trigger steer was accepted without a v2 receipt guarantee', { + mode: 'steer', + certainty: 'ambiguous', + retryable: false, + code: 'STEER_IDEMPOTENCY_UNAVAILABLE', + status: response.status, + }); + } + const parsed = parseSteerResult(payload); + if (parsed == null) { + throw executionError('Agent trigger steer returned an invalid success response', { + mode: 'steer', + certainty: 'ambiguous', + retryable: true, + code: 'INVALID_RESPONSE', + status: response.status, + }); + } + if (parsed.conversationId !== envelope.target.conversationId) { + throw executionError('Agent trigger steer returned a mismatched conversation', { + mode: 'steer', + certainty: 'ambiguous', + retryable: true, + code: 'INVALID_RESPONSE', + status: response.status, + }); + } + return parsed; + } finally { + scope.cleanup(); + } +} + +function runAsPrincipal( + envelope: AgentTriggerEnvelope, + context: AgentTriggerDispatchContext, + fn: () => Promise, +): Promise { + return tenantStorage.run( + { + userId: envelope.principal.userId, + tenantId: envelope.principal.tenantId, + requestId: context.idempotencyKey, + }, + fn, + ); +} + +/** + * Creates the trusted, source-neutral execution host. Schedules, webhooks, + * queues, MCP adapters, and internal events all call the same `dispatch` and + * therefore share authorization, idempotency, tenant, and failure semantics. + */ +export function createAgentTriggerExecutionHost( + deps: AgentTriggerExecutionHostDeps, +): AgentTriggerExecutionHost { + const timeoutMs = requireTimeout(deps.timeoutMs); + return { + dispatch: (envelope, options) => + Promise.resolve().then(() => + dispatchAgentTrigger( + envelope, + { + fire: (normalized, context) => + runAsPrincipal(normalized, context, () => fire(normalized, context, deps, timeoutMs)), + steer: (normalized, context) => + runAsPrincipal(normalized, context, () => + steer(normalized, context, deps, timeoutMs), + ), + }, + options, + ), + ), + }; +} diff --git a/packages/api/src/agents/triggers/index.ts b/packages/api/src/agents/triggers/index.ts new file mode 100644 index 0000000000..3c75f2ba09 --- /dev/null +++ b/packages/api/src/agents/triggers/index.ts @@ -0,0 +1,3 @@ +export * from './dispatch'; +export * from './envelope'; +export * from './host';