🤖 feat: make Event Actor HITL durable (#15305)

* 🤖 feat: make event actor HITL durable

* 🤖 fix: break event actor outcome cycle

* 🤖 fix: close durable actor terminal races

* fix: harden durable event actor recovery proofs

* fix: close event actor resume publication races

Author: danny-avila

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. Pause-capable actors remain on the existing resumable path until forked HITL has an explicit contract.
+- **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. Shared-store deployments publish new suspensions only under generation protocol v2, after the homogeneous-fleet cutover; protocol v1 keeps pause-capable actors on the legacy history path so an old resume consumer cannot consume unknown evidence. 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.
 - **Theme definition**: a versioned, data-only description of LibreChat semantic colors and shared appearance roles, optionally specialized by light or dark mode. The theme module validates and resolves partial definitions against bundled defaults before adapters apply them. A theme definition does not contain arbitrary CSS, application behavior, or alternate feature layouts.

api/package.json

@@ -46,7 +46,7 @@
     "@azure/storage-blob": "^12.30.0",
     "@google/genai": "^2.8.0",
     "@keyv/redis": "5.1.6",
-    "@librechat/agents": "^3.7.6",
+    "@librechat/agents": "^3.7.7",
     "@librechat/api": "*",
     "@librechat/data-schemas": "*",
     "@microsoft/microsoft-graph-client": "^3.0.7",

api/server/controllers/agents/__tests__/client.eventActorPause.spec.js

@@ -0,0 +1,74 @@
+const mockPause = jest.fn();
+const mockGetJob = jest.fn();
+
+jest.mock('@librechat/api', () => ({
+  ...jest.requireActual('@librechat/api'),
+  GenerationJobManager: {
+    approvals: { pause: (...args) => mockPause(...args) },
+    getJob: (...args) => mockGetJob(...args),
+  },
+}));
+
+const AgentClient = require('../client');
+
+function clientForProjection() {
+  const pendingAction = { actionId: 'action-1', expiresAt: Date.now() + 60_000 };
+  return {
+    stagedApproval: {
+      streamId: 'conversation-1',
+      pendingAction,
+      discoveredTools: [],
+      activityPhaseSnapshot: null,
+    },
+    pendingApproval: null,
+    jobCreatedAt: 123,
+  };
+}
+
+describe('AgentClient Event Actor pause projection', () => {
+  beforeEach(() => {
+    jest.clearAllMocks();
+  });
+
+  it('confirms the exact durable projection when Redis loses the pause reply', async () => {
+    const self = clientForProjection();
+    const suspension = { version: 1, suspensionId: 'suspension-1', attempt: 2 };
+    mockPause.mockRejectedValue(new Error('reply lost'));
+    mockGetJob.mockResolvedValue({
+      createdAt: 123,
+      status: 'requires_action',
+      metadata: {
+        pendingAction: self.stagedApproval.pendingAction,
+        agentEventSuspension: suspension,
+      },
+    });
+
+    await expect(AgentClient.prototype.publishStagedApproval.call(self, suspension)).resolves.toBe(
+      true,
+    );
+    expect(self.pendingApproval).toBe(self.stagedApproval.pendingAction);
+  });
+
+  it('propagates an ambiguous failure when the durable projection does not match', async () => {
+    const self = clientForProjection();
+    const error = new Error('reply lost');
+    mockPause.mockRejectedValue(error);
+    mockGetJob.mockResolvedValue({
+      createdAt: 123,
+      status: 'requires_action',
+      metadata: {
+        pendingAction: self.stagedApproval.pendingAction,
+        agentEventSuspension: { version: 1, suspensionId: 'different', attempt: 2 },
+      },
+    });
+
+    await expect(
+      AgentClient.prototype.publishStagedApproval.call(self, {
+        version: 1,
+        suspensionId: 'suspension-1',
+        attempt: 2,
+      }),
+    ).rejects.toBe(error);
+    expect(self.pendingApproval).toBeNull();
+  });
+});

api/server/controllers/agents/__tests__/request.resumeMetadata.spec.js

@@ -85,8 +85,8 @@ const mockResolveAgentTurnExecutionPlan = jest.fn((input) => {
     input.event?.binding != null &&
     input.event?.expectedAction != null &&
     input.checkpointerType !== 'memory' &&
-    !input.canPause &&
-    !input.expectedActionMayDetach;
+    !input.expectedActionMayDetach &&
+    (!input.canPause || input.durableEventActorSuspensions);
   let strategy = 'history';
   if (input.isNewConversation) {
     strategy = 'fresh';
@@ -3042,10 +3042,12 @@ describe('ResumableAgentController resume metadata', () => {
     });
     let observedHookResult;
     const completedResponseWrite = jest.fn();
+    const exposePendingApproval = jest.fn().mockResolvedValue(undefined);
     const client = {
       options: {},
       jobCreatedAt: 1000,
       pendingApproval: { actionId: 'action-pause-barrier' },
+      exposePendingApproval,
       skipSaveUserMessage: false,
       skipSaveConvo: false,
       getSaveOptions: jest.fn(() => ({ endpoint: 'agents' })),
@@ -3125,6 +3127,9 @@ describe('ResumableAgentController resume metadata', () => {
     expect(mockSaveMessage.mock.invocationCallOrder[0]).toBeLessThan(
       mockGenerationJobManager.approvals.finishPausePersistence.mock.invocationCallOrder[0],
     );
+    expect(exposePendingApproval.mock.invocationCallOrder[0]).toBeLessThan(
+      mockGenerationJobManager.approvals.finishPausePersistence.mock.invocationCallOrder[0],
+    );
     expect(mockGenerationJobManager.claimTerminalJob).not.toHaveBeenCalled();
   });
 
@@ -4482,15 +4487,36 @@ describe('ResumableAgentController resume metadata', () => {
       undefined,
       undefined,
     ],
+    [
+      'pre-cutover pause-capable fleet',
+      { toolDefinitions: [] },
+      { toolApproval: { enabled: true } },
+      undefined,
+      undefined,
+      1,
+    ],
   ])(
-    'keeps %s event actors on the existing resumable path',
-    async (_label, agent, config, agentConfigs, clientOptions) => {
+    'routes %s event actors through the compatible continuation path',
+    async (_label, agent, config, agentConfigs, clientOptions, generationProtocolVersion = 2) => {
       mockGenerationJobManager.claimGeneration.mockResolvedValue(
         wonGenerationClaim({
           streamId: 'child-conversation',
           conversationId: 'child-conversation',
+          generationProtocolVersion,
         }),
       );
+      mockGenerationJobManager.createJob.mockResolvedValueOnce({
+        createdAt: 1000,
+        metadata: {
+          checkpointNamespace: '1000',
+          providerExecutionId: 'provider-segment-1',
+          providerDrained: true,
+          generationProtocolVersion,
+        },
+        readyPromise: Promise.resolve(),
+        abortController: new AbortController(),
+        emitter: { on: jest.fn() },
+      });
       mockGetConvo.mockResolvedValue({
         conversationId: 'parent-conversation',
         agent_id: 'parent-agent',
@@ -4503,12 +4529,28 @@ describe('ResumableAgentController resume metadata', () => {
           throw new Error('stop after legacy event invocation started');
         }),
       };
+      const shouldCheckpoint =
+        _label !== 'memory-checkpointer' &&
+        _label !== 'background-capable expected action' &&
+        _label !== 'pre-cutover pause-capable fleet';
+      if (shouldCheckpoint) {
+        mockExecuteAgentEventActor.mockImplementationOnce(async (input) => {
+          await input.invoke({
+            checkpointNamespace: 'event-actor/pause-capable',
+            checkpointId: 'checkpoint-pause-capable',
+            invocationId: 'req-event-hitl',
+            continuation: 'warm',
+            signal: input.signal,
+          });
+        });
+      }
       const req = {
         user: { id: 'user-123', tenantId: 'tenant-1' },
         body: {
           text: 'Continue with a pause-capable actor.',
           clientRequestId: 'req-event-hitl',
           conversationId: 'child-conversation',
+          generationProtocolVersion,
           endpointOption: { endpoint: 'agents', modelOptions: { model: 'gpt-4.1' } },
           agentEventDelivery: {
             deliveryKey: 'req-event-hitl',
@@ -4541,6 +4583,18 @@ describe('ResumableAgentController resume metadata', () => {
       );
       await nextTick();
 
+      if (shouldCheckpoint) {
+        expect(mockExecuteAgentEventActor).toHaveBeenCalled();
+        expect(mockGetMessages).not.toHaveBeenCalled();
+        expect(mockBeginAgentEventActorLegacyTurn).not.toHaveBeenCalled();
+        expect(mockGenerationJobManager.updateMetadata).not.toHaveBeenCalledWith(
+          'child-conversation',
+          expect.objectContaining({ agentEventLegacyTurnToken: expect.any(String) }),
+          1000,
+        );
+        expect(client.sendMessage).toHaveBeenCalledTimes(1);
+        return;
+      }
       expect(mockExecuteAgentEventActor).not.toHaveBeenCalled();
       expect(mockBeginAgentEventActorLegacyTurn).toHaveBeenCalledWith({
         user: 'user-123',