mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-08-27 04:07:05 +00:00
🪢 fix: Persist Failed Agent Turns Before Error Publication (#14118)
This commit is contained in:
Danny Avila
GitHub
parent
a9d99b3771
commit
862ebf3235
4 changed files with 592 additions and 13 deletions
|
|
@ -63,6 +63,7 @@ const mockFilterPersistableAbortContent = jest.fn((content) =>
|
|||
const mockGetConvo = jest.fn();
|
||||
const mockGetMessages = jest.fn();
|
||||
const mockSaveMessage = jest.fn();
|
||||
const mockSaveConvo = jest.fn();
|
||||
const mockIsAgentTriggerPrincipalActive = jest.fn();
|
||||
const mockIsSubagentOwnerAdmissible = jest.fn();
|
||||
const mockAcquireEventChildGenerationLease = jest.fn();
|
||||
|
|
@ -224,6 +225,7 @@ jest.mock('~/cache', () => ({
|
|||
|
||||
jest.mock('~/models', () => ({
|
||||
saveMessage: (...args) => mockSaveMessage(...args),
|
||||
saveConvo: (...args) => mockSaveConvo(...args),
|
||||
getMessages: (...args) => mockGetMessages(...args),
|
||||
getConvo: (...args) => mockGetConvo(...args),
|
||||
isAgentTriggerPrincipalActive: (...args) => mockIsAgentTriggerPrincipalActive(...args),
|
||||
|
|
@ -319,7 +321,12 @@ describe('ResumableAgentController resume metadata', () => {
|
|||
}),
|
||||
);
|
||||
mockGenerationJobManager.finishTerminalJob.mockResolvedValue(undefined);
|
||||
mockGenerationJobManager.completeJob.mockResolvedValue(true);
|
||||
mockGenerationJobManager.completeJob.mockImplementation(
|
||||
async (_streamId, _error, _createdAt, options) => {
|
||||
await options?.beforeErrorPublication?.();
|
||||
return true;
|
||||
},
|
||||
);
|
||||
mockGenerationJobManager.beginProviderExecution.mockResolvedValue(true);
|
||||
mockGenerationJobManager.markProviderExecutionDrained.mockResolvedValue(true);
|
||||
mockGenerationJobManager.failPausePersistence.mockResolvedValue(true);
|
||||
|
|
@ -334,6 +341,7 @@ describe('ResumableAgentController resume metadata', () => {
|
|||
mockGenerationJobManager.steering.park.mockResolvedValue(undefined);
|
||||
mockGenerationJobManager.steering.consumeRecovered.mockResolvedValue(true);
|
||||
mockSaveMessage.mockResolvedValue({});
|
||||
mockSaveConvo.mockResolvedValue({});
|
||||
mockDeleteAgentCheckpoint.mockResolvedValue(undefined);
|
||||
});
|
||||
|
||||
|
|
@ -1362,6 +1370,8 @@ describe('ResumableAgentController resume metadata', () => {
|
|||
|
||||
await AgentController(req, res, jest.fn(), initializeClient, null);
|
||||
expect(allSubscribersLeftHandler).toEqual(expect.any(Function));
|
||||
mockSaveMessage.mockClear();
|
||||
mockSaveConvo.mockClear();
|
||||
|
||||
const oauthPart = {
|
||||
type: 'tool_call',
|
||||
|
|
@ -2289,6 +2299,7 @@ describe('ResumableAgentController resume metadata', () => {
|
|||
error: 'Attached resources could not be restored',
|
||||
}),
|
||||
1000,
|
||||
expect.objectContaining({ beforeErrorPublication: expect.any(Function) }),
|
||||
);
|
||||
});
|
||||
|
||||
|
|
@ -2326,6 +2337,7 @@ describe('ResumableAgentController resume metadata', () => {
|
|||
error: 'Stateful code environment is not allowed by this deployment: conversation',
|
||||
}),
|
||||
1000,
|
||||
expect.objectContaining({ beforeErrorPublication: expect.any(Function) }),
|
||||
);
|
||||
});
|
||||
|
||||
|
|
@ -2516,6 +2528,7 @@ describe('ResumableAgentController resume metadata', () => {
|
|||
'conversation-123',
|
||||
'provider init failed',
|
||||
1000,
|
||||
expect.objectContaining({ beforeErrorPublication: expect.any(Function) }),
|
||||
);
|
||||
});
|
||||
|
||||
|
|
@ -2552,6 +2565,7 @@ describe('ResumableAgentController resume metadata', () => {
|
|||
'conversation-123',
|
||||
'Recovered steer cannot skip user message persistence',
|
||||
1000,
|
||||
expect.objectContaining({ beforeErrorPublication: expect.any(Function) }),
|
||||
);
|
||||
});
|
||||
|
||||
|
|
@ -3024,6 +3038,247 @@ describe('ResumableAgentController resume metadata', () => {
|
|||
expect(mockGenerationJobManager.claimTerminalJob).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
describe('failed-turn persistence', () => {
|
||||
const conversationId = 'conversation-123';
|
||||
|
||||
const createFailedRequest = (bodyOverrides = {}) => ({
|
||||
user: { id: 'user-123' },
|
||||
body: {
|
||||
text: 'Hello with a removed model.',
|
||||
messageId: 'user-message',
|
||||
parentMessageId: 'prior-response',
|
||||
conversationId,
|
||||
endpointOption: {
|
||||
endpoint: 'azureOpenAI',
|
||||
modelOptions: { model: 'gpt-4o' },
|
||||
},
|
||||
...bodyOverrides,
|
||||
},
|
||||
config: {},
|
||||
});
|
||||
|
||||
async function flushBackgroundGeneration() {
|
||||
for (let i = 0; i < 10; i++) {
|
||||
await nextTick();
|
||||
}
|
||||
}
|
||||
|
||||
it('persists an initialization failure before terminal error publication', async () => {
|
||||
const events = [];
|
||||
mockSaveConvo.mockImplementation(async () => {
|
||||
events.push('turn-persisted');
|
||||
return {};
|
||||
});
|
||||
mockGenerationJobManager.completeJob.mockImplementation(
|
||||
async (_streamId, _error, _createdAt, options) => {
|
||||
await options.beforeErrorPublication();
|
||||
events.push('error-published');
|
||||
return true;
|
||||
},
|
||||
);
|
||||
const initializeClient = jest
|
||||
.fn()
|
||||
.mockRejectedValue(new Error('The model "gpt-4o" is not available.'));
|
||||
|
||||
await AgentController(
|
||||
createFailedRequest(),
|
||||
createResumableResponse(),
|
||||
jest.fn(),
|
||||
initializeClient,
|
||||
null,
|
||||
);
|
||||
|
||||
expect(mockSaveMessage).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ userId: 'user-123' }),
|
||||
expect.objectContaining({
|
||||
messageId: 'user-message',
|
||||
parentMessageId: 'prior-response',
|
||||
conversationId,
|
||||
text: 'Hello with a removed model.',
|
||||
isCreatedByUser: true,
|
||||
error: false,
|
||||
}),
|
||||
expect.any(Object),
|
||||
);
|
||||
expect(mockSaveMessage).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ userId: 'user-123' }),
|
||||
expect.objectContaining({
|
||||
messageId: 'user-message_',
|
||||
parentMessageId: 'user-message',
|
||||
conversationId,
|
||||
endpoint: 'azureOpenAI',
|
||||
model: 'gpt-4o',
|
||||
text: 'The model "gpt-4o" is not available.',
|
||||
error: true,
|
||||
isCreatedByUser: false,
|
||||
}),
|
||||
expect.any(Object),
|
||||
);
|
||||
expect(events).toEqual(['turn-persisted', 'error-published']);
|
||||
expect(mockSaveConvo).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ userId: 'user-123' }),
|
||||
{ conversationId },
|
||||
expect.objectContaining({ noUpsert: true }),
|
||||
);
|
||||
});
|
||||
|
||||
it('allows a follow-up to chain from the persisted failed response', async () => {
|
||||
const initializeClient = jest.fn().mockRejectedValue(new Error('model unavailable'));
|
||||
await AgentController(
|
||||
createFailedRequest(),
|
||||
createResumableResponse(),
|
||||
jest.fn(),
|
||||
initializeClient,
|
||||
null,
|
||||
);
|
||||
|
||||
expect(mockSaveMessage.mock.calls.map(([, message]) => message.messageId)).toContain(
|
||||
'user-message_',
|
||||
);
|
||||
mockGetMessages.mockResolvedValue([{ _id: 'persisted-error-turn' }]);
|
||||
const followUpRes = createResumableResponse();
|
||||
|
||||
await AgentController(
|
||||
createFailedRequest({
|
||||
text: 'Retry with a valid model.',
|
||||
messageId: 'follow-up-user',
|
||||
parentMessageId: 'user-message_',
|
||||
}),
|
||||
followUpRes,
|
||||
jest.fn(),
|
||||
initializeClient,
|
||||
null,
|
||||
);
|
||||
|
||||
expect(followUpRes.status).not.toHaveBeenCalledWith(409);
|
||||
expect(mockCheckAndIncrementPendingRequest).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it('persists failures raised before generation saves any message', async () => {
|
||||
const client = {
|
||||
options: {},
|
||||
sendMessage: jest.fn().mockRejectedValue(new Error('provider exploded')),
|
||||
};
|
||||
|
||||
await AgentController(
|
||||
createFailedRequest(),
|
||||
createResumableResponse(),
|
||||
jest.fn(),
|
||||
jest.fn().mockResolvedValue({ client }),
|
||||
null,
|
||||
);
|
||||
await flushBackgroundGeneration();
|
||||
|
||||
expect(mockSaveMessage).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ userId: 'user-123' }),
|
||||
expect.objectContaining({
|
||||
messageId: 'user-message_',
|
||||
text: 'provider exploded',
|
||||
error: true,
|
||||
}),
|
||||
expect.any(Object),
|
||||
);
|
||||
expect(mockGenerationJobManager.completeJob).toHaveBeenCalledWith(
|
||||
conversationId,
|
||||
'provider exploded',
|
||||
1000,
|
||||
expect.objectContaining({ beforeErrorPublication: expect.any(Function) }),
|
||||
);
|
||||
});
|
||||
|
||||
it('uses the live user identity after generation starts', async () => {
|
||||
const serverUserMessage = {
|
||||
messageId: 'server-user',
|
||||
parentMessageId: 'prior-response',
|
||||
conversationId,
|
||||
sender: 'User',
|
||||
text: 'Hello with a removed model.',
|
||||
isCreatedByUser: true,
|
||||
};
|
||||
const client = {
|
||||
options: {},
|
||||
sendMessage: jest.fn(async (_text, options) => {
|
||||
options.onStart(serverUserMessage, 'server-response-uuid');
|
||||
throw new Error('failed after onStart');
|
||||
}),
|
||||
};
|
||||
|
||||
await AgentController(
|
||||
createFailedRequest(),
|
||||
createResumableResponse(),
|
||||
jest.fn(),
|
||||
jest.fn().mockResolvedValue({ client }),
|
||||
null,
|
||||
);
|
||||
await flushBackgroundGeneration();
|
||||
|
||||
const savedIds = mockSaveMessage.mock.calls.map(([, message]) => message.messageId);
|
||||
expect(savedIds).toEqual(expect.arrayContaining(['server-user', 'server-user_']));
|
||||
expect(savedIds).not.toContain('user-message_');
|
||||
});
|
||||
|
||||
it('does not overwrite an existing response row', async () => {
|
||||
mockGetMessages.mockResolvedValue([{ _id: 'already-saved' }]);
|
||||
|
||||
await AgentController(
|
||||
createFailedRequest(),
|
||||
createResumableResponse(),
|
||||
jest.fn(),
|
||||
jest.fn().mockRejectedValue(new Error('late failure')),
|
||||
null,
|
||||
);
|
||||
|
||||
expect(mockSaveMessage).not.toHaveBeenCalled();
|
||||
expect(mockSaveConvo).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('creates the conversation row for a failed first turn', async () => {
|
||||
const res = createResumableResponse();
|
||||
mockGenerationJobManager.claimGeneration.mockImplementation(
|
||||
async (_userId, _clientRequestId, streamId, claimedConversationId) =>
|
||||
wonGenerationClaim({ streamId, conversationId: claimedConversationId }),
|
||||
);
|
||||
const req = createFailedRequest({
|
||||
conversationId: undefined,
|
||||
clientRequestId: 'failed-new-conversation',
|
||||
parentMessageId: '00000000-0000-0000-0000-000000000000',
|
||||
endpointOption: {
|
||||
endpoint: 'azureOpenAI',
|
||||
modelOptions: { model: 'gpt-4o' },
|
||||
chatProjectId: '507f1f77bcf86cd799439011',
|
||||
},
|
||||
});
|
||||
|
||||
await AgentController(
|
||||
req,
|
||||
res,
|
||||
jest.fn(),
|
||||
jest.fn().mockRejectedValue(new Error('model unavailable')),
|
||||
null,
|
||||
);
|
||||
|
||||
const mintedConversationId = res.json.mock.calls[0][0].conversationId;
|
||||
expect(mockSaveMessage).toHaveBeenCalledWith(
|
||||
expect.any(Object),
|
||||
expect.objectContaining({
|
||||
messageId: 'user-message_',
|
||||
conversationId: mintedConversationId,
|
||||
}),
|
||||
expect.any(Object),
|
||||
);
|
||||
expect(mockSaveConvo).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ userId: 'user-123' }),
|
||||
expect.objectContaining({
|
||||
conversationId: mintedConversationId,
|
||||
endpoint: 'azureOpenAI',
|
||||
model: 'gpt-4o',
|
||||
chatProjectId: '507f1f77bcf86cd799439011',
|
||||
}),
|
||||
expect.any(Object),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
it('finalizes the failed job before releasing the idempotency claim', async () => {
|
||||
mockGenerationJobManager.claimGeneration.mockResolvedValue(wonGenerationClaim());
|
||||
const initializeClient = jest.fn().mockRejectedValue(new Error('init boom after res.json'));
|
||||
|
|
@ -3046,6 +3301,7 @@ describe('ResumableAgentController resume metadata', () => {
|
|||
'conversation-123',
|
||||
expect.any(String),
|
||||
1000,
|
||||
expect.objectContaining({ beforeErrorPublication: expect.any(Function) }),
|
||||
);
|
||||
expect(mockGenerationJobManager.releaseGeneration).toHaveBeenCalledWith(
|
||||
'user-123',
|
||||
|
|
@ -3112,6 +3368,7 @@ describe('ResumableAgentController resume metadata', () => {
|
|||
'conversation-123',
|
||||
'init boom after res.json',
|
||||
1000,
|
||||
expect.objectContaining({ beforeErrorPublication: expect.any(Function) }),
|
||||
);
|
||||
expect(mockGenerationJobManager.releaseGeneration).toHaveBeenCalledWith(
|
||||
'user-123',
|
||||
|
|
@ -3227,6 +3484,7 @@ describe('ResumableAgentController resume metadata', () => {
|
|||
'conversation-123',
|
||||
generationError.message,
|
||||
1000,
|
||||
expect.objectContaining({ beforeErrorPublication: expect.any(Function) }),
|
||||
);
|
||||
expect(mockGenerationJobManager.completeJob.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
mockDecrementPendingRequest.mock.invocationCallOrder[0],
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ const { logViolation } = require('~/cache');
|
|||
const { recordScheduleOutcome, isScheduleLive } = require('~/server/services/Schedules');
|
||||
const {
|
||||
saveMessage,
|
||||
saveConvo,
|
||||
getMessages,
|
||||
getConvo,
|
||||
isAgentTriggerPrincipalActive,
|
||||
|
|
@ -107,6 +108,18 @@ async function attachConversationCreatedAt(req, conversationId, conversationAnch
|
|||
}
|
||||
}
|
||||
|
||||
function getPreliminaryResponseMessageId({ messageId, responseMessageId }) {
|
||||
if (typeof responseMessageId === 'string' && responseMessageId.length > 0) {
|
||||
return responseMessageId;
|
||||
}
|
||||
|
||||
if (typeof messageId !== 'string' || messageId.length === 0) {
|
||||
return null;
|
||||
}
|
||||
|
||||
return `${messageId.replace(/_+$/, '')}_`;
|
||||
}
|
||||
|
||||
function getPreliminaryUserMessage(
|
||||
{ messageId, parentMessageId, text, quotes, files, manualSkills, alwaysAppliedSkills },
|
||||
conversationId,
|
||||
|
|
@ -190,6 +203,165 @@ async function finishResumableRequest(req, userId) {
|
|||
}
|
||||
}
|
||||
|
||||
async function saveErrorTurn(
|
||||
req,
|
||||
{
|
||||
conversationId,
|
||||
endpointOption,
|
||||
isNewConvo,
|
||||
errorText,
|
||||
liveUserMessage,
|
||||
liveResponseMessageId,
|
||||
sender,
|
||||
},
|
||||
) {
|
||||
try {
|
||||
const { isContinued, isRegenerate, editedContent, responseMessageId, overrideParentMessageId } =
|
||||
req.body ?? {};
|
||||
if (
|
||||
isContinued ||
|
||||
editedContent != null ||
|
||||
(responseMessageId && !isRegenerate) ||
|
||||
req.body?.recoverySteerId != null ||
|
||||
req.body?.clientRequestId?.startsWith?.('steer-recovery:') === true
|
||||
) {
|
||||
return;
|
||||
}
|
||||
|
||||
let userMessage = null;
|
||||
let errorMessageId = null;
|
||||
let errorParentMessageId = null;
|
||||
if (isRegenerate) {
|
||||
errorMessageId =
|
||||
typeof responseMessageId === 'string' && responseMessageId.length > 0
|
||||
? responseMessageId
|
||||
: null;
|
||||
errorParentMessageId = liveUserMessage?.messageId ?? overrideParentMessageId ?? null;
|
||||
} else {
|
||||
userMessage =
|
||||
liveUserMessage != null
|
||||
? {
|
||||
...liveUserMessage,
|
||||
...(liveUserMessage.files == null &&
|
||||
Array.isArray(req.body?.files) &&
|
||||
req.body.files.length > 0 && { files: req.body.files }),
|
||||
...(liveUserMessage.manualSkills == null &&
|
||||
Array.isArray(req.body?.manualSkills) &&
|
||||
req.body.manualSkills.length > 0 && { manualSkills: req.body.manualSkills }),
|
||||
...(liveUserMessage.alwaysAppliedSkills == null &&
|
||||
Array.isArray(req.body?.alwaysAppliedSkills) &&
|
||||
req.body.alwaysAppliedSkills.length > 0 && {
|
||||
alwaysAppliedSkills: req.body.alwaysAppliedSkills,
|
||||
}),
|
||||
}
|
||||
: getPreliminaryUserMessage(req.body, conversationId);
|
||||
if (!userMessage) {
|
||||
return;
|
||||
}
|
||||
errorMessageId = getPreliminaryResponseMessageId(
|
||||
liveUserMessage != null ? { messageId: liveUserMessage.messageId } : req.body,
|
||||
);
|
||||
errorParentMessageId = userMessage.messageId;
|
||||
}
|
||||
if (!errorMessageId || !errorParentMessageId) {
|
||||
return;
|
||||
}
|
||||
|
||||
const userId = req.user.id;
|
||||
const existing = await getMessages(
|
||||
{ user: userId, messageId: errorMessageId, conversationId },
|
||||
'_id',
|
||||
);
|
||||
if (existing.length > 0) {
|
||||
return;
|
||||
}
|
||||
if (liveResponseMessageId != null && liveResponseMessageId !== errorMessageId) {
|
||||
const partial = await getMessages(
|
||||
{ user: userId, messageId: liveResponseMessageId, conversationId },
|
||||
'_id',
|
||||
);
|
||||
if (partial.length > 0) {
|
||||
return;
|
||||
}
|
||||
}
|
||||
|
||||
const reqCtx = {
|
||||
userId,
|
||||
isTemporary: req?._agentEventBindingRetention?.isTemporary ?? req?.body?.isTemporary,
|
||||
expiredAt: req?._agentEventBindingRetention?.expiredAt,
|
||||
interfaceConfig: req?.config?.interfaceConfig,
|
||||
};
|
||||
const context = 'api/server/controllers/agents/request.js - failed turn';
|
||||
const endpoint = endpointOption?.endpoint;
|
||||
const model = getAgentResponseModel(req, endpointOption);
|
||||
const iconURL = getEndpointIconURL(req, endpointOption);
|
||||
|
||||
if (userMessage) {
|
||||
const savedUserMessage = await saveMessage(
|
||||
reqCtx,
|
||||
{
|
||||
...userMessage,
|
||||
user: userId,
|
||||
sender: 'User',
|
||||
isCreatedByUser: true,
|
||||
error: false,
|
||||
unfinished: false,
|
||||
},
|
||||
{ context },
|
||||
);
|
||||
if (!savedUserMessage) {
|
||||
throw new Error('Failed user message could not be persisted');
|
||||
}
|
||||
}
|
||||
const savedErrorMessage = await saveMessage(
|
||||
reqCtx,
|
||||
{
|
||||
messageId: errorMessageId,
|
||||
conversationId,
|
||||
parentMessageId: errorParentMessageId,
|
||||
sender: sender ?? 'AI',
|
||||
...(endpoint != null && { endpoint }),
|
||||
...(model != null && { model }),
|
||||
...(iconURL != null && { iconURL }),
|
||||
user: userId,
|
||||
text: errorText,
|
||||
error: true,
|
||||
unfinished: false,
|
||||
isCreatedByUser: false,
|
||||
},
|
||||
{ context },
|
||||
);
|
||||
if (!savedErrorMessage) {
|
||||
throw new Error('Failed response message could not be persisted');
|
||||
}
|
||||
|
||||
const agentId = endpointOption?.agent_id ?? req.body?.agent_id;
|
||||
const chatProjectId = endpointOption?.chatProjectId ?? req.body?.chatProjectId;
|
||||
const seedConvo = isNewConvo || req.resolvedConversation === null;
|
||||
const convoFields = seedConvo
|
||||
? {
|
||||
...(endpoint != null && { endpoint }),
|
||||
...(endpointOption?.endpointType != null && {
|
||||
endpointType: endpointOption.endpointType,
|
||||
}),
|
||||
...(model != null && { model }),
|
||||
...(iconURL != null && { iconURL }),
|
||||
...(endpointOption?.spec != null && { spec: endpointOption.spec }),
|
||||
...(agentId != null && { agent_id: agentId }),
|
||||
...(typeof chatProjectId === 'string' && chatProjectId.length > 0 && { chatProjectId }),
|
||||
}
|
||||
: {};
|
||||
await saveConvo(
|
||||
reqCtx,
|
||||
{ conversationId, ...convoFields },
|
||||
seedConvo ? { context } : { context, noUpsert: true },
|
||||
);
|
||||
} catch (err) {
|
||||
logger.error('[AgentController] Failed to persist error turn', err);
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
function classifyScheduledFailure(error, aborted = false) {
|
||||
if (aborted || error?.code === 'SCHEDULE_NO_LONGER_ACTIVE') {
|
||||
return { status: 'interrupted', error: error?.message };
|
||||
|
|
@ -1450,11 +1622,15 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
|
|||
}
|
||||
|
||||
let userMessage;
|
||||
let liveResponseMessageId = preallocatedResponseMessageId;
|
||||
|
||||
const getReqData = (data = {}) => {
|
||||
if (data.userMessage) {
|
||||
userMessage = data.userMessage;
|
||||
}
|
||||
if (data.responseMessageId) {
|
||||
liveResponseMessageId = data.responseMessageId;
|
||||
}
|
||||
// conversationId is pre-generated, no need to update from callback
|
||||
};
|
||||
|
||||
|
|
@ -1593,6 +1769,7 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
|
|||
try {
|
||||
const onStart = (userMsg, respMsgId, _isNewConvo) => {
|
||||
userMessage = userMsg;
|
||||
liveResponseMessageId = respMsgId;
|
||||
|
||||
// Store userMessage and responseMessageId upfront for resume capability
|
||||
GenerationJobManager.updateMetadata(
|
||||
|
|
@ -2179,8 +2356,18 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
|
|||
// completeJob first wins running -> error and atomically parks
|
||||
// steers, then publishes. A competing abort/pause emits nothing.
|
||||
ownsScheduledFailure =
|
||||
(await GenerationJobManager.completeJob(streamId, generationError, jobCreatedAt)) ===
|
||||
true;
|
||||
(await GenerationJobManager.completeJob(streamId, generationError, jobCreatedAt, {
|
||||
beforeErrorPublication: () =>
|
||||
saveErrorTurn(req, {
|
||||
conversationId,
|
||||
endpointOption,
|
||||
isNewConvo,
|
||||
errorText: generationError,
|
||||
liveUserMessage: userMessage,
|
||||
liveResponseMessageId,
|
||||
sender: client?.sender,
|
||||
}),
|
||||
})) === true;
|
||||
} catch (completeErr) {
|
||||
logger.warn(
|
||||
'[ResumableAgentController] completeJob failed during generation-error cleanup',
|
||||
|
|
@ -2262,6 +2449,7 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
|
|||
} catch (error) {
|
||||
logger.error('[ResumableAgentController] Initialization error:', error);
|
||||
const initializationFailure = getInitializationFailure(error);
|
||||
const streamStarted = res.headersSent;
|
||||
try {
|
||||
if (!res.headersSent) {
|
||||
if (error?.code === 'GENERATION_PREDECESSOR_MISMATCH') {
|
||||
|
|
@ -2354,16 +2542,25 @@ const ResumableAgentController = async (req, res, next, initializeClient, addTit
|
|||
const initializationError = initializationFailure
|
||||
? JSON.stringify(initializationFailure)
|
||||
: error.message || 'Failed to start generation';
|
||||
const completionPromise = streamStarted
|
||||
? GenerationJobManager.completeJob(streamId, initializationError, jobCreatedAt, {
|
||||
beforeErrorPublication: () =>
|
||||
saveErrorTurn(req, {
|
||||
conversationId,
|
||||
endpointOption,
|
||||
isNewConvo,
|
||||
errorText: initializationError,
|
||||
}),
|
||||
})
|
||||
: GenerationJobManager.completeJob(streamId, initializationError, jobCreatedAt);
|
||||
initializationFinalized =
|
||||
(await GenerationJobManager.completeJob(streamId, initializationError, jobCreatedAt).catch(
|
||||
(completeErr) => {
|
||||
logger.warn(
|
||||
'[ResumableAgentController] completeJob failed during init-error cleanup',
|
||||
completeErr,
|
||||
);
|
||||
return false;
|
||||
},
|
||||
)) === true;
|
||||
(await completionPromise.catch((completeErr) => {
|
||||
logger.warn(
|
||||
'[ResumableAgentController] completeJob failed during init-error cleanup',
|
||||
completeErr,
|
||||
);
|
||||
return false;
|
||||
})) === true;
|
||||
}
|
||||
if (initializationFinalized && !scheduleTerminalOutcomeRecorded) {
|
||||
await settleScheduledRun(classifyScheduledFailure(error));
|
||||
|
|
|
|||
|
|
@ -710,6 +710,10 @@ class GenerationJobManagerClass {
|
|||
* reconnect can replay that authoritative final payload. */
|
||||
private terminalPublicationFailures = new WeakSet<TerminalJobClaim>();
|
||||
|
||||
/** Persistence-pending error claims whose terminal output was already
|
||||
* reconciled by a competing owner or stale-owner recovery. */
|
||||
private terminalErrorPublicationSuppressions = new WeakSet<TerminalJobClaim>();
|
||||
|
||||
private cleanupInterval: NodeJS.Timeout | null = null;
|
||||
|
||||
/** Generation-scoped retirement callbacks must not outlive the configured
|
||||
|
|
@ -3535,7 +3539,7 @@ class GenerationJobManagerClass {
|
|||
// Error jobs stay durable long enough for late subscribers to receive the
|
||||
// stored error. A publication failure must never bypass the finally cleanup.
|
||||
try {
|
||||
if (status === 'error') {
|
||||
if (status === 'error' && !this.terminalErrorPublicationSuppressions.has(claim)) {
|
||||
const terminalError = error ?? 'Generation failed';
|
||||
if (runtime) {
|
||||
runtime.errorEvent = terminalError;
|
||||
|
|
@ -3606,6 +3610,7 @@ class GenerationJobManagerClass {
|
|||
|
||||
this.releaseJobOwnership(streamId, createdAt);
|
||||
this.terminalPublicationFailures.delete(claim);
|
||||
this.terminalErrorPublicationSuppressions.delete(claim);
|
||||
let metricStatus: 'completed' | 'error' | 'aborted' = 'aborted';
|
||||
if (status === 'complete') {
|
||||
metricStatus = 'completed';
|
||||
|
|
@ -3640,16 +3645,55 @@ class GenerationJobManagerClass {
|
|||
streamId: string,
|
||||
error?: string,
|
||||
expectedCreatedAt?: number,
|
||||
options: { beforeErrorPublication?: () => Promise<void> } = {},
|
||||
): Promise<boolean> {
|
||||
const beforeErrorPublication = error ? options.beforeErrorPublication : undefined;
|
||||
const claim = await this.claimTerminalJob(
|
||||
streamId,
|
||||
error ? 'error' : 'complete',
|
||||
error,
|
||||
expectedCreatedAt,
|
||||
beforeErrorPublication ? { persistencePending: true } : undefined,
|
||||
);
|
||||
if (!claim) {
|
||||
return false;
|
||||
}
|
||||
|
||||
if (beforeErrorPublication) {
|
||||
let persistenceFinalized = false;
|
||||
try {
|
||||
await beforeErrorPublication();
|
||||
persistenceFinalized = await this.jobStore.finalizeTerminalPersistence(
|
||||
streamId,
|
||||
claim.createdAt,
|
||||
JSON.stringify(
|
||||
buildTerminalPersistenceReconcile({
|
||||
createdAt: claim.createdAt,
|
||||
conversationId: claim.conversationId,
|
||||
status: claim.status,
|
||||
}),
|
||||
),
|
||||
);
|
||||
} catch (persistenceError) {
|
||||
logger.error(
|
||||
`[GenerationJobManager] Failed required error persistence for ${streamId}:`,
|
||||
persistenceError,
|
||||
);
|
||||
try {
|
||||
await this.publishTerminalClaim(claim, null);
|
||||
} catch (publishError) {
|
||||
logger.error(
|
||||
`[GenerationJobManager] Failed to publish error persistence reconciliation for ${streamId}:`,
|
||||
publishError,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
if (!persistenceFinalized) {
|
||||
this.terminalErrorPublicationSuppressions.add(claim);
|
||||
}
|
||||
}
|
||||
|
||||
await this.finishTerminalJob(claim);
|
||||
return true;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2783,6 +2783,86 @@ describe('GenerationJobManager startup telemetry', () => {
|
|||
await manager.destroy();
|
||||
});
|
||||
|
||||
it('holds terminal error publication until required persistence finishes', async () => {
|
||||
const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60_000 });
|
||||
const manager = new GenerationJobManagerClass();
|
||||
manager.configure({
|
||||
jobStore,
|
||||
eventTransport: new InMemoryEventTransport(),
|
||||
isRedis: false,
|
||||
cleanupOnComplete: false,
|
||||
});
|
||||
manager.initialize();
|
||||
const streamId = 'stream-error-persistence-barrier';
|
||||
const job = await manager.createJob(streamId, 'user-1');
|
||||
const onError = jest.fn();
|
||||
const subscription = await manager.subscribe(streamId, () => undefined, undefined, onError);
|
||||
let releasePersistence!: () => void;
|
||||
const persistence = new Promise<void>((resolve) => {
|
||||
releasePersistence = resolve;
|
||||
});
|
||||
|
||||
const completing = manager.completeJob(streamId, 'initialization failed', job.createdAt, {
|
||||
beforeErrorPublication: () => persistence,
|
||||
});
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
await expect(jobStore.getJob(streamId)).resolves.toMatchObject({
|
||||
status: 'error',
|
||||
error: 'initialization failed',
|
||||
terminalPersistencePending: true,
|
||||
});
|
||||
|
||||
releasePersistence();
|
||||
await expect(completing).resolves.toBe(true);
|
||||
|
||||
expect(onError).toHaveBeenCalledWith('initialization failed');
|
||||
await expect(jobStore.getJob(streamId)).resolves.toMatchObject({
|
||||
status: 'error',
|
||||
error: 'initialization failed',
|
||||
terminalPersistencePending: false,
|
||||
});
|
||||
subscription?.unsubscribe();
|
||||
await manager.destroy();
|
||||
});
|
||||
|
||||
it('publishes reconciliation when required error persistence fails', async () => {
|
||||
const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60_000 });
|
||||
const manager = new GenerationJobManagerClass();
|
||||
manager.configure({
|
||||
jobStore,
|
||||
eventTransport: new InMemoryEventTransport(),
|
||||
isRedis: false,
|
||||
cleanupOnComplete: false,
|
||||
});
|
||||
manager.initialize();
|
||||
const streamId = 'stream-error-persistence-fails';
|
||||
const job = await manager.createJob(streamId, 'user-1');
|
||||
const onDone = jest.fn();
|
||||
const onError = jest.fn();
|
||||
const subscription = await manager.subscribe(streamId, () => undefined, onDone, onError);
|
||||
|
||||
await expect(
|
||||
manager.completeJob(streamId, 'initialization failed', job.createdAt, {
|
||||
beforeErrorPublication: async () => {
|
||||
throw new Error('message store unavailable');
|
||||
},
|
||||
}),
|
||||
).resolves.toBe(true);
|
||||
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
expect(onDone).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
final: true,
|
||||
reconcile: true,
|
||||
terminalStatus: 'error',
|
||||
}),
|
||||
);
|
||||
subscription?.unsubscribe();
|
||||
await manager.destroy();
|
||||
});
|
||||
|
||||
it('atomically terminalizes a paused job when post-HITL persistence fails', async () => {
|
||||
const jobStore = new InMemoryJobStore({ ttlAfterComplete: 60_000 });
|
||||
const manager = new GenerationJobManagerClass();
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue