mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-09-21 15:45:22 +00:00
🤖 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
This commit is contained in:
Danny Avila
GitHub
parent
31325b4f61
commit
c06b09c945
35 changed files with 3812 additions and 121 deletions
|
|
@ -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();
|
||||
});
|
||||
});
|
||||
|
|
@ -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',
|
||||
|
|
|
|||
|
|
@ -103,6 +103,13 @@ const mockAcquireEventChildGenerationLease = jest.fn();
|
|||
const mockReleaseEventChildLease = jest.fn();
|
||||
const mockIsSubagentOwnerAdmissible = jest.fn();
|
||||
const mockCompleteAgentEventActorLegacyTurn = jest.fn();
|
||||
const mockGetAgentEventActorSnapshot = jest.fn();
|
||||
const mockCommitAgentEventActorState = jest.fn();
|
||||
const mockStoreAgentEventActorSuspension = jest.fn();
|
||||
const mockClaimAgentEventActorSuspension = jest.fn();
|
||||
const mockSettleAgentEventActorSuspension = jest.fn();
|
||||
const mockRecordAgentEventActorReconciliation = jest.fn();
|
||||
const mockResumeAgentEventActor = jest.fn();
|
||||
|
||||
jest.mock('@librechat/data-schemas', () => ({
|
||||
...jest.requireActual('@librechat/data-schemas'),
|
||||
|
|
@ -124,6 +131,7 @@ jest.mock('@librechat/api', () => ({
|
|||
}),
|
||||
getAgentCheckpointer: (...args) => mockGetAgentCheckpointer(...args),
|
||||
checkAccess: (...args) => mockCheckAccess(...args),
|
||||
resumeAgentEventActor: (...args) => mockResumeAgentEventActor(...args),
|
||||
}));
|
||||
|
||||
jest.mock('~/models', () => ({
|
||||
|
|
@ -137,6 +145,13 @@ jest.mock('~/models', () => ({
|
|||
getRoleByName: (...args) => mockGetRoleByName(...args),
|
||||
isSubagentOwnerAdmissible: (...args) => mockIsSubagentOwnerAdmissible(...args),
|
||||
completeAgentEventActorLegacyTurn: (...args) => mockCompleteAgentEventActorLegacyTurn(...args),
|
||||
getAgentEventActorSnapshot: (...args) => mockGetAgentEventActorSnapshot(...args),
|
||||
commitAgentEventActorState: (...args) => mockCommitAgentEventActorState(...args),
|
||||
storeAgentEventActorSuspension: (...args) => mockStoreAgentEventActorSuspension(...args),
|
||||
claimAgentEventActorSuspension: (...args) => mockClaimAgentEventActorSuspension(...args),
|
||||
settleAgentEventActorSuspension: (...args) => mockSettleAgentEventActorSuspension(...args),
|
||||
recordAgentEventActorReconciliation: (...args) =>
|
||||
mockRecordAgentEventActorReconciliation(...args),
|
||||
}));
|
||||
|
||||
jest.mock('~/server/services/Endpoints/agents/eventChildLease', () => ({
|
||||
|
|
@ -354,6 +369,12 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => {
|
|||
mockReleaseEventChildLease.mockResolvedValue(undefined);
|
||||
mockIsSubagentOwnerAdmissible.mockResolvedValue(true);
|
||||
mockCompleteAgentEventActorLegacyTurn.mockResolvedValue(true);
|
||||
mockGetAgentEventActorSnapshot.mockResolvedValue(undefined);
|
||||
mockCommitAgentEventActorState.mockResolvedValue({ status: 'committed' });
|
||||
mockStoreAgentEventActorSuspension.mockResolvedValue({ status: 'stored' });
|
||||
mockClaimAgentEventActorSuspension.mockResolvedValue({ status: 'claimed' });
|
||||
mockSettleAgentEventActorSuspension.mockResolvedValue({ status: 'settled' });
|
||||
mockRecordAgentEventActorReconciliation.mockResolvedValue(true);
|
||||
endpointAgent = {
|
||||
_id: 'mongo-agent-abc',
|
||||
id: AGENT_ID,
|
||||
|
|
@ -441,6 +462,307 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => {
|
|||
};
|
||||
|
||||
describe('event-bound actor resume lifecycle', () => {
|
||||
it('claims a versioned Conversation suspension and recovers an ambiguous job projection ACK', async () => {
|
||||
configureEventActorResume();
|
||||
requestStateOverrides._agentEventBindingId = 'binding-1';
|
||||
const expectedAction = { toolName: 'lookup' };
|
||||
const suspension = {
|
||||
version: 1,
|
||||
suspensionId: 'suspension-1',
|
||||
attempt: 0,
|
||||
issuedAt: Date.now(),
|
||||
expiresAt: Date.now() + 60_000,
|
||||
invocation: {
|
||||
invocationId: 'trigger_event_delivery',
|
||||
continuation: 'warm',
|
||||
base: { actorThreadId: CONVO_ID, generation: 1 },
|
||||
fork: {
|
||||
threadId: CONVO_ID,
|
||||
checkpointNs: 'event-actor',
|
||||
checkpointId: 'checkpoint-paused',
|
||||
invocationId: 'trigger_event_delivery',
|
||||
},
|
||||
},
|
||||
checkpoint: {
|
||||
threadId: CONVO_ID,
|
||||
checkpointNs: 'event-actor',
|
||||
checkpointId: 'checkpoint-paused',
|
||||
invocationId: 'trigger_event_delivery',
|
||||
},
|
||||
interrupt: {
|
||||
id: 'interrupt-1',
|
||||
payload: {
|
||||
type: 'tool_approval',
|
||||
_librechatEventActor: { expectedAction },
|
||||
},
|
||||
},
|
||||
suspensionDigest: 'signed-digest',
|
||||
};
|
||||
const pausedJob = makeToolApprovalJob({
|
||||
metadata: {
|
||||
idempotencyClientRequestId: 'trigger_event_delivery',
|
||||
agentEventExpectedAction: expectedAction,
|
||||
agentEventSuspension: {
|
||||
version: 1,
|
||||
suspensionId: suspension.suspensionId,
|
||||
attempt: suspension.attempt,
|
||||
},
|
||||
},
|
||||
});
|
||||
pausedJob.metadata.pendingAction.payload.review_configs = [
|
||||
{ tool_call_id: 'tc1', allowed_decisions: ['respond'] },
|
||||
];
|
||||
mockGenerationJobManager.getJob.mockResolvedValue(pausedJob);
|
||||
let projectedProviderExecutionId;
|
||||
mockGenerationJobManager.approvals.resolve.mockImplementation(
|
||||
async (_streamId, _actionId, resumePatch) => {
|
||||
projectedProviderExecutionId = resumePatch.providerExecutionId;
|
||||
mockGenerationJobManager.getJob.mockResolvedValue({
|
||||
...pausedJob,
|
||||
status: 'running',
|
||||
metadata: {
|
||||
...pausedJob.metadata,
|
||||
providerExecutionId: resumePatch.providerExecutionId,
|
||||
},
|
||||
});
|
||||
throw new Error('redis committed the CAS but lost its reply');
|
||||
},
|
||||
);
|
||||
mockGetAgentEventActorSnapshot.mockResolvedValue({
|
||||
state: null,
|
||||
epoch: 1,
|
||||
legacyTurn: null,
|
||||
reconciliations: [],
|
||||
suspension: {
|
||||
suspension,
|
||||
actionId: ACTION_ID,
|
||||
jobCreatedAt: 1000,
|
||||
status: 'pending',
|
||||
},
|
||||
});
|
||||
const resumedClient = makeClient({
|
||||
contentParts: [makeToolCallContent({ output: 'human supplied output' })],
|
||||
run: {
|
||||
getRunSteps: () => [
|
||||
{
|
||||
type: 'tool_calls',
|
||||
status: 'completed',
|
||||
stepDetails: {
|
||||
type: 'tool_calls',
|
||||
tool_calls: [
|
||||
{
|
||||
id: 'tc1',
|
||||
name: 'lookup',
|
||||
args: {},
|
||||
output: 'human supplied output',
|
||||
},
|
||||
],
|
||||
},
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
mockInitializeClient.mockResolvedValue({ client: resumedClient, userMCPAuthMap: {} });
|
||||
mockResumeAgentEventActor.mockImplementation(async (input, dependencies) => {
|
||||
await dependencies.claimSuspension({
|
||||
user: USER_ID,
|
||||
tenantId: TENANT_ID,
|
||||
conversationId: CONVO_ID,
|
||||
suspensionId: suspension.suspensionId,
|
||||
attempt: suspension.attempt,
|
||||
actionId: ACTION_ID,
|
||||
jobCreatedAt: 1000,
|
||||
resumeAttemptId: input.resumeAttemptId,
|
||||
});
|
||||
expect(await input.claimProjection()).toBe(true);
|
||||
expect(input.resumeAttemptId).toBe(projectedProviderExecutionId);
|
||||
const value = await input.resume({
|
||||
checkpointNamespace: 'event-actor',
|
||||
checkpointId: 'checkpoint-paused',
|
||||
invocationId: 'trigger_event_delivery',
|
||||
continuation: 'warm',
|
||||
signal: input.signal,
|
||||
});
|
||||
expect(input.readAppliedAction()).toBeUndefined();
|
||||
return {
|
||||
value,
|
||||
execution: { status: 'completed_no_action' },
|
||||
};
|
||||
});
|
||||
|
||||
const res = await post(
|
||||
approveBody({
|
||||
decisions: [
|
||||
{ tool_call_id: 'tc1', decision: 'respond', responseText: 'human supplied output' },
|
||||
],
|
||||
}),
|
||||
);
|
||||
expect(res.status).toBe(200);
|
||||
await settled;
|
||||
await flush();
|
||||
|
||||
expect(mockGetAgentEventActorSnapshot).toHaveBeenCalledWith({
|
||||
user: USER_ID,
|
||||
tenantId: TENANT_ID,
|
||||
conversationId: CONVO_ID,
|
||||
});
|
||||
expect(mockClaimAgentEventActorSuspension.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
mockGenerationJobManager.approvals.resolve.mock.invocationCallOrder[0],
|
||||
);
|
||||
expect(mockGenerationJobManager.getJob).toHaveBeenCalledTimes(2);
|
||||
expect(resumedClient.resumeCompletion).toHaveBeenCalledTimes(1);
|
||||
expect(mockRecordAgentEventActorReconciliation).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('does not record provider execution when client reconstruction fails before continuation', async () => {
|
||||
configureEventActorResume();
|
||||
const suspension = {
|
||||
version: 1,
|
||||
suspensionId: 'suspension-init-failure',
|
||||
attempt: 0,
|
||||
issuedAt: Date.now(),
|
||||
expiresAt: Date.now() + 60_000,
|
||||
invocation: {
|
||||
invocationId: 'trigger_event_delivery',
|
||||
continuation: 'warm',
|
||||
base: { actorThreadId: CONVO_ID, generation: 0 },
|
||||
fork: {
|
||||
threadId: CONVO_ID,
|
||||
checkpointNs: 'event-actor',
|
||||
checkpointId: 'checkpoint-paused',
|
||||
invocationId: 'trigger_event_delivery',
|
||||
},
|
||||
},
|
||||
checkpoint: {
|
||||
threadId: CONVO_ID,
|
||||
checkpointNs: 'event-actor',
|
||||
checkpointId: 'checkpoint-paused',
|
||||
invocationId: 'trigger_event_delivery',
|
||||
},
|
||||
interrupt: {
|
||||
id: 'interrupt-init-failure',
|
||||
payload: {
|
||||
type: 'tool_approval',
|
||||
_librechatEventActor: { expectedAction: { toolName: 'lookup' } },
|
||||
},
|
||||
},
|
||||
suspensionDigest: 'signed-digest',
|
||||
};
|
||||
mockGenerationJobManager.getJob.mockResolvedValue(
|
||||
makeToolApprovalJob({
|
||||
metadata: {
|
||||
idempotencyClientRequestId: 'trigger_event_delivery',
|
||||
agentEventSuspension: {
|
||||
version: 1,
|
||||
suspensionId: suspension.suspensionId,
|
||||
attempt: suspension.attempt,
|
||||
},
|
||||
},
|
||||
}),
|
||||
);
|
||||
mockGetAgentEventActorSnapshot.mockResolvedValue({
|
||||
state: null,
|
||||
epoch: 1,
|
||||
legacyTurn: null,
|
||||
reconciliations: [],
|
||||
suspension: {
|
||||
suspension,
|
||||
actionId: ACTION_ID,
|
||||
jobCreatedAt: 1000,
|
||||
status: 'pending',
|
||||
},
|
||||
});
|
||||
mockInitializeClient.mockRejectedValue(new Error('client reconstruction failed'));
|
||||
mockResumeAgentEventActor.mockImplementation(async (input) => {
|
||||
expect(await input.claimProjection()).toBe(true);
|
||||
return input.resume({
|
||||
checkpointNamespace: 'event-actor',
|
||||
checkpointId: 'checkpoint-paused',
|
||||
invocationId: 'trigger_event_delivery',
|
||||
continuation: 'warm',
|
||||
signal: input.signal,
|
||||
});
|
||||
});
|
||||
|
||||
const res = await post(approveBody());
|
||||
expect(res.status).toBe(200);
|
||||
await settled;
|
||||
await flush();
|
||||
|
||||
expect(mockInitializeClient).toHaveBeenCalledTimes(1);
|
||||
expect(mockGenerationJobManager.beginProviderExecution).not.toHaveBeenCalled();
|
||||
expect(mockGenerationJobManager.completeJob).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('fails closed when a versioned job marker no longer matches canonical suspension', async () => {
|
||||
configureEventActorResume();
|
||||
mockGenerationJobManager.getJob.mockResolvedValue(
|
||||
makeToolApprovalJob({
|
||||
metadata: {
|
||||
agentEventSuspension: { version: 1, suspensionId: 'stale', attempt: 0 },
|
||||
},
|
||||
}),
|
||||
);
|
||||
mockGetAgentEventActorSnapshot.mockResolvedValue({
|
||||
state: null,
|
||||
epoch: 1,
|
||||
legacyTurn: null,
|
||||
reconciliations: [],
|
||||
suspension: null,
|
||||
});
|
||||
|
||||
const res = await post(approveBody());
|
||||
|
||||
expect(res.status).toBe(409);
|
||||
expect(res.body).toMatchObject({ code: 'EVENT_ACTOR_SUSPENSION_STALE' });
|
||||
expect(mockGenerationJobManager.approvals.resolve).not.toHaveBeenCalled();
|
||||
expect(mockResumeAgentEventActor).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('fails promptly when suspension validation rejects before the job claim callback', async () => {
|
||||
configureEventActorResume();
|
||||
const suspension = {
|
||||
version: 1,
|
||||
suspensionId: 'suspension-invalid-signature',
|
||||
attempt: 0,
|
||||
invocation: { invocationId: 'trigger_event_delivery' },
|
||||
interrupt: {
|
||||
payload: { _librechatEventActor: { expectedAction: { toolName: 'lookup' } } },
|
||||
},
|
||||
};
|
||||
mockGenerationJobManager.getJob.mockResolvedValue(
|
||||
makeToolApprovalJob({
|
||||
metadata: {
|
||||
idempotencyClientRequestId: 'trigger_event_delivery',
|
||||
agentEventSuspension: {
|
||||
version: 1,
|
||||
suspensionId: suspension.suspensionId,
|
||||
attempt: 0,
|
||||
},
|
||||
},
|
||||
}),
|
||||
);
|
||||
mockGetAgentEventActorSnapshot.mockResolvedValue({
|
||||
state: null,
|
||||
epoch: 1,
|
||||
legacyTurn: null,
|
||||
reconciliations: [],
|
||||
suspension: {
|
||||
suspension,
|
||||
actionId: ACTION_ID,
|
||||
jobCreatedAt: 1000,
|
||||
status: 'pending',
|
||||
},
|
||||
});
|
||||
mockResumeAgentEventActor.mockRejectedValue(new Error('invalid signed suspension'));
|
||||
|
||||
const res = await post(approveBody());
|
||||
|
||||
expect(res.status).toBe(500);
|
||||
expect(mockGenerationJobManager.approvals.resolve).not.toHaveBeenCalled();
|
||||
expect(mockDecrementPendingRequest).toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('leaves the approval pending when the previous segment still owns the lease', async () => {
|
||||
configureEventActorResume();
|
||||
mockGenerationJobManager.getJob.mockResolvedValue(makeToolApprovalJob());
|
||||
|
|
@ -1674,7 +1996,7 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => {
|
|||
);
|
||||
expect(
|
||||
mockGenerationJobManager.beginProviderExecution.mock.invocationCallOrder[0],
|
||||
).toBeLessThan(mockInitializeClient.mock.invocationCallOrder[0]);
|
||||
).toBeGreaterThan(mockInitializeClient.mock.invocationCallOrder[0]);
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -3134,8 +3456,12 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => {
|
|||
describe('non-finalizing outcomes', () => {
|
||||
it('re-pause: does not finalize when the run pauses again', async () => {
|
||||
mockGenerationJobManager.getJob.mockResolvedValue(makeToolApprovalJob());
|
||||
const exposePendingApproval = jest.fn().mockResolvedValue(undefined);
|
||||
mockInitializeClient.mockResolvedValue({
|
||||
client: makeClient({ pendingApproval: { actionId: NEXT_ACTION_ID } }),
|
||||
client: makeClient({
|
||||
pendingApproval: { actionId: NEXT_ACTION_ID },
|
||||
exposePendingApproval,
|
||||
}),
|
||||
userMCPAuthMap: {},
|
||||
});
|
||||
|
||||
|
|
@ -3165,6 +3491,9 @@ describe('ResumeAgentController (POST /agents/chat/resume)', () => {
|
|||
1000,
|
||||
);
|
||||
expect(mockGenerationJobManager.failPausePersistence).not.toHaveBeenCalled();
|
||||
expect(exposePendingApproval.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
mockGenerationJobManager.approvals.finishPausePersistence.mock.invocationCallOrder[0],
|
||||
);
|
||||
// The slot is still released and the client disposed.
|
||||
expect(mockDecrementPendingRequest).toHaveBeenCalledWith(USER_ID);
|
||||
expect(mockDisposeClient).toHaveBeenCalledTimes(1);
|
||||
|
|
|
|||
|
|
@ -3410,12 +3410,135 @@ class AgentClient extends BaseClient {
|
|||
activityPhase?.complete?.();
|
||||
}
|
||||
|
||||
/** Returns the exact staged approval envelope the SDK signs into a suspension. */
|
||||
readEventActorSuspension() {
|
||||
const staged = this.stagedApproval;
|
||||
if (staged == null || this.eventActorInvocationId == null) {
|
||||
return undefined;
|
||||
}
|
||||
return {
|
||||
actionId: staged.pendingAction.actionId,
|
||||
jobCreatedAt: this.jobCreatedAt,
|
||||
interrupt: {
|
||||
id: staged.interruptId,
|
||||
payload: { ...staged.pendingAction, type: staged.interruptType },
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
/** Projects an already-staged pause into the shared job store. Event Actors
|
||||
* call this only after their signed Conversation suspension is durable. */
|
||||
async publishStagedApproval(eventActorSuspension) {
|
||||
const staged = this.stagedApproval;
|
||||
if (staged == null) {
|
||||
return false;
|
||||
}
|
||||
if (this.pendingApproval?.actionId === staged.pendingAction.actionId) {
|
||||
return true;
|
||||
}
|
||||
const pauseProjection = {
|
||||
expectedCreatedAt: this.jobCreatedAt,
|
||||
...(staged.discoveredTools.length > 0 ? { discoveredTools: staged.discoveredTools } : {}),
|
||||
...(staged.activityPhaseSnapshot == null
|
||||
? {}
|
||||
: { activityPhaseSnapshot: staged.activityPhaseSnapshot }),
|
||||
persistencePending: true,
|
||||
...(eventActorSuspension == null
|
||||
? {}
|
||||
: {
|
||||
agentEventSuspension: {
|
||||
version: eventActorSuspension.version,
|
||||
suspensionId: eventActorSuspension.suspensionId,
|
||||
attempt: eventActorSuspension.attempt,
|
||||
},
|
||||
}),
|
||||
};
|
||||
let paused;
|
||||
try {
|
||||
paused = await GenerationJobManager.approvals.pause(
|
||||
staged.streamId,
|
||||
staged.pendingAction,
|
||||
pauseProjection,
|
||||
);
|
||||
} catch (error) {
|
||||
/** Redis may commit running -> requires_action and lose only its reply.
|
||||
* The Conversation suspension is already canonical at this point, so
|
||||
* confirm this exact generation/action/projection before declaring the
|
||||
* publication failed and driving terminal compensation. */
|
||||
const currentJob = await GenerationJobManager.getJob(staged.streamId).catch(() => null);
|
||||
const projected = currentJob?.metadata?.agentEventSuspension;
|
||||
const expectedProjection = pauseProjection.agentEventSuspension;
|
||||
if (
|
||||
currentJob?.createdAt === this.jobCreatedAt &&
|
||||
currentJob.status === 'requires_action' &&
|
||||
currentJob.metadata?.pendingAction?.actionId === staged.pendingAction.actionId &&
|
||||
expectedProjection != null &&
|
||||
projected?.version === expectedProjection.version &&
|
||||
projected.suspensionId === expectedProjection.suspensionId &&
|
||||
projected.attempt === expectedProjection.attempt
|
||||
) {
|
||||
paused = true;
|
||||
} else {
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
if (!paused) {
|
||||
logger.debug(
|
||||
`[AgentClient] Interrupt fired but job ${staged.streamId} was not running; not pausing`,
|
||||
);
|
||||
return false;
|
||||
}
|
||||
this.pendingApproval = staged.pendingAction;
|
||||
return true;
|
||||
}
|
||||
|
||||
/** Exposes a durable pause after its controller-owned history barrier clears. */
|
||||
async exposePendingApproval() {
|
||||
const staged = this.stagedApproval;
|
||||
if (
|
||||
staged == null ||
|
||||
this.pendingApproval?.actionId !== staged.pendingAction.actionId ||
|
||||
this.exposedApprovalActionId === staged.pendingAction.actionId
|
||||
) {
|
||||
return false;
|
||||
}
|
||||
if (!this.pendingRequestReleased) {
|
||||
try {
|
||||
if (this.options.req?._scheduleConcurrencyExempt !== true) {
|
||||
await decrementPendingRequest(this.options.req?.user?.id);
|
||||
}
|
||||
this.pendingRequestReleased = true;
|
||||
} catch (err) {
|
||||
logger.error(
|
||||
`[AgentClient] Failed to release request slot on pause ${staged.streamId}`,
|
||||
getSafeErrorMetadata(err),
|
||||
);
|
||||
}
|
||||
}
|
||||
// Steers accepted before the pause remain in the shared store throughout
|
||||
// review. The resumed run rehydrates them; exposing the action never moves
|
||||
// their only copy into this replica's ephemeral client state.
|
||||
await GenerationJobManager.emitChunk(
|
||||
staged.streamId,
|
||||
{
|
||||
event: ApprovalEvents.ON_PENDING_ACTION,
|
||||
data: toClientPendingAction(staged.pendingAction),
|
||||
},
|
||||
{ expectedCreatedAt: this.jobCreatedAt },
|
||||
);
|
||||
this.exposedApprovalActionId = staged.pendingAction.actionId;
|
||||
logger.debug(
|
||||
`[AgentClient] Paused ${staged.streamId} for ${staged.interruptType} (action ${staged.pendingAction.actionId})`,
|
||||
);
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* Surface any human-in-the-loop interrupt the SDK captured during the most
|
||||
* recent `processStream` / `resume`. When the run paused for tool approval (or
|
||||
* an ask-user question), mark the job `requires_action`, persist the pending
|
||||
* review record, and emit it to live clients — then set `this.pendingApproval`
|
||||
* so the controller leaves the turn unfinalized for the resume route to continue.
|
||||
* an ask-user question), stage its exact envelope. Ordinary turns immediately
|
||||
* publish and expose it; Event Actors let the SDK persist signed suspension
|
||||
* evidence first, then publish under the same history barrier.
|
||||
*
|
||||
* No-op when the run completed without an interrupt, or when the job was aborted
|
||||
* between the interrupt firing and this mark (a late interrupt must not pause a
|
||||
|
|
@ -3554,59 +3677,20 @@ class AgentClient extends BaseClient {
|
|||
);
|
||||
}
|
||||
|
||||
const paused = await GenerationJobManager.approvals.pause(streamId, pendingAction, {
|
||||
expectedCreatedAt: this.jobCreatedAt,
|
||||
...(discoveredTools.length > 0 ? { discoveredTools } : {}),
|
||||
...(this.activityPhaseWiring?.snapshot != null && {
|
||||
activityPhaseSnapshot: this.activityPhaseWiring.snapshot(),
|
||||
}),
|
||||
persistencePending: true,
|
||||
});
|
||||
if (!paused) {
|
||||
logger.debug(
|
||||
`[AgentClient] Interrupt fired but job ${streamId} was not running; not pausing`,
|
||||
);
|
||||
this.stagedApproval = {
|
||||
streamId,
|
||||
pendingAction,
|
||||
interruptId: interrupt.interruptId,
|
||||
interruptType: interrupt.payload.type,
|
||||
discoveredTools,
|
||||
activityPhaseSnapshot: this.activityPhaseWiring?.snapshot?.(),
|
||||
};
|
||||
if (this.eventActorInvocationId != null) {
|
||||
return;
|
||||
}
|
||||
|
||||
this.pendingApproval = pendingAction;
|
||||
// Release the concurrency slot this request held the MOMENT the turn is durably
|
||||
// paused — before the approval card is emitted — so the user's `/resume` can
|
||||
// re-acquire one immediately. Otherwise a fast Approve races the HTTP-driver
|
||||
// teardown (request.js pause branch / resume.js finally) that would otherwise
|
||||
// release it, and `/resume` 429s under LIMIT_CONCURRENT_MESSAGES. Idempotent via
|
||||
// the flag; if it fails here, the teardown still releases (it checks the flag).
|
||||
if (!this.pendingRequestReleased) {
|
||||
try {
|
||||
if (this.options.req?._scheduleConcurrencyExempt !== true) {
|
||||
await decrementPendingRequest(this.options.req?.user?.id);
|
||||
}
|
||||
this.pendingRequestReleased = true;
|
||||
} catch (err) {
|
||||
logger.error(
|
||||
`[AgentClient] Failed to release request slot on pause ${streamId}`,
|
||||
getSafeErrorMetadata(err),
|
||||
);
|
||||
}
|
||||
if (await this.publishStagedApproval()) {
|
||||
await this.exposePendingApproval();
|
||||
}
|
||||
await GenerationJobManager.emitChunk(
|
||||
streamId,
|
||||
{
|
||||
event: ApprovalEvents.ON_PENDING_ACTION,
|
||||
data: toClientPendingAction(pendingAction),
|
||||
},
|
||||
{ expectedCreatedAt: this.jobCreatedAt },
|
||||
);
|
||||
// Steers queued before this pause stay IN the store for the whole approval
|
||||
// window: `resumeState.pendingSteers` re-seeds the client's chips on
|
||||
// reload, and the resumed run drains them at its first tool boundary.
|
||||
// Draining here would leave the only copy in ephemeral client state — a
|
||||
// reload during the pause would silently lose the user's message. New
|
||||
// steers are rejected while paused (enqueue is status-guarded), and the
|
||||
// requires_action TTL extension keeps the queue key alive.
|
||||
logger.debug(
|
||||
`[AgentClient] Paused ${streamId} for ${interrupt.payload.type} (action ${pendingAction.actionId})`,
|
||||
);
|
||||
}
|
||||
|
||||
async chatCompletion({ payload, userMCPAuthMap, abortController = null }) {
|
||||
|
|
|
|||
|
|
@ -754,6 +754,80 @@ describe('AgentClient - interrupt discovery persistence', () => {
|
|||
await GenerationJobManager.destroy();
|
||||
});
|
||||
|
||||
it('stages an event-actor interrupt until its signed suspension is durable', async () => {
|
||||
const streamId = 'conversation-event-actor-staged-pause';
|
||||
const job = await GenerationJobManager.createJob(streamId, 'user-123', streamId);
|
||||
const client = new AgentClient({
|
||||
req: {
|
||||
user: { id: 'user-123' },
|
||||
body: { endpoint: EModelEndpoint.agents, agent_id: 'agent-123' },
|
||||
config: { endpoints: { [EModelEndpoint.agents]: {} } },
|
||||
},
|
||||
res: {},
|
||||
agent: {
|
||||
id: 'agent-123',
|
||||
endpoint: EModelEndpoint.openAI,
|
||||
provider: EModelEndpoint.openAI,
|
||||
model_parameters: { model: 'gpt-4' },
|
||||
},
|
||||
contentParts: [],
|
||||
collectedUsage: [],
|
||||
artifactPromises: [],
|
||||
});
|
||||
client.conversationId = streamId;
|
||||
client.responseMessageId = 'response-event-actor-pause';
|
||||
client.jobCreatedAt = job.createdAt;
|
||||
client.eventActorInvocationId = 'event-pause';
|
||||
|
||||
await client.handleRunInterrupt(
|
||||
{
|
||||
getInterrupt: () => ({
|
||||
interruptId: 'interrupt-event-actor',
|
||||
threadId: streamId,
|
||||
payload: {
|
||||
type: 'ask_user_question',
|
||||
question: { question: 'Proceed?' },
|
||||
},
|
||||
}),
|
||||
getDiscoveredTools: () => ['save_issue_mcp_linear'],
|
||||
},
|
||||
streamId,
|
||||
);
|
||||
|
||||
await expect(GenerationJobManager.getJobStatus(streamId)).resolves.toBe('running');
|
||||
expect(client.pendingApproval).toBeUndefined();
|
||||
expect(client.readEventActorSuspension()).toMatchObject({
|
||||
actionId: expect.any(String),
|
||||
jobCreatedAt: job.createdAt,
|
||||
interrupt: {
|
||||
id: 'interrupt-event-actor',
|
||||
payload: {
|
||||
type: 'ask_user_question',
|
||||
actionId: expect.any(String),
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
await expect(
|
||||
client.publishStagedApproval({ version: 1, suspensionId: 'signed-suspension', attempt: 0 }),
|
||||
).resolves.toBe(true);
|
||||
await expect(GenerationJobManager.getJobStatus(streamId)).resolves.toBe('requires_action');
|
||||
await expect(GenerationJobManager.getJob(streamId)).resolves.toMatchObject({
|
||||
metadata: {
|
||||
agentEventSuspension: {
|
||||
version: 1,
|
||||
suspensionId: 'signed-suspension',
|
||||
attempt: 0,
|
||||
},
|
||||
},
|
||||
});
|
||||
expect(client.pendingApproval).toMatchObject({ actionId: expect.any(String) });
|
||||
expect(client.pendingRequestReleased).toBeFalsy();
|
||||
|
||||
await client.exposePendingApproval();
|
||||
expect(client.pendingRequestReleased).toBe(true);
|
||||
});
|
||||
|
||||
it('makes the run discovery snapshot durable when the run pauses', async () => {
|
||||
const streamId = 'conversation-discovered-pause';
|
||||
const job = await GenerationJobManager.createJob(streamId, 'user-123', streamId);
|
||||
|
|
|
|||
|
|
@ -51,6 +51,7 @@ const {
|
|||
getConvo,
|
||||
getAgentEventActorSnapshot,
|
||||
commitAgentEventActorState,
|
||||
storeAgentEventActorSuspension,
|
||||
beginAgentEventActorLegacyTurn,
|
||||
completeAgentEventActorLegacyTurn,
|
||||
recordAgentEventActorReconciliation,
|
||||
|
|
@ -1725,6 +1726,10 @@ 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. */
|
||||
durableEventActorSuspensions: generationProtocolVersion >= GENERATION_PROTOCOL_V2,
|
||||
checkpointerType: agentsConfig?.checkpointer?.type,
|
||||
expectedActionMayDetach: eventActorActionMayDetach,
|
||||
});
|
||||
|
|
@ -2087,10 +2092,12 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
|
|||
client?.run?.getRunSteps?.() ?? [],
|
||||
client?.contentParts ?? [],
|
||||
),
|
||||
readSuspension: () => client.readEventActorSuspension(),
|
||||
},
|
||||
{
|
||||
getSnapshot: getAgentEventActorSnapshot,
|
||||
commitState: commitAgentEventActorState,
|
||||
storeSuspension: storeAgentEventActorSuspension,
|
||||
recordReconciliation: recordAgentEventActorReconciliation,
|
||||
resolveReconciliation: resolveAgentEventActorReconciliation,
|
||||
admitAction: admitAgentEventActorAction,
|
||||
|
|
@ -2099,7 +2106,7 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
|
|||
getReceipt: getAgentEventActorReceipt,
|
||||
clearReconciliation: clearAgentEventActorReconciliation,
|
||||
},
|
||||
).then(({ value, execution }) => {
|
||||
).then(async ({ value, execution }) => {
|
||||
if (execution.status === 'applied') {
|
||||
appliedEventActor = {
|
||||
invocationId: eventTaskId,
|
||||
|
|
@ -2107,6 +2114,10 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
|
|||
checkpoint: execution.head.checkpoint,
|
||||
action: execution.result.action,
|
||||
};
|
||||
} else if (execution.status === 'suspended') {
|
||||
if (!(await client.publishStagedApproval(execution.suspension))) {
|
||||
throw new Error('Event actor suspension could not be projected to its job');
|
||||
}
|
||||
}
|
||||
logger.info('[event-actor] Bound child event completed', {
|
||||
conversationId,
|
||||
|
|
@ -2313,6 +2324,7 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
|
|||
}
|
||||
throw pausePersistenceError;
|
||||
}
|
||||
await client.exposePendingApproval?.();
|
||||
const released = await GenerationJobManager.approvals.finishPausePersistence(
|
||||
streamId,
|
||||
pauseActionId,
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
const { randomUUID } = require('crypto');
|
||||
const { isDeepStrictEqual } = require('util');
|
||||
const { logger } = require('@librechat/data-schemas');
|
||||
const {
|
||||
Constants,
|
||||
|
|
@ -39,6 +40,9 @@ const {
|
|||
createMCPRuntimeRequestBody,
|
||||
getSafeErrorMetadata,
|
||||
isAgentEventRetentionActive,
|
||||
resumeAgentEventActor,
|
||||
createAgentEventActionRecorder,
|
||||
findAgentEventAppliedAction,
|
||||
} = require('@librechat/api');
|
||||
const { disposeClient } = require('~/server/cleanup');
|
||||
const { decryptMetadata } = require('~/server/services/ActionService');
|
||||
|
|
@ -57,6 +61,12 @@ const {
|
|||
getUserMemories,
|
||||
getRoleByName,
|
||||
isSubagentOwnerAdmissible,
|
||||
getAgentEventActorSnapshot,
|
||||
commitAgentEventActorState,
|
||||
storeAgentEventActorSuspension,
|
||||
claimAgentEventActorSuspension,
|
||||
settleAgentEventActorSuspension,
|
||||
recordAgentEventActorReconciliation,
|
||||
completeAgentEventActorLegacyTurn,
|
||||
} = require('~/models');
|
||||
const {
|
||||
|
|
@ -92,6 +102,25 @@ function sendGenerationJson(res, status, body, generationProtocolVersion) {
|
|||
*/
|
||||
const STEER_RESUME_SETUP_TIMEOUT_MS = 1000;
|
||||
|
||||
function deferred() {
|
||||
let resolve;
|
||||
let reject;
|
||||
const promise = new Promise((resolvePromise, rejectPromise) => {
|
||||
resolve = resolvePromise;
|
||||
reject = rejectPromise;
|
||||
});
|
||||
return { promise, resolve, reject };
|
||||
}
|
||||
|
||||
function getSuspendedEventActorExpectedAction(suspension) {
|
||||
const payload = suspension?.interrupt?.payload;
|
||||
const expectedAction =
|
||||
payload != null && typeof payload === 'object' && !Array.isArray(payload)
|
||||
? payload._librechatEventActor?.expectedAction
|
||||
: undefined;
|
||||
return expectedAction != null && typeof expectedAction === 'object' ? expectedAction : undefined;
|
||||
}
|
||||
|
||||
/**
|
||||
* New jobs are physically isolated by an immutable saver namespace, so a
|
||||
* terminal owner deletes the whole namespace and catches writes that landed
|
||||
|
|
@ -347,6 +376,7 @@ async function finalizeResumedTurn({
|
|||
conversationId,
|
||||
addTitle,
|
||||
checkpointGeneration,
|
||||
appliedEventActor,
|
||||
}) {
|
||||
const userId = req.user.id;
|
||||
const checkpointerCfg = req.config?.endpoints?.[EModelEndpoint.agents]?.checkpointer;
|
||||
|
|
@ -498,6 +528,26 @@ async function finalizeResumedTurn({
|
|||
if (!savedResponseMessage) {
|
||||
throw new Error('Resumed response could not be persisted before terminal publication');
|
||||
}
|
||||
if (appliedEventActor != null) {
|
||||
const recorded = await recordAgentEventActorReconciliation({
|
||||
user: userId,
|
||||
conversationId,
|
||||
...(req._agentEventBindingTenantId == null
|
||||
? {}
|
||||
: { tenantId: req._agentEventBindingTenantId }),
|
||||
reconciliation: {
|
||||
invocationId: appliedEventActor.invocationId,
|
||||
actionAdmitted: true,
|
||||
status: 'history_persisted',
|
||||
checkpoint: appliedEventActor.checkpoint,
|
||||
action: appliedEventActor.action,
|
||||
observedAt: new Date(),
|
||||
},
|
||||
});
|
||||
if (!recorded) {
|
||||
throw new Error('Resumed event actor history barrier could not be durably recorded');
|
||||
}
|
||||
}
|
||||
/** The response row is now the durable history barrier for the resumed
|
||||
* legacy turn. Seal its exact pre-pause token before publishing FINAL; a
|
||||
* failed seal remains fail-closed and is recovered by the bounded path. */
|
||||
|
|
@ -1145,6 +1195,12 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
|
|||
|
||||
let releaseEventChildLease;
|
||||
let eventLeaseTransferredToRun = false;
|
||||
let durableEventActorSuspension;
|
||||
let eventActorResumePromise;
|
||||
let eventActorStartGate;
|
||||
let eventActorContinuationStarted = false;
|
||||
let eventActorActionRecorder;
|
||||
let appliedEventActor;
|
||||
const providerExecutionId = randomUUID();
|
||||
try {
|
||||
if (req._agentEventBindingParentConversationId != null) {
|
||||
|
|
@ -1243,6 +1299,74 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
|
|||
await decrementPendingRequest(userId);
|
||||
return sendGenerationJson(res, 409, eventActorRejection, generationProtocolVersion);
|
||||
}
|
||||
/** Missing means this pause was produced by a pre-durable-suspension
|
||||
* replica and must retain the legacy resume path during rolling deploys.
|
||||
* Presence opts the job into the fail-closed, Conversation-authoritative
|
||||
* protocol; malformed or stale markers never downgrade to legacy. */
|
||||
const suspensionProjection = job.metadata?.agentEventSuspension;
|
||||
if (suspensionProjection != null) {
|
||||
const projectionValid =
|
||||
suspensionProjection.version === 1 &&
|
||||
typeof suspensionProjection.suspensionId === 'string' &&
|
||||
suspensionProjection.suspensionId.length > 0 &&
|
||||
Number.isSafeInteger(suspensionProjection.attempt) &&
|
||||
suspensionProjection.attempt >= 0;
|
||||
const actorSnapshot = projectionValid
|
||||
? await getAgentEventActorSnapshot({
|
||||
user: userId,
|
||||
conversationId,
|
||||
...(req._agentEventBindingTenantId == null
|
||||
? {}
|
||||
: { tenantId: req._agentEventBindingTenantId }),
|
||||
})
|
||||
: undefined;
|
||||
const suspensionRecord = actorSnapshot?.suspension;
|
||||
if (
|
||||
projectionValid &&
|
||||
suspensionRecord?.status === 'pending' &&
|
||||
suspensionRecord.actionId === pendingAction.actionId &&
|
||||
suspensionRecord.jobCreatedAt === job.createdAt &&
|
||||
suspensionRecord.suspension.suspensionId === suspensionProjection.suspensionId &&
|
||||
suspensionRecord.suspension.attempt === suspensionProjection.attempt
|
||||
) {
|
||||
durableEventActorSuspension = suspensionRecord.suspension;
|
||||
const signedExpectedAction = getSuspendedEventActorExpectedAction(
|
||||
durableEventActorSuspension,
|
||||
);
|
||||
if (
|
||||
job.metadata.agentEventExpectedAction != null &&
|
||||
!isDeepStrictEqual(signedExpectedAction, job.metadata.agentEventExpectedAction)
|
||||
) {
|
||||
const currentJob = await GenerationJobManager.getJob(streamId).catch(() => null);
|
||||
await rollbackUnconsumedScheduleClaim(currentJob);
|
||||
await releaseScheduleFence();
|
||||
await decrementPendingRequest(userId);
|
||||
return sendGenerationJson(
|
||||
res,
|
||||
409,
|
||||
{
|
||||
code: 'EVENT_ACTOR_SUSPENSION_STALE',
|
||||
error: 'This event actor action is no longer current',
|
||||
},
|
||||
generationProtocolVersion,
|
||||
);
|
||||
}
|
||||
} else {
|
||||
const currentJob = await GenerationJobManager.getJob(streamId).catch(() => null);
|
||||
await rollbackUnconsumedScheduleClaim(currentJob);
|
||||
await releaseScheduleFence();
|
||||
await decrementPendingRequest(userId);
|
||||
return sendGenerationJson(
|
||||
res,
|
||||
409,
|
||||
{
|
||||
code: 'EVENT_ACTOR_SUSPENSION_STALE',
|
||||
error: 'This event actor action is no longer current',
|
||||
},
|
||||
generationProtocolVersion,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Atomically claim the resume. The single winner drives the run; a racing second
|
||||
|
|
@ -1253,18 +1377,12 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
|
|||
// that releases it, so a store/Redis error here (unlike the clean `!claimed` branch)
|
||||
// would leak the concurrency slot until the counter TTL expires — spuriously 429'ing
|
||||
// the user when they retry the still-paused approval. Release the slot on that path too.
|
||||
let claimed;
|
||||
try {
|
||||
/** The CAS that reopens steering must also publish THIS owner's seal
|
||||
* capability. A separate write after status=`running` leaves a window in
|
||||
* which steer/arm requests read the previous replica's capability. */
|
||||
claimed = await GenerationJobManager.approvals.resolve(
|
||||
const claimJobApproval = () =>
|
||||
GenerationJobManager.approvals.resolve(
|
||||
streamId,
|
||||
pendingAction.actionId,
|
||||
{
|
||||
preemptCapable: isSteerPreemptSupported(),
|
||||
// The handover owner's quote handling replaces the previous
|
||||
// replica's flag, mirroring `preemptCapable` above.
|
||||
steerQuotesCapable: true,
|
||||
providerExecutionId,
|
||||
providerDrained: true,
|
||||
|
|
@ -1272,6 +1390,90 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
|
|||
},
|
||||
job.createdAt,
|
||||
);
|
||||
let claimed;
|
||||
try {
|
||||
/** The CAS that reopens steering must also publish THIS owner's seal
|
||||
* capability. A separate write after status=`running` leaves a window in
|
||||
* which steer/arm requests read the previous replica's capability. */
|
||||
if (durableEventActorSuspension == null) {
|
||||
claimed = await claimJobApproval();
|
||||
} else {
|
||||
const claimGate = deferred();
|
||||
eventActorStartGate = deferred();
|
||||
const expectedAction = getSuspendedEventActorExpectedAction(durableEventActorSuspension);
|
||||
eventActorActionRecorder = createAgentEventActionRecorder(expectedAction);
|
||||
req._agentEventActionObserver = eventActorActionRecorder.observeToolEnd;
|
||||
eventActorResumePromise = resumeAgentEventActor(
|
||||
{
|
||||
user: userId,
|
||||
conversationId,
|
||||
...(req._agentEventBindingTenantId == null
|
||||
? {}
|
||||
: { tenantId: req._agentEventBindingTenantId }),
|
||||
bindingId: req._agentEventBindingId,
|
||||
suspension: durableEventActorSuspension,
|
||||
/** One identity spans the Conversation claim and the job's
|
||||
* provider-owner CAS. A terminal hook can therefore prove whether
|
||||
* an abort won before or after the resume projection. */
|
||||
resumeAttemptId: providerExecutionId,
|
||||
resumeValue: mapped.resumeValue,
|
||||
signal: job.abortController.signal,
|
||||
checkpointer: checkpointerCfg,
|
||||
expectedAction,
|
||||
claimProjection: async () => {
|
||||
try {
|
||||
const projected = await claimJobApproval();
|
||||
claimGate.resolve(projected);
|
||||
return projected;
|
||||
} catch (error) {
|
||||
/** Redis can commit its CAS and lose only the reply. Read back
|
||||
* this exact resume capability before declaring the earlier
|
||||
* Conversation claim orphaned. */
|
||||
const currentJob = await GenerationJobManager.getJob(streamId).catch(() => null);
|
||||
if (
|
||||
currentJob?.createdAt === job.createdAt &&
|
||||
currentJob.status === 'running' &&
|
||||
currentJob.metadata?.providerExecutionId === providerExecutionId
|
||||
) {
|
||||
claimGate.resolve(true);
|
||||
return true;
|
||||
}
|
||||
claimGate.reject(error);
|
||||
throw error;
|
||||
}
|
||||
},
|
||||
resume: async (actorContext) => {
|
||||
const start = await eventActorStartGate.promise;
|
||||
return start(actorContext);
|
||||
},
|
||||
readAppliedAction: () =>
|
||||
eventActorActionRecorder.read() ??
|
||||
findAgentEventAppliedAction(
|
||||
expectedAction,
|
||||
client?.run?.getRunSteps?.() ?? [],
|
||||
client?.contentParts ?? [],
|
||||
{ userSubmittedMessageFieldPaths },
|
||||
),
|
||||
readSuspension: () => client?.readEventActorSuspension(),
|
||||
readResultContext: () => client?.getEventActorContext(),
|
||||
},
|
||||
{
|
||||
getSnapshot: getAgentEventActorSnapshot,
|
||||
commitState: commitAgentEventActorState,
|
||||
storeSuspension: storeAgentEventActorSuspension,
|
||||
claimSuspension: claimAgentEventActorSuspension,
|
||||
settleSuspension: settleAgentEventActorSuspension,
|
||||
recordReconciliation: recordAgentEventActorReconciliation,
|
||||
},
|
||||
);
|
||||
eventActorResumePromise.catch(() => {});
|
||||
claimed = await Promise.race([
|
||||
claimGate.promise,
|
||||
eventActorResumePromise.then(() => {
|
||||
throw new Error('Event actor suspension completed before claiming its job projection');
|
||||
}),
|
||||
]);
|
||||
}
|
||||
} catch (err) {
|
||||
const currentJob = await GenerationJobManager.getJob(streamId).catch(() => null);
|
||||
await rollbackUnconsumedScheduleClaim(currentJob);
|
||||
|
|
@ -1460,17 +1662,6 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
|
|||
let pausePersistenceFailed = false;
|
||||
let pausePersistenceFailureFinalized = false;
|
||||
try {
|
||||
if (
|
||||
!(await GenerationJobManager.beginProviderExecution(
|
||||
streamId,
|
||||
job.createdAt,
|
||||
providerExecutionId,
|
||||
))
|
||||
) {
|
||||
throw Object.assign(new Error('Generation stopped before provider resume'), {
|
||||
code: 'RUN_REPLACED',
|
||||
});
|
||||
}
|
||||
if (userSubmittedPaths.length > 0) {
|
||||
job.metadata.userSubmittedPaths = userSubmittedPaths;
|
||||
}
|
||||
|
|
@ -1520,22 +1711,58 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
|
|||
GenerationJobManager.setContentParts(streamId, client.contentParts, job.createdAt);
|
||||
}
|
||||
|
||||
await client.resumeCompletion({
|
||||
resumeValue: mapped.resumeValue,
|
||||
seedContent,
|
||||
runSteps: resumeState?.runSteps ?? [],
|
||||
storedMessages,
|
||||
abortController: job.abortController,
|
||||
// Carry the user's MCP auth so approved MCP tools run with their credentials.
|
||||
userMCPAuthMap: result.userMCPAuthMap,
|
||||
// Replay deferred tools discovered before the pause (captured at pause). The rebuilt
|
||||
// graph passes `messages: []`, so without these the model would lose their schemas.
|
||||
discoveredToolNames: job.metadata?.discoveredTools,
|
||||
activityPhaseSnapshot: job.metadata?.activityPhaseSnapshot,
|
||||
});
|
||||
const resumeClient = () =>
|
||||
client.resumeCompletion({
|
||||
resumeValue: mapped.resumeValue,
|
||||
seedContent,
|
||||
runSteps: resumeState?.runSteps ?? [],
|
||||
storedMessages,
|
||||
abortController: job.abortController,
|
||||
// Carry the user's MCP auth so approved MCP tools run with their credentials.
|
||||
userMCPAuthMap: result.userMCPAuthMap,
|
||||
// Replay deferred tools discovered before the pause (captured at pause). The rebuilt
|
||||
// graph passes `messages: []`, so without these the model would lose their schemas.
|
||||
discoveredToolNames: job.metadata?.discoveredTools,
|
||||
activityPhaseSnapshot: job.metadata?.activityPhaseSnapshot,
|
||||
});
|
||||
if (
|
||||
!(await GenerationJobManager.beginProviderExecution(
|
||||
streamId,
|
||||
job.createdAt,
|
||||
providerExecutionId,
|
||||
))
|
||||
) {
|
||||
throw Object.assign(new Error('Generation stopped before provider resume'), {
|
||||
code: 'RUN_REPLACED',
|
||||
});
|
||||
}
|
||||
if (eventActorResumePromise == null) {
|
||||
await resumeClient();
|
||||
} else {
|
||||
eventActorContinuationStarted = true;
|
||||
eventActorStartGate.resolve(async (actorContext) => {
|
||||
client.checkpointNamespace = actorContext.checkpointNamespace;
|
||||
client.eventActorCheckpointId = actorContext.checkpointId;
|
||||
client.eventActorInvocationId = actorContext.invocationId;
|
||||
client.eventActorContinuation = actorContext.continuation;
|
||||
return resumeClient();
|
||||
});
|
||||
const actorResult = await eventActorResumePromise;
|
||||
if (actorResult.execution.status === 'suspended') {
|
||||
if (!(await client.publishStagedApproval(actorResult.execution.suspension))) {
|
||||
throw new Error('Re-paused event actor suspension could not be projected to its job');
|
||||
}
|
||||
} else if (actorResult.execution.status === 'applied') {
|
||||
appliedEventActor = {
|
||||
invocationId: durableEventActorSuspension.invocation.invocationId,
|
||||
checkpoint: actorResult.execution.head.checkpoint,
|
||||
action: actorResult.execution.result.action,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
// The model may pause AGAIN (another tool, or a follow-up question). The pending
|
||||
// action is already persisted + emitted; leave the job `requires_action`.
|
||||
// action is durably projected; persist progress before exposing it to clients.
|
||||
if (client.pendingApproval) {
|
||||
logger.debug(`[ResumeAgentController] Re-paused for approval: ${streamId}`);
|
||||
const pauseActionId = client.pendingApproval.actionId;
|
||||
|
|
@ -1576,6 +1803,7 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
|
|||
}
|
||||
throw pausePersistenceError;
|
||||
}
|
||||
await client.exposePendingApproval?.();
|
||||
const released = await GenerationJobManager.approvals.finishPausePersistence(
|
||||
streamId,
|
||||
pauseActionId,
|
||||
|
|
@ -1635,8 +1863,17 @@ const ResumeAgentController = async (req, res, next, initializeClient, addTitle)
|
|||
conversationId,
|
||||
addTitle,
|
||||
checkpointGeneration,
|
||||
appliedEventActor,
|
||||
});
|
||||
} catch (err) {
|
||||
if (
|
||||
eventActorResumePromise != null &&
|
||||
eventActorStartGate != null &&
|
||||
!eventActorContinuationStarted
|
||||
) {
|
||||
eventActorStartGate.reject(err);
|
||||
await eventActorResumePromise.catch(() => {});
|
||||
}
|
||||
logger.error('[ResumeAgentController] Resume failed', getSafeErrorMetadata(err));
|
||||
if (pausePersistenceFailed) {
|
||||
// failPausePersistence already performed the exact requires_action ->
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue