CERBERUS/LibreChat
1
0
Fork 0
mirror of https://github.com/danny-avila/LibreChat.git synced 2026-10-22 10:02:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

🛰️ feat: Execute Generic Agent Trigger Deliveries (#14921)

Browse source 
* feat: add generic agent trigger dispatch seam

* refactor: harden trigger dispatch contract

* fix: annotate envelope depth alias

* style: sort trigger envelope imports

* fix: reject unknown trigger dispatch modes

* fix: reject unknown trigger envelope versions

* refactor: validate complete trigger envelopes

* feat: add agent trigger execution host

* fix: enforce trigger delivery contracts

* fix: harden trigger admission path

* fix: finish trigger cancellation handling

* fix: retry strict steer rollout gaps

* fix: retry paused trigger steers

* fix: parallelize trigger admission setup
This commit is contained in:
Danny Avila 2026-08-17 08:45:49 -04:00 • committed by GitHub
parent 6b9fe97990
commit f8f118ef29
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 2465 additions and 107 deletions
Download patch file Download diff file Expand all files Collapse all files

42
api/server/controllers/agents/__tests__/steer.spec.js
View file

@ -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'));

43
api/server/controllers/agents/steer.js
View file

@ -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;

16
api/server/routes/agents/index.js
View file

@ -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,

118
packages/api/src/agents/envelope.ts
View file

@ -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<T>(value: T, path: string, ancestors: WeakSet<object>, depth: number): T;
function cloneJsonValue(
value: unknown,
path: string,
ancestors: WeakSet<object>,
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),
),
};
}

1
packages/api/src/agents/index.ts
View file

@ -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';

109
packages/api/src/agents/json.ts Normal file
View file

@ -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<T>(
value: T,
path: string,
createError: ErrorFactory,
ancestors: WeakSet<object> = 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);
}
}

101
packages/api/src/agents/steering/__tests__/request.spec.ts
View file

@ -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, {

22
packages/api/src/agents/steering/request.ts
View file

@ -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 }),

114
packages/api/src/agents/triggers/dispatch.spec.ts Normal file
View file

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

55
packages/api/src/agents/triggers/dispatch.ts Normal file
View file

@ -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<FireResult, SteerResult> {
fire: (
envelope: AgentFireTriggerEnvelope,
context: AgentTriggerDispatchContext,
) => Promise<FireResult>;
steer: (
envelope: AgentSteerTriggerEnvelope,
context: AgentTriggerDispatchContext,
) => Promise<SteerResult>;
}
/** Routes a normalized trigger without coupling its source to an execution transport. */
export function dispatchAgentTrigger<FireResult, SteerResult>(
envelope: unknown,
handlers: AgentTriggerDispatchHandlers<FireResult, SteerResult>,
options?: { signal?: AbortSignal },
): Promise<FireResult | SteerResult> {
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);
}

217
packages/api/src/agents/triggers/envelope.spec.ts Normal file
View file

@ -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');
});
});

296
packages/api/src/agents/triggers/envelope.ts Normal file
View file

@ -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<string, unknown> {
if (value == null || typeof value !== 'object' || Array.isArray(value)) {
throw error(`${path} must be an object`);
}
return value as Record<string, unknown>;
}
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}`;
}

624
packages/api/src/agents/triggers/host.spec.ts Normal file
View file

@ -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<AgentTriggerFetch> {
return jest.fn(implementation);
}
function deps(
fetcher: jest.MockedFunction<AgentTriggerFetch>,
overrides: Partial<AgentTriggerExecutionHostDeps> = {},
): AgentTriggerExecutionHostDeps {
return {
getBaseUrl: () => 'http://127.0.0.1:3080',
mintToken: () => 'signed-token',
fetch: fetcher,
...overrides,
};
}
function expectExecutionError(error: unknown, expected: Partial<AgentTriggerExecutionError>): 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<string>(() => 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<string>((resolve) => {
resolveToken = resolve;
});
const timezone = new Promise<string>((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<string>(() => 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<string>(() => 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,
});
});
});
});

811
packages/api/src/agents/triggers/host.ts Normal file
View file

@ -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> = T | Promise<T>;
type FireStatus = 'started' | 'resumed' | 'replaced' | 'settled';
export type AgentTriggerFetch = (input: string | URL, init?: RequestInit) => Promise<Response>;
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<string>;
/** Optional user-timezone resolver for dynamic date variables in a new run. */
getTimezone?: (
principal: AgentRunPrincipal,
envelope: AgentFireTriggerEnvelope,
) => MaybePromise<string | undefined>;
fetch?: AgentTriggerFetch;
/** Total bound for setup, admission, and the bounded response read. */
timeoutMs?: number;
}
export interface AgentTriggerExecutionHost {
dispatch: (
envelope: unknown,
options?: { signal?: AbortSignal },
) => Promise<AgentTriggerExecutionResult>;
}
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<T>(
operation: () => MaybePromise<T>,
mode: 'fire' | 'steer',
scope: AbortScope,
parent: AbortSignal | undefined,
): Promise<T> {
if (scope.signal.aborted) {
return Promise.reject(abortError(mode, scope, parent, 'before dispatch', 'definite'));
}
return new Promise<T>((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<T>(
operation: () => MaybePromise<T>,
mode: 'fire' | 'steer',
scope: AbortScope,
parent: AbortSignal | undefined,
): Promise<T> {
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<BoundedResponseBody> {
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<string, unknown>;
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<AgentTriggerFireResult> {
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<AgentTriggerSteerResult> {
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<T>(
envelope: AgentTriggerEnvelope,
context: AgentTriggerDispatchContext,
fn: () => Promise<T>,
): Promise<T> {
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,
),
),
};
}

3
packages/api/src/agents/triggers/index.ts Normal file
View file

@ -0,0 +1,3 @@
export * from './dispatch';
export * from './envelope';
export * from './host';