From 773127bff261886083b2fd595a3667fc4da52c15 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Fri, 28 Aug 2026 17:30:23 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=8E=A0=20refactor:=20Route=20Every=20Even?= =?UTF-8?q?t=20Actor=20Turn=20Through=20One=20Lifecycle=20(#15325)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * refactor: unify Event Actor turn lifecycle * fix: retain Event Actor fence ownership * fix: preserve mixed-version actor suspension safety --- CONTEXT.md | 2 +- .../__tests__/request.resumeMetadata.spec.js | 49 +++- .../agents/__tests__/resume.spec.js | 1 + api/server/controllers/agents/request.js | 242 ++++++++---------- api/server/controllers/agents/resume.js | 16 +- packages/api/src/agents/plan.spec.ts | 2 +- packages/api/src/agents/plan.ts | 4 +- packages/api/src/agents/triggers/index.ts | 1 + packages/api/src/agents/triggers/turn.spec.ts | 192 ++++++++++++++ packages/api/src/agents/triggers/turn.ts | 187 ++++++++++++++ 10 files changed, 546 insertions(+), 150 deletions(-) create mode 100644 packages/api/src/agents/triggers/turn.spec.ts create mode 100644 packages/api/src/agents/triggers/turn.ts diff --git a/CONTEXT.md b/CONTEXT.md index 965b42b79f..2d6b137ad4 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -12,7 +12,7 @@ - **Agent event handling outcome**: the durable, generation-fenced result of a previously accepted event delivery. `started` proves generation admission; terminal states distinguish verified tool application, clean completion without action, failure, and cancellation. Transport success remains separate so an accepted event cannot masquerade as completed work. - **Agent event expected action**: an optional source-declared tool name and bounded argument subset evaluated against host-observed completed run steps. It is evidence policy, not authorization and not a model-authored success claim. - **Event actor head**: the private, durable pointer on an event-bound child conversation to its latest committed LangGraph checkpoint, plus one previous checkpoint for safe cleanup. Only a qualifying applied action advances it through compare-and-swap; failed, cancelled, or no-action invocations leave it unchanged. A legacy-path event marks the head for a cold rebuild from durable message history before fork mode can resume. Every applied commit conflict, unverified commit, or post-commit persistence failure is retained in a private reconciliation journal that blocks later actor turns instead of continuing from stale state; an exact marker can be cleared only after its checkpoint is verified authoritative, its history is repaired, or its external action is explicitly compensated. -- **Event actor invocation fork**: a delivery-owned checkpoint namespace copied from the event actor head. A warm invocation receives only the new trusted event, then commits its terminal checkpoint when the expected action is observed or deletes the fork otherwise. When the invocation pauses for approval or Ask User, the SDK emits signed, versioned suspension evidence. The child Conversation is the canonical one-shot suspension authority; the generation job carries only a versioned projection for UI, rolling-deploy routing, and the existing resume endpoint. Current Event Actor hosts select generation protocol v2 automatically; immutable protocol-v1 jobs remain consumable through the legacy history path while they drain. A resume shares one identity between its Conversation claim and provider-owner CAS, clears the predecessor projection, and publishes a successor only after a re-pause is canonical. A pending interrupt takes precedence over expected-action evidence from the same segment; if that segment already applied the expected action, publishing its successor pause cold-marks the prior head until a later applied commit replaces it. An ambiguous projection write is accepted only after reading back the exact generation, action, and suspension. The provider-start CAS is written only after client reconstruction and immediately before the continuation gate opens, then retains its exact execution identity after drain, so terminal recovery can compensate a projected claim only when that identity proves execution never began. Durable approval projection is exposed before the persistence barrier opens, preventing a resolved action from being announced afterward. Terminal no-action retirement cancels or settles the exact suspension and releases its delivery-side action admission before public settlement; if retention already removed the child Conversation, the delivery remains authoritative for its exact admission identity. Resume, re-pause, cancellation, and expiry claim or replace that exact suspension before touching its job projection, so later mailbox deliveries stay blocked until terminal history and handling evidence settle. +- **Event actor invocation fork**: a delivery-owned checkpoint namespace copied from the event actor head. Every bound Event Actor enters one turn module; `fresh`, `history`, and `checkpoint` are internal state-loading adapters selected automatically from actor state and immutable request capability, never from operator configuration. The history adapter owns its durable turn fence and token ordering, while immutable protocol-v1 tokens remain read-compatible until their jobs drain. A warm invocation receives only the new trusted event, then commits its terminal checkpoint when the expected action is observed or deletes the fork otherwise. When the invocation pauses for approval or Ask User, the SDK emits signed, versioned suspension evidence. The child Conversation is the canonical one-shot suspension authority; the generation job carries only a versioned projection for UI, rolling-deploy routing, and the existing resume endpoint. Current Event Actor hosts select generation protocol v2 automatically; a trusted pre-capability producer remains on history only for the mixed-version drain. A resume shares one identity between its Conversation claim and provider-owner CAS, clears the predecessor projection, and publishes a successor only after a re-pause is canonical. A pending interrupt takes precedence over expected-action evidence from the same segment; if that segment already applied the expected action, publishing its successor pause cold-marks the prior head until a later applied commit replaces it. An ambiguous projection write is accepted only after reading back the exact generation, action, and suspension. The provider-start CAS is written only after client reconstruction and immediately before the continuation gate opens, then retains its exact execution identity after drain, so terminal recovery can compensate a projected claim only when that identity proves execution never began. Durable approval projection is exposed before the persistence barrier opens, preventing a resolved action from being announced afterward. Terminal no-action retirement cancels or settles the exact suspension and releases its delivery-side action admission before public settlement; if retention already removed the child Conversation, the delivery remains authoritative for its exact admission identity. Resume, re-pause, cancellation, and expiry claim or replace that exact suspension before touching its job projection, so later mailbox deliveries stay blocked until terminal history and handling evidence settle. - **Event actor receipt**: the private, terminal proof stored on the authoritative `AgentTriggerDelivery` row for one bound actor invocation. Its unique delivery identity, terminal resolution, exact checkpoint, and bounded action identity provide replay and recovery for the retention window without storing prompts, events, tool arguments, tool output, or conversation history. It does not own the actor checkpoint; the conversation keeps only the actor head and any active unresolved reconciliation until this receipt is durable. - **Agent event actor mailbox**: the automatic durable delivery-ordering lane for one authenticated source binding. It keeps later deliveries queued after transport admission until the current child turn records an authoritative terminal handling outcome. It serializes existing coalesced batches and individual events without becoming a second execution controller or actor checkpoint store. - **Agent trigger capability shield**: the durable mixed-version representation for internal trigger work that only a capability-aware worker may execute. Mongo uses an old-publishable `staging` shell; a queued `leased` shell without an owner or deadline, which old workers cannot claim but can use for bounded lane rechecks; a private lease only during execution; and a legacy-terminal `capability_dead` shell once dead. Private capability fields own current claiming, retry, and dead-letter truth. Redis uses a versioned fail-closed terminal status and recovery index that old replacement scripts and sweepers cannot consume. The shield is an implementation detail at the storage seam, never a deployment switch or user-configured product mode. diff --git a/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js b/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js index ae4bb202b8..d8bbb55b26 100644 --- a/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js +++ b/api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js @@ -78,6 +78,48 @@ const mockIsScheduleLive = jest.fn(); const mockDeleteAgentCheckpoint = jest.fn(); const mockExecuteAgentEventActor = jest.fn(); const mockResumeAgentEventActor = jest.fn(); +const mockCreateAgentEventActorTurn = jest.fn((input, dependencies) => { + let historyToken; + let historyOwner; + return { + run: async () => { + if (input.strategy === 'checkpoint') { + const result = + input.checkpoint.kind === 'resume' + ? await mockResumeAgentEventActor(input.checkpoint.input, dependencies.actor) + : await mockExecuteAgentEventActor(input.checkpoint.input, dependencies.actor); + return { adapter: 'checkpoint', ...result }; + } + const token = 'event-actor-history-token'; + const acquired = await dependencies.history.begin({ ...input.history.owner, token }); + if (!acquired) { + throw Object.assign(new Error('The event actor is temporarily unavailable'), { + code: 'EVENT_ACTOR_NOT_READY', + status: 409, + }); + } + historyToken = token; + historyOwner = input.history.owner; + try { + await input.history.persistToken(token); + } catch (error) { + historyToken = undefined; + historyOwner = undefined; + await dependencies.history.complete({ ...input.history.owner, token }); + throw error; + } + return { adapter: 'history', value: await input.history.invoke() }; + }, + historyPersisted: async () => { + if (historyToken == null) { + return; + } + const token = historyToken; + historyToken = undefined; + await dependencies.history.complete({ ...historyOwner, token }); + }, + }; +}); const mockCreateAgentEventActorDetachedActionLifecycle = jest.fn(() => undefined); const mockFindAgentEventAppliedAction = jest.fn(); const mockResolveAgentTurnExecutionPlan = jest.fn((input) => { @@ -265,8 +307,7 @@ jest.mock('@librechat/api', () => ({ return messages.length === 0; }, deleteAgentCheckpoint: (...args) => mockDeleteAgentCheckpoint(...args), - executeAgentEventActor: (...args) => mockExecuteAgentEventActor(...args), - resumeAgentEventActor: (...args) => mockResumeAgentEventActor(...args), + createAgentEventActorTurn: (...args) => mockCreateAgentEventActorTurn(...args), createAgentEventActorDetachedActionLifecycle: (...args) => mockCreateAgentEventActorDetachedActionLifecycle(...args), parseAgentEventActorDetachedCompletion: (value) => (value?.version === 1 ? value : undefined), @@ -4664,7 +4705,7 @@ describe('ResumableAgentController resume metadata', () => { undefined, ], [ - 'pre-cutover pause-capable fleet', + 'pre-capability pause consumer fleet', { toolDefinitions: [] }, { toolApproval: { enabled: true } }, undefined, @@ -4706,7 +4747,7 @@ describe('ResumableAgentController resume metadata', () => { }), }; const shouldCheckpoint = - _label !== 'memory-checkpointer' && _label !== 'pre-cutover pause-capable fleet'; + _label !== 'memory-checkpointer' && _label !== 'pre-capability pause consumer fleet'; if (shouldCheckpoint) { mockExecuteAgentEventActor.mockImplementationOnce(async (input) => { await input.invoke({ diff --git a/api/server/controllers/agents/__tests__/resume.spec.js b/api/server/controllers/agents/__tests__/resume.spec.js index 5f236036ab..d399cbd670 100644 --- a/api/server/controllers/agents/__tests__/resume.spec.js +++ b/api/server/controllers/agents/__tests__/resume.spec.js @@ -140,6 +140,7 @@ jest.mock('@librechat/api', () => ({ getAgentCheckpointer: (...args) => mockGetAgentCheckpointer(...args), checkAccess: (...args) => mockCheckAccess(...args), resumeAgentEventActor: (...args) => mockResumeAgentEventActor(...args), + settleAgentEventActorHistoryTurn: (input, complete) => complete(input), createAgentEventActorDetachedActionLifecycle: (...args) => mockCreateAgentEventActorDetachedActionLifecycle(...args), })); diff --git a/api/server/controllers/agents/request.js b/api/server/controllers/agents/request.js index 81187ea1f7..1941594534 100644 --- a/api/server/controllers/agents/request.js +++ b/api/server/controllers/agents/request.js @@ -30,8 +30,7 @@ const { getAttachmentTitleText, createMCPRuntimeRequestBody, isAgentEventRetentionActive, - executeAgentEventActor, - resumeAgentEventActor, + createAgentEventActorTurn, createAgentEventActorDetachedActionLifecycle, parseAgentEventActorDetachedCompletion, EVENT_ACTOR_DETACHED_COMPLETION_SOURCE, @@ -1747,9 +1746,9 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit } : undefined, canPause: eventActorMayPause, - /** Protocol v2 is the existing homogeneous-fleet cutover. Keeping - * pause-capable producers on the legacy path under v1 prevents an old - * `/resume` replica from consuming a signed suspension it cannot claim. */ + /** Old trusted producers can coexist during a direct rolling upgrade. + * Their immutable v1 request keeps pause-capable turns on the history + * adapter until every consumer understands durable suspensions. */ durableEventActorSuspensions: generationProtocolVersion >= GENERATION_PROTOCOL_V2, checkpointerType: agentsConfig?.checkpointer?.type, }); @@ -1923,36 +1922,7 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit const eventActorTenantId = req._agentEventBindingTenantId; let appliedEventActor; let eventActorPersistenceComplete = false; - let legacyEventActorTurnToken; - /** Closes the durable legacy-turn fence once this turn's history is - * persisted, clearing the token and advancing the epoch in one write. A - * failure here leaves the token SET, which keeps blocking forks until an - * explicit reconciliation proves the ambiguous turn safe to release. */ - const sealLegacyEventActorTurn = async () => { - if (legacyEventActorTurnToken == null) { - return; - } - const token = legacyEventActorTurnToken; - legacyEventActorTurnToken = undefined; - try { - const sealed = await completeAgentEventActorLegacyTurn({ - user: userId, - conversationId, - ...(eventActorTenantId == null ? {} : { tenantId: eventActorTenantId }), - token, - }); - if (!sealed) { - logger.error( - `[event-actor] Legacy turn fence ${token} was not sealed; forks stay blocked pending reconciliation`, - ); - } - } catch (sealError) { - logger.error( - `[event-actor] Failed to seal legacy turn fence ${token}; forks stay blocked pending reconciliation`, - sealError, - ); - } - }; + let eventActorTurn; const recordEventActorPersistenceFailure = async (error) => { if (appliedEventActor == null || eventActorPersistenceComplete) { return; @@ -2191,73 +2161,108 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit getReceipt: getAgentEventActorReceipt, clearReconciliation: clearAgentEventActorReconciliation, }; - const sendPromise = usesCheckpointStrategy - ? (isInternalDetachedCompletion - ? resumeAgentEventActor( - { + let checkpointTurn; + if (usesCheckpointStrategy && isInternalDetachedCompletion) { + checkpointTurn = { + kind: 'resume', + input: { + user: userId, + ...(eventActorTenantId == null ? {} : { tenantId: eventActorTenantId }), + conversationId, + bindingId: turnExecutionPlan.binding.bindingId, + suspension: internalDetachedSuspension.suspension, + resumeAttemptId: clientRequestId, + resumeValue: { + type: EVENT_ACTOR_DETACHED_COMPLETION_TYPE, + taskId: internalDetachedAction.taskId, + status: internalDetachedAction.status, + ...(internalDetachedAction.result == null + ? {} + : { result: internalDetachedAction.result }), + ...(internalDetachedAction.error == null + ? {} + : { error: internalDetachedAction.error }), + }, + signal: job.abortController.signal, + checkpointer: req.config?.endpoints?.[EModelEndpoint.agents]?.checkpointer, + expectedAction: turnExecutionPlan.expectedAction, + resume: async (actorContext) => { + client.checkpointNamespace = actorContext.checkpointNamespace; + client.eventActorCheckpointId = actorContext.checkpointId; + client.eventActorInvocationId = actorContext.invocationId; + client.eventActorContinuation = actorContext.continuation; + return client.sendMessage(text, messageOptions); + }, + readAppliedAction: readAppliedEventAction, + readSuspension: () => + eventActorDetachedAction?.readSuspension() ?? client.readEventActorSuspension(), + readResultContext: () => client.getEventActorContext(), + }, + }; + } else if (usesCheckpointStrategy) { + checkpointTurn = { + kind: 'execute', + input: { + user: userId, + ...(eventActorTenantId == null ? {} : { tenantId: eventActorTenantId }), + conversationId, + bindingId: turnExecutionPlan.binding.bindingId, + invocationId: actorInvocationId, + event: agentEventDelivery.event, + expectedAction: turnExecutionPlan.expectedAction, + signal: job.abortController.signal, + checkpointer: req.config?.endpoints?.[EModelEndpoint.agents]?.checkpointer, + resolveContext: (state) => client.prepareEventActorContext(state), + readResultContext: () => client.getEventActorContext(), + invoke: async (actorContext) => { + client.checkpointNamespace = actorContext.checkpointNamespace; + client.eventActorCheckpointId = actorContext.checkpointId; + client.eventActorInvocationId = actorContext.invocationId; + client.eventActorContinuation = actorContext.continuation; + return client.sendMessage(text, messageOptions); + }, + readAppliedAction: readAppliedEventAction, + readSuspension: () => + eventActorDetachedAction?.readSuspension() ?? client.readEventActorSuspension(), + }, + }; + } + const isBoundEventActor = + agentEventDelivery?.event != null && req._agentEventBindingParentConversationId != null; + eventActorTurn = isBoundEventActor + ? createAgentEventActorTurn( + { + strategy: turnExecutionPlan.strategy, + ...(checkpointTurn == null ? {} : { checkpoint: checkpointTurn }), + history: { + owner: { user: userId, - ...(eventActorTenantId == null ? {} : { tenantId: eventActorTenantId }), conversationId, - bindingId: turnExecutionPlan.binding.bindingId, - suspension: internalDetachedSuspension.suspension, - resumeAttemptId: clientRequestId, - resumeValue: { - type: EVENT_ACTOR_DETACHED_COMPLETION_TYPE, - taskId: internalDetachedAction.taskId, - status: internalDetachedAction.status, - ...(internalDetachedAction.result == null - ? {} - : { result: internalDetachedAction.result }), - ...(internalDetachedAction.error == null - ? {} - : { error: internalDetachedAction.error }), - }, - signal: job.abortController.signal, - checkpointer: req.config?.endpoints?.[EModelEndpoint.agents]?.checkpointer, - expectedAction: turnExecutionPlan.expectedAction, - resume: async (actorContext) => { - client.checkpointNamespace = actorContext.checkpointNamespace; - client.eventActorCheckpointId = actorContext.checkpointId; - client.eventActorInvocationId = actorContext.invocationId; - client.eventActorContinuation = actorContext.continuation; - return client.sendMessage(text, messageOptions); - }, - readAppliedAction: readAppliedEventAction, - readSuspension: () => - eventActorDetachedAction?.readSuspension() ?? - client.readEventActorSuspension(), - readResultContext: () => client.getEventActorContext(), - }, - actorDependencies, - ) - : executeAgentEventActor( - { - user: userId, ...(eventActorTenantId == null ? {} : { tenantId: eventActorTenantId }), - conversationId, - bindingId: turnExecutionPlan.binding.bindingId, - invocationId: actorInvocationId, - event: agentEventDelivery.event, - expectedAction: turnExecutionPlan.expectedAction, - signal: job.abortController.signal, - checkpointer: req.config?.endpoints?.[EModelEndpoint.agents]?.checkpointer, - resolveContext: (state) => client.prepareEventActorContext(state), - readResultContext: () => client.getEventActorContext(), - invoke: async (actorContext) => { - client.checkpointNamespace = actorContext.checkpointNamespace; - client.eventActorCheckpointId = actorContext.checkpointId; - client.eventActorInvocationId = actorContext.invocationId; - client.eventActorContinuation = actorContext.continuation; - return client.sendMessage(text, messageOptions); - }, - readAppliedAction: readAppliedEventAction, - readSuspension: () => - eventActorDetachedAction?.readSuspension() ?? - client.readEventActorSuspension(), }, - actorDependencies, - ) - ).then(async ({ value, execution }) => { + persistToken: (token) => + GenerationJobManager.updateMetadata( + streamId, + { agentEventLegacyTurnToken: token }, + jobCreatedAt, + ), + invoke: () => client.sendMessage(text, messageOptions), + }, + }, + { + actor: actorDependencies, + history: { + begin: beginAgentEventActorLegacyTurn, + complete: completeAgentEventActorLegacyTurn, + }, + }, + ) + : undefined; + const sendPromise = eventActorTurn + ? eventActorTurn.run().then(async ({ adapter, value, execution }) => { + if (adapter !== 'checkpoint') { + return value; + } if (execution.status === 'applied') { appliedEventActor = { invocationId: actorInvocationId, @@ -2291,42 +2296,7 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit }); return value; }) - : (async () => { - if ( - agentEventDelivery?.event != null && - req._agentEventBindingParentConversationId != null - ) { - const token = crypto.randomUUID(); - /** Open the fence BEFORE execution so a fork can neither run - * nor commit while this turn's messages are not yet durable. */ - const legacyTurnOwner = { - user: userId, - conversationId, - ...(eventActorTenantId == null ? {} : { tenantId: eventActorTenantId }), - }; - const acquiredLegacyTurn = await beginAgentEventActorLegacyTurn({ - ...legacyTurnOwner, - token, - }); - if (!acquiredLegacyTurn) { - throw Object.assign(new Error('The event actor is temporarily unavailable'), { - code: 'EVENT_ACTOR_NOT_READY', - status: 409, - }); - } - /** The same exact token must survive a HITL pause. Retain local - * ownership before the Redis write so a metadata failure can - * still seal this token after its durable error turn is saved; - * provider execution starts only after metadata persists. */ - legacyEventActorTurnToken = token; - await GenerationJobManager.updateMetadata( - streamId, - { agentEventLegacyTurnToken: token }, - jobCreatedAt, - ); - } - return client.sendMessage(text, messageOptions); - })(); + : client.sendMessage(text, messageOptions); if (titleEligible && titleTiming === 'immediate') { immediateTitlePromise = addTitle(req, { @@ -2670,7 +2640,7 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit throw new Error('Applied event actor history barrier could not be durably recorded'); } } - await sealLegacyEventActorTurn(); + await eventActorTurn?.historyPersisted(); eventActorPersistenceComplete = true; // If the user stopped this turn — or an empty preempt boundary truncated @@ -2900,7 +2870,7 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit * fails. Time cannot prove whether an external action occurred, so an * ambiguous fence remains fail-closed pending explicit reconciliation. */ if (legacyEventActorErrorHistoryDurable) { - await sealLegacyEventActorTurn(); + await eventActorTurn?.historyPersisted(); } if (ownsScheduledFailure && !scheduleTerminalOutcomeRecorded) { diff --git a/api/server/controllers/agents/resume.js b/api/server/controllers/agents/resume.js index bdc1abfca2..00c32ff777 100644 --- a/api/server/controllers/agents/resume.js +++ b/api/server/controllers/agents/resume.js @@ -41,6 +41,7 @@ const { getSafeErrorMetadata, isAgentEventRetentionActive, resumeAgentEventActor, + settleAgentEventActorHistoryTurn, createAgentEventActionRecorder, createAgentEventActorDetachedActionLifecycle, findAgentEventAppliedAction, @@ -154,12 +155,15 @@ async function sealResumedLegacyEventActorTurn({ userId, conversationId, metadat return; } try { - const sealed = await completeAgentEventActorLegacyTurn({ - user: userId, - conversationId, - ...(metadata?.tenantId == null ? {} : { tenantId: metadata.tenantId }), - token, - }); + const sealed = await settleAgentEventActorHistoryTurn( + { + user: userId, + conversationId, + ...(metadata?.tenantId == null ? {} : { tenantId: metadata.tenantId }), + token, + }, + completeAgentEventActorLegacyTurn, + ); if (!sealed) { logger.error( `[event-actor] Resumed legacy turn fence ${token} was not sealed; forks stay blocked until bounded reclaim`, diff --git a/packages/api/src/agents/plan.spec.ts b/packages/api/src/agents/plan.spec.ts index dff85c32aa..699d94dae0 100644 --- a/packages/api/src/agents/plan.spec.ts +++ b/packages/api/src/agents/plan.spec.ts @@ -67,7 +67,7 @@ describe('resolveAgentTurnExecutionPlan', () => { ['no expected action', { event: { ...boundEvent(), expectedAction: undefined } }], ['memory checkpointer', { event: boundEvent(), checkpointerType: 'memory' }], [ - 'pre-cutover pause consumer fleet', + 'pre-capability pause consumer fleet', { event: boundEvent(), canPause: true, durableEventActorSuspensions: false }, ], ] as const)('falls back to history for %s', (_label, overrides) => { diff --git a/packages/api/src/agents/plan.ts b/packages/api/src/agents/plan.ts index ff47d60c4d..bafef4c823 100644 --- a/packages/api/src/agents/plan.ts +++ b/packages/api/src/agents/plan.ts @@ -34,8 +34,8 @@ export interface ResolveAgentTurnExecutionPlanInput { expectedAction?: AgentTriggerExpectedAction; }; canPause: boolean; - /** Fleet-wide producer gate. Shared-store deployments enable this only - * after every resume consumer understands signed Event Actor suspensions. */ + /** Immutable request capability used only during mixed-version drain. This + * is negotiated automatically and is never an operator-selected runtime. */ durableEventActorSuspensions: boolean; checkpointerType?: TCheckpointerType; } diff --git a/packages/api/src/agents/triggers/index.ts b/packages/api/src/agents/triggers/index.ts index 742e8ac79c..6a17e7b6e3 100644 --- a/packages/api/src/agents/triggers/index.ts +++ b/packages/api/src/agents/triggers/index.ts @@ -11,3 +11,4 @@ export * from './lease'; export * from './outcome'; export * from './service'; export * from './engine'; +export * from './turn'; diff --git a/packages/api/src/agents/triggers/turn.spec.ts b/packages/api/src/agents/triggers/turn.spec.ts new file mode 100644 index 0000000000..d4d64c2181 --- /dev/null +++ b/packages/api/src/agents/triggers/turn.spec.ts @@ -0,0 +1,192 @@ +const mockExecuteAgentEventActor = jest.fn(); +const mockResumeAgentEventActor = jest.fn(); + +jest.mock('./actor', () => ({ + executeAgentEventActor: (...args: Parameters) => + mockExecuteAgentEventActor(...args), + resumeAgentEventActor: (...args: Parameters) => + mockResumeAgentEventActor(...args), +})); + +import type { + AgentEventActorDependencies, + ExecuteAgentEventActorInput, + ResumeAgentEventActorInput, +} from './actor'; +import { createAgentEventActorTurn, settleAgentEventActorHistoryTurn } from './turn'; + +const owner = { + user: 'user-1', + tenantId: 'tenant-1', + conversationId: 'actor-1', +}; + +describe('Event Actor turn module', () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + it('keeps history token ordering behind one turn interface', async () => { + const order: string[] = []; + const begin = jest.fn(async () => { + order.push('begin'); + return true; + }); + const persistToken = jest.fn(async () => { + order.push('persist'); + }); + const invoke = jest.fn(async () => { + order.push('invoke'); + return 'history-result'; + }); + const complete = jest.fn(async () => { + order.push('complete'); + return true; + }); + const turn = createAgentEventActorTurn( + { strategy: 'history', history: { owner, persistToken, invoke } }, + { history: { begin, complete } }, + ); + + await expect(turn.run()).resolves.toEqual({ + adapter: 'history', + value: 'history-result', + }); + expect(order).toEqual(['begin', 'persist', 'invoke']); + + await turn.historyPersisted(); + await turn.historyPersisted(); + + expect(order).toEqual(['begin', 'persist', 'invoke', 'complete']); + expect(complete).toHaveBeenCalledTimes(1); + expect(complete).toHaveBeenCalledWith({ + ...owner, + token: expect.any(String), + }); + }); + + it('releases an acquired history fence when token persistence fails before invocation', async () => { + const begin = jest.fn().mockResolvedValue(true); + const complete = jest.fn().mockResolvedValue(true); + const invoke = jest.fn(); + const turn = createAgentEventActorTurn( + { + strategy: 'history', + history: { + owner, + persistToken: jest.fn().mockRejectedValue(new Error('metadata unavailable')), + invoke, + }, + }, + { history: { begin, complete } }, + ); + + await expect(turn.run()).rejects.toThrow('metadata unavailable'); + expect(invoke).not.toHaveBeenCalled(); + expect(complete).toHaveBeenCalledWith({ + ...owner, + token: expect.any(String), + }); + }); + + it('retains an unstarted history fence until transient release failure can be retried', async () => { + const complete = jest + .fn() + .mockRejectedValueOnce(new Error('conversation store unavailable')) + .mockResolvedValueOnce(true); + const turn = createAgentEventActorTurn( + { + strategy: 'history', + history: { + owner, + persistToken: jest.fn().mockRejectedValue(new Error('metadata unavailable')), + invoke: jest.fn(), + }, + }, + { history: { begin: jest.fn().mockResolvedValue(true), complete } }, + ); + + await expect(turn.run()).rejects.toThrow('metadata unavailable'); + await turn.historyPersisted(); + + expect(complete).toHaveBeenCalledTimes(2); + expect(complete.mock.calls[1][0]).toEqual(complete.mock.calls[0][0]); + }); + + it('retains an invoked history fence until durable history is acknowledged', async () => { + const complete = jest.fn().mockResolvedValue(true); + const turn = createAgentEventActorTurn( + { + strategy: 'fresh', + history: { + owner, + persistToken: jest.fn().mockResolvedValue(undefined), + invoke: jest.fn().mockRejectedValue(new Error('provider failed')), + }, + }, + { history: { begin: jest.fn().mockResolvedValue(true), complete } }, + ); + + await expect(turn.run()).rejects.toThrow('provider failed'); + expect(complete).not.toHaveBeenCalled(); + + await turn.historyPersisted(); + expect(complete).toHaveBeenCalledTimes(1); + }); + + it('delegates checkpoint execution without exposing history state', async () => { + const input = { + invocationId: 'delivery-1', + } as ExecuteAgentEventActorInput; + const actor = {} as AgentEventActorDependencies; + mockExecuteAgentEventActor.mockResolvedValue({ + value: 'checkpoint-result', + execution: { status: 'completed' }, + }); + const turn = createAgentEventActorTurn( + { strategy: 'checkpoint', checkpoint: { kind: 'execute', input } }, + { actor }, + ); + + await expect(turn.run()).resolves.toEqual({ + adapter: 'checkpoint', + value: 'checkpoint-result', + execution: { status: 'completed' }, + }); + expect(mockExecuteAgentEventActor).toHaveBeenCalledWith(input, actor); + await expect(turn.historyPersisted()).resolves.toBeUndefined(); + }); + + it('delegates checkpoint resume through the same interface', async () => { + const input = { + resumeAttemptId: 'resume-1', + } as ResumeAgentEventActorInput; + const actor = {} as AgentEventActorDependencies; + mockResumeAgentEventActor.mockResolvedValue({ + value: 'resume-result', + execution: { status: 'applied' }, + }); + const turn = createAgentEventActorTurn( + { strategy: 'checkpoint', checkpoint: { kind: 'resume', input } }, + { actor }, + ); + + await expect(turn.run()).resolves.toEqual({ + adapter: 'checkpoint', + value: 'resume-result', + execution: { status: 'applied' }, + }); + expect(mockResumeAgentEventActor).toHaveBeenCalledWith(input, actor); + }); + + it('settles an historical token without exposing its storage vocabulary to callers', async () => { + const complete = jest.fn().mockResolvedValue(true); + await expect( + settleAgentEventActorHistoryTurn({ ...owner, token: 'historical-token' }, complete), + ).resolves.toBe(true); + expect(complete).toHaveBeenCalledWith({ + ...owner, + token: 'historical-token', + }); + }); +}); diff --git a/packages/api/src/agents/triggers/turn.ts b/packages/api/src/agents/triggers/turn.ts new file mode 100644 index 0000000000..aeab25e593 --- /dev/null +++ b/packages/api/src/agents/triggers/turn.ts @@ -0,0 +1,187 @@ +import { randomUUID } from 'node:crypto'; +import { logger } from '@librechat/data-schemas'; +import type { ConversationMethods } from '@librechat/data-schemas'; +import type { + AgentEventActorDependencies, + ExecuteAgentEventActorInput, + ExecuteAgentEventActorResult, + ResumeAgentEventActorInput, +} from './actor'; +import type { AgentTurnContinuationStrategy } from '../plan'; +import { executeAgentEventActor, resumeAgentEventActor } from './actor'; + +export interface AgentEventActorTurnOwner { + user: string; + tenantId?: string; + conversationId: string; +} + +export type AgentEventActorCheckpointTurn = + | { kind: 'execute'; input: ExecuteAgentEventActorInput } + | { kind: 'resume'; input: ResumeAgentEventActorInput }; + +export interface AgentEventActorHistoryTurn { + owner: AgentEventActorTurnOwner; + persistToken(token: string): Promise; + invoke(): Promise; +} + +export interface CreateAgentEventActorTurnInput { + strategy: AgentTurnContinuationStrategy; + checkpoint?: AgentEventActorCheckpointTurn; + history?: AgentEventActorHistoryTurn; +} + +export interface AgentEventActorTurnDependencies { + actor?: AgentEventActorDependencies; + history?: { + begin: ConversationMethods['beginAgentEventActorLegacyTurn']; + complete: ConversationMethods['completeAgentEventActorLegacyTurn']; + }; +} + +export type AgentEventActorTurnResult = + | ({ adapter: 'checkpoint' } & ExecuteAgentEventActorResult) + | { adapter: 'history'; value: T }; + +export interface AgentEventActorTurn { + run(): Promise>; + historyPersisted(): Promise; +} + +type EventActorTurnError = Error & { code: string; status: number }; + +function requireCheckpoint( + input: CreateAgentEventActorTurnInput, + deps: AgentEventActorTurnDependencies, +): { + turn: AgentEventActorCheckpointTurn; + actor: AgentEventActorDependencies; +} { + if (input.checkpoint == null || deps.actor == null) { + throw new Error('Event actor checkpoint adapter is unavailable'); + } + return { turn: input.checkpoint, actor: deps.actor }; +} + +function requireHistory( + input: CreateAgentEventActorTurnInput, + deps: AgentEventActorTurnDependencies, +): { + turn: AgentEventActorHistoryTurn; + history: NonNullable; +} { + if (input.history == null || deps.history == null) { + throw new Error('Event actor history adapter is unavailable'); + } + return { turn: input.history, history: deps.history }; +} + +async function completeHistoryTurn( + owner: AgentEventActorTurnOwner, + token: string, + complete: ConversationMethods['completeAgentEventActorLegacyTurn'], +): Promise { + return complete({ + user: owner.user, + conversationId: owner.conversationId, + ...(owner.tenantId == null ? {} : { tenantId: owner.tenantId }), + token, + }); +} + +/** One Event Actor turn interface with checkpoint and history as private state-loading adapters. */ +export function createAgentEventActorTurn( + input: CreateAgentEventActorTurnInput, + deps: AgentEventActorTurnDependencies, +): AgentEventActorTurn { + let started = false; + let historyToken: string | undefined; + let historyOwner: AgentEventActorTurnOwner | undefined; + + const run = async (): Promise> => { + if (started) { + throw new Error('Event actor turn already started'); + } + started = true; + + if (input.strategy === 'checkpoint') { + const { turn, actor } = requireCheckpoint(input, deps); + const result = + turn.kind === 'resume' + ? await resumeAgentEventActor(turn.input, actor) + : await executeAgentEventActor(turn.input, actor); + return { adapter: 'checkpoint', ...result }; + } + + const { turn, history } = requireHistory(input, deps); + const token = randomUUID(); + const acquired = await history.begin({ + user: turn.owner.user, + conversationId: turn.owner.conversationId, + ...(turn.owner.tenantId == null ? {} : { tenantId: turn.owner.tenantId }), + token, + }); + if (!acquired) { + throw Object.assign(new Error('The event actor is temporarily unavailable'), { + code: 'EVENT_ACTOR_NOT_READY', + status: 409, + }) as EventActorTurnError; + } + historyToken = token; + historyOwner = turn.owner; + try { + await turn.persistToken(token); + } catch (error) { + try { + const completed = await completeHistoryTurn(turn.owner, token, history.complete); + if (completed) { + historyToken = undefined; + historyOwner = undefined; + } else { + logger.error( + `[event-actor] Unstarted history turn ${token} was not released; durable error handling will retry`, + ); + } + } catch (completionError) { + logger.error('[event-actor] Failed to release an unstarted history turn', completionError); + } + throw error; + } + return { adapter: 'history', value: await turn.invoke() }; + }; + + const historyPersisted = async (): Promise => { + if (historyToken == null || historyOwner == null || deps.history == null) { + return; + } + const token = historyToken; + const owner = historyOwner; + try { + const completed = await completeHistoryTurn(owner, token, deps.history.complete); + if (completed) { + historyToken = undefined; + historyOwner = undefined; + } else { + logger.error( + `[event-actor] History turn ${token} was not settled; invocation forks remain blocked`, + ); + } + } catch (error) { + logger.error( + `[event-actor] Failed to settle history turn ${token}; invocation forks remain blocked`, + error, + ); + } + }; + + return Object.freeze({ run, historyPersisted }); +} + +/** Read-compatible settlement for a history turn created by an earlier generation. */ +export function settleAgentEventActorHistoryTurn( + input: AgentEventActorTurnOwner & { token: string }, + complete: ConversationMethods['completeAgentEventActorLegacyTurn'], +): Promise { + return completeHistoryTurn(input, input.token, complete); +}