mirror of
https://github.com/danny-avila/LibreChat.git
synced 2026-09-21 15:45:22 +00:00
🛎️ fix: Enroll Remote Agent Runs in the Generation Lifecycle (#15349)
* fix: enroll remote agent runs in generation lifecycle * fix: close remote lifecycle ownership gaps * fix: close remote conversation drain races * fix: fence remote runs during conversation deletion * fix: reconcile remote deletion and settlement races * fix: complete owner deletion recovery * fix: preserve remote cleanup receipts * fix: consume deletion receipts before cleanup * fix: expose idempotent deletion option * chore: sort remote lifecycle imports
This commit is contained in:
Danny Avila
GitHub
parent
8fcab7e44f
commit
70f735336d
22 changed files with 1550 additions and 184 deletions
|
|
@ -26,6 +26,27 @@ const mockCompletionUsage = {
|
|||
subagent: { prompt_tokens: 25, completion_tokens: 10, total_tokens: 35 },
|
||||
};
|
||||
const mockBuildCompletionUsage = jest.fn().mockReturnValue(mockCompletionUsage);
|
||||
const mockEnrollAgentExecution = jest.fn();
|
||||
let mockExecution;
|
||||
|
||||
function resetMockExecution() {
|
||||
const controller = new AbortController();
|
||||
mockExecution = {
|
||||
signal: controller.signal,
|
||||
abort: jest.fn((reason) => controller.abort(reason)),
|
||||
track: jest.fn((promise) => promise),
|
||||
beginProviderExecution: jest.fn(async () => {
|
||||
if (controller.signal.aborted) {
|
||||
throw Object.assign(new Error('request disconnected'), {
|
||||
code: 'RUN_REPLACED',
|
||||
status: 409,
|
||||
});
|
||||
}
|
||||
}),
|
||||
settle: jest.fn().mockResolvedValue(undefined),
|
||||
};
|
||||
mockEnrollAgentExecution.mockResolvedValue(mockExecution);
|
||||
}
|
||||
const mockInitialSessions = new Map([['execute_code', { session_id: 'seeded' }]]);
|
||||
const mockGetSafeErrorMetadata = jest.fn((error) => {
|
||||
const status = error?.status ?? error?.statusCode ?? error?.response?.status;
|
||||
|
|
@ -274,6 +295,12 @@ jest.mock('@librechat/api', () => ({
|
|||
userMCPAuthMap: undefined,
|
||||
}),
|
||||
resolveSubagentGraphs: jest.fn().mockResolvedValue(undefined),
|
||||
enrollAgentExecution: (...args) => mockEnrollAgentExecution(...args),
|
||||
waitForAgentExecutionWrites: async (writes) => {
|
||||
const results = await Promise.allSettled(writes);
|
||||
const failure = results.find((result) => result.status === 'rejected');
|
||||
if (failure?.status === 'rejected') throw failure.reason;
|
||||
},
|
||||
}));
|
||||
|
||||
jest.mock('~/server/controllers/ModelController', () => ({
|
||||
|
|
@ -365,6 +392,7 @@ jest.mock('~/models', () => ({
|
|||
getConvoFiles: jest.fn().mockResolvedValue([]),
|
||||
getFormattedMemories: jest.fn().mockResolvedValue({ withKeys: '', withoutKeys: '' }),
|
||||
getConvo: jest.fn().mockResolvedValue(null),
|
||||
isSubagentOwnerAdmissible: jest.fn().mockResolvedValue(true),
|
||||
}));
|
||||
|
||||
describe('OpenAIChatCompletionController', () => {
|
||||
|
|
@ -373,6 +401,7 @@ describe('OpenAIChatCompletionController', () => {
|
|||
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks();
|
||||
resetMockExecution();
|
||||
|
||||
const controller = require('../openai');
|
||||
OpenAIChatCompletionController = controller.OpenAIChatCompletionController;
|
||||
|
|
@ -389,7 +418,8 @@ describe('OpenAIChatCompletionController', () => {
|
|||
agents: { allowedProviders: ['openAI'] },
|
||||
},
|
||||
},
|
||||
on: jest.fn(),
|
||||
once: jest.fn(),
|
||||
off: jest.fn(),
|
||||
};
|
||||
|
||||
res = {
|
||||
|
|
@ -399,9 +429,85 @@ describe('OpenAIChatCompletionController', () => {
|
|||
flushHeaders: jest.fn(),
|
||||
end: jest.fn(),
|
||||
write: jest.fn(),
|
||||
once: jest.fn(),
|
||||
off: jest.fn(),
|
||||
};
|
||||
});
|
||||
|
||||
it('enrolls, starts, and settles the remote execution lifecycle', async () => {
|
||||
await OpenAIChatCompletionController(req, res);
|
||||
|
||||
expect(mockEnrollAgentExecution).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
runId: 'chatcmpl-mock-nanoid-123',
|
||||
userId: 'user-123',
|
||||
agentId: 'agent-123',
|
||||
protocol: 'chat.completions',
|
||||
}),
|
||||
);
|
||||
expect(mockExecution.beginProviderExecution).toHaveBeenCalledTimes(1);
|
||||
expect(mockExecution.beginProviderExecution.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
require('@librechat/api').initializeAgent.mock.invocationCallOrder[0],
|
||||
);
|
||||
expect(mockExecution.beginProviderExecution.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
mockProcessStream.mock.invocationCallOrder[0],
|
||||
);
|
||||
expect(mockExecution.settle).toHaveBeenCalledWith(undefined);
|
||||
expect(res.once).toHaveBeenCalledWith('close', expect.any(Function));
|
||||
expect(res.off).toHaveBeenCalledWith('close', expect.any(Function));
|
||||
});
|
||||
|
||||
it('covers artifact writes when provider execution fails', async () => {
|
||||
const providerError = new Error('provider aborted');
|
||||
const artifactWrite = Promise.resolve(null);
|
||||
const { createToolEndCallback } = require('~/server/controllers/agents/callbacks');
|
||||
createToolEndCallback.mockImplementationOnce(({ artifactPromises }) => {
|
||||
artifactPromises.push(artifactWrite);
|
||||
return jest.fn();
|
||||
});
|
||||
mockProcessStream.mockRejectedValueOnce(providerError);
|
||||
|
||||
await OpenAIChatCompletionController(req, res);
|
||||
|
||||
expect(mockExecution.track).toHaveBeenCalledWith(expect.any(Promise));
|
||||
expect(mockExecution.track.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
mockExecution.settle.mock.invocationCallOrder[0],
|
||||
);
|
||||
expect(mockExecution.settle).toHaveBeenCalledWith(providerError);
|
||||
});
|
||||
|
||||
it('does not initialize a provider after disconnecting during enrollment', async () => {
|
||||
let finishEnrollment;
|
||||
mockEnrollAgentExecution.mockImplementationOnce(
|
||||
() =>
|
||||
new Promise((resolve) => {
|
||||
finishEnrollment = resolve;
|
||||
}),
|
||||
);
|
||||
|
||||
const request = OpenAIChatCompletionController(req, res);
|
||||
await Promise.resolve();
|
||||
res.once.mock.calls[0][1]();
|
||||
finishEnrollment(mockExecution);
|
||||
await request;
|
||||
|
||||
expect(mockExecution.abort).toHaveBeenCalledTimes(1);
|
||||
expect(mockExecution.beginProviderExecution).toHaveBeenCalledTimes(1);
|
||||
expect(require('@librechat/api').initializeAgent).not.toHaveBeenCalled();
|
||||
expect(mockExecution.settle).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ code: 'RUN_REPLACED' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('does not treat a consumed request stream as a response disconnect', async () => {
|
||||
req.destroyed = true;
|
||||
|
||||
await OpenAIChatCompletionController(req, res);
|
||||
|
||||
expect(mockExecution.abort).not.toHaveBeenCalled();
|
||||
expect(mockExecution.beginProviderExecution).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('resolves saved graph subagents for remote chat-completion runs', async () => {
|
||||
const {
|
||||
initializeAgent,
|
||||
|
|
|
|||
|
|
@ -148,6 +148,27 @@ const mockResponsesUsage = {
|
|||
subagent: { input_tokens: 25, output_tokens: 10, total_tokens: 35 },
|
||||
};
|
||||
const mockBuildResponsesUsage = jest.fn().mockReturnValue(mockResponsesUsage);
|
||||
const mockEnrollAgentExecution = jest.fn();
|
||||
let mockExecution;
|
||||
|
||||
function resetMockExecution() {
|
||||
const controller = new AbortController();
|
||||
mockExecution = {
|
||||
signal: controller.signal,
|
||||
abort: jest.fn((reason) => controller.abort(reason)),
|
||||
track: jest.fn((promise) => promise),
|
||||
beginProviderExecution: jest.fn(async () => {
|
||||
if (controller.signal.aborted) {
|
||||
throw Object.assign(new Error('request disconnected'), {
|
||||
code: 'RUN_REPLACED',
|
||||
status: 409,
|
||||
});
|
||||
}
|
||||
}),
|
||||
settle: jest.fn().mockResolvedValue(undefined),
|
||||
};
|
||||
mockEnrollAgentExecution.mockResolvedValue(mockExecution);
|
||||
}
|
||||
|
||||
jest.mock('nanoid', () => ({
|
||||
nanoid: jest.fn(() => 'mock-nanoid-123'),
|
||||
|
|
@ -340,6 +361,12 @@ jest.mock('@librechat/api', () => ({
|
|||
on_run_step_delta: { handle: jest.fn() },
|
||||
on_chat_model_end: { handle: jest.fn() },
|
||||
}),
|
||||
enrollAgentExecution: (...args) => mockEnrollAgentExecution(...args),
|
||||
waitForAgentExecutionWrites: async (writes) => {
|
||||
const results = await Promise.allSettled(writes);
|
||||
const failure = results.find((result) => result.status === 'rejected');
|
||||
if (failure?.status === 'rejected') throw failure.reason;
|
||||
},
|
||||
}));
|
||||
|
||||
jest.mock('~/server/services/ToolService', () => ({
|
||||
|
|
@ -441,6 +468,7 @@ jest.mock('~/models', () => ({
|
|||
getFormattedMemories: jest.fn().mockResolvedValue({ withKeys: '', withoutKeys: '' }),
|
||||
saveConvo: jest.fn().mockResolvedValue({}),
|
||||
getConvo: jest.fn().mockResolvedValue(null),
|
||||
isSubagentOwnerAdmissible: jest.fn().mockResolvedValue(true),
|
||||
}));
|
||||
|
||||
let mockGlobalDiscoveredAgentConfigs = null;
|
||||
|
|
@ -451,6 +479,7 @@ describe('createResponse controller', () => {
|
|||
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks();
|
||||
resetMockExecution();
|
||||
mockGlobalDiscoveredAgentConfigs = null;
|
||||
require('@librechat/api').inspectContent.mockReset().mockReturnValue(null);
|
||||
|
||||
|
|
@ -469,7 +498,8 @@ describe('createResponse controller', () => {
|
|||
agents: { allowedProviders: ['anthropic'] },
|
||||
},
|
||||
},
|
||||
on: jest.fn(),
|
||||
once: jest.fn(),
|
||||
off: jest.fn(),
|
||||
};
|
||||
|
||||
res = {
|
||||
|
|
@ -479,9 +509,89 @@ describe('createResponse controller', () => {
|
|||
flushHeaders: jest.fn(),
|
||||
end: jest.fn(),
|
||||
write: jest.fn(),
|
||||
once: jest.fn(),
|
||||
off: jest.fn(),
|
||||
};
|
||||
});
|
||||
|
||||
it('enrolls, starts, and settles the remote execution lifecycle', async () => {
|
||||
await createResponse(req, res);
|
||||
|
||||
expect(mockEnrollAgentExecution).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
runId: 'resp_mock-123',
|
||||
userId: 'user-123',
|
||||
agentId: 'agent-123',
|
||||
protocol: 'responses',
|
||||
}),
|
||||
);
|
||||
const { createRun } = require('@librechat/api');
|
||||
const processStream = await createRun.mock.results.at(-1).value;
|
||||
expect(mockExecution.beginProviderExecution).toHaveBeenCalledTimes(1);
|
||||
expect(mockExecution.beginProviderExecution.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
require('@librechat/api').initializeAgent.mock.invocationCallOrder[0],
|
||||
);
|
||||
expect(mockExecution.beginProviderExecution.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
processStream.processStream.mock.invocationCallOrder[0],
|
||||
);
|
||||
expect(mockExecution.settle).toHaveBeenCalledWith(undefined);
|
||||
expect(res.once).toHaveBeenCalledWith('close', expect.any(Function));
|
||||
expect(res.off).toHaveBeenCalledWith('close', expect.any(Function));
|
||||
});
|
||||
|
||||
it('covers artifact writes when provider execution fails', async () => {
|
||||
const providerError = new Error('provider aborted');
|
||||
const artifactWrite = Promise.resolve(null);
|
||||
const processStream = jest.fn().mockRejectedValue(providerError);
|
||||
const { createRun } = require('@librechat/api');
|
||||
const { createToolEndCallback } = require('~/server/controllers/agents/callbacks');
|
||||
createRun.mockResolvedValueOnce({ processStream });
|
||||
createToolEndCallback.mockImplementationOnce(({ artifactPromises }) => {
|
||||
artifactPromises.push(artifactWrite);
|
||||
return jest.fn();
|
||||
});
|
||||
|
||||
await createResponse(req, res);
|
||||
|
||||
expect(mockExecution.track).toHaveBeenCalledWith(expect.any(Promise));
|
||||
expect(mockExecution.track.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
mockExecution.settle.mock.invocationCallOrder[0],
|
||||
);
|
||||
expect(mockExecution.settle).toHaveBeenCalledWith(providerError);
|
||||
});
|
||||
|
||||
it('does not initialize a provider after disconnecting during enrollment', async () => {
|
||||
let finishEnrollment;
|
||||
mockEnrollAgentExecution.mockImplementationOnce(
|
||||
() =>
|
||||
new Promise((resolve) => {
|
||||
finishEnrollment = resolve;
|
||||
}),
|
||||
);
|
||||
|
||||
const request = createResponse(req, res);
|
||||
await Promise.resolve();
|
||||
res.once.mock.calls[0][1]();
|
||||
finishEnrollment(mockExecution);
|
||||
await request;
|
||||
|
||||
expect(mockExecution.abort).toHaveBeenCalledTimes(1);
|
||||
expect(mockExecution.beginProviderExecution).toHaveBeenCalledTimes(1);
|
||||
expect(require('@librechat/api').initializeAgent).not.toHaveBeenCalled();
|
||||
expect(mockExecution.settle).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ code: 'RUN_REPLACED' }),
|
||||
);
|
||||
});
|
||||
|
||||
it('does not treat a consumed request stream as a response disconnect', async () => {
|
||||
req.destroyed = true;
|
||||
|
||||
await createResponse(req, res);
|
||||
|
||||
expect(mockExecution.abort).not.toHaveBeenCalled();
|
||||
expect(mockExecution.beginProviderExecution).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('resolves saved graph subagents for remote Responses API runs', async () => {
|
||||
const { initializeAgent, resolveSubagentGraphs } = require('@librechat/api');
|
||||
const primaryConfig = {
|
||||
|
|
|
|||
|
|
@ -60,6 +60,8 @@ const {
|
|||
createOpenAIContentAggregator,
|
||||
isChatCompletionValidationFailure,
|
||||
stripActivityLabelParts,
|
||||
enrollAgentExecution,
|
||||
waitForAgentExecutionWrites,
|
||||
} = require('@librechat/api');
|
||||
const {
|
||||
buildSummarizationHandlers,
|
||||
|
|
@ -345,18 +347,40 @@ const executeOpenAIChatCompletion = async (envelope, { req, res }) => {
|
|||
`[OpenAI API] Response ${responseId} started for agent ${agentId}, stream: ${request.stream}`,
|
||||
);
|
||||
|
||||
// Set up abort controller
|
||||
const abortController = new AbortController();
|
||||
|
||||
// Handle client disconnect
|
||||
req.on('close', () => {
|
||||
if (!abortController.signal.aborted) {
|
||||
abortController.abort();
|
||||
const conversationId = request.conversation_id ?? nanoid();
|
||||
let execution;
|
||||
let executionError;
|
||||
let responseClosed = res.destroyed === true && res.writableEnded !== true;
|
||||
/** @type {Promise<import('librechat-data-provider').TAttachment | null>[]} */
|
||||
const artifactPromises = [];
|
||||
let artifactWritesCovered = false;
|
||||
const abortOnResponseClose = () => {
|
||||
if (res.writableEnded === true) {
|
||||
return;
|
||||
}
|
||||
responseClosed = true;
|
||||
if (execution && !execution.signal.aborted) {
|
||||
execution.abort();
|
||||
logger.debug('[OpenAI API] Client disconnected, aborting');
|
||||
}
|
||||
});
|
||||
|
||||
};
|
||||
res.once('close', abortOnResponseClose);
|
||||
try {
|
||||
execution = await enrollAgentExecution({
|
||||
runId: responseId,
|
||||
userId: principal.userId,
|
||||
conversationId,
|
||||
agentId,
|
||||
protocol: 'chat.completions',
|
||||
/** Conversation delete-all uses the shared owner-admission fence. Remote
|
||||
* execution must observe it after durable enrollment and before provider work. */
|
||||
isPrincipalActive: db.isSubagentOwnerAdmissible,
|
||||
});
|
||||
if (responseClosed || (res.destroyed === true && res.writableEnded !== true)) {
|
||||
execution.abort();
|
||||
}
|
||||
await execution.beginProviderExecution();
|
||||
|
||||
if (request.conversation_id != null) {
|
||||
if (typeof request.conversation_id !== 'string') {
|
||||
return sendErrorResponse(
|
||||
|
|
@ -371,7 +395,6 @@ const executeOpenAIChatCompletion = async (envelope, { req, res }) => {
|
|||
}
|
||||
}
|
||||
|
||||
const conversationId = request.conversation_id ?? nanoid();
|
||||
const parentMessageId = request.parent_message_id ?? null;
|
||||
let mcpParentMessageId;
|
||||
if (typeof request.parent_message_id === 'string' && request.parent_message_id.trim() !== '') {
|
||||
|
|
@ -389,7 +412,7 @@ const executeOpenAIChatCompletion = async (envelope, { req, res }) => {
|
|||
const allowedProviders = new Set(agentsEConfig?.allowedProviders);
|
||||
|
||||
// Create tool loader
|
||||
const loadTools = createToolLoader(abortController.signal);
|
||||
const loadTools = createToolLoader(execution.signal);
|
||||
|
||||
// Initialize the agent first to check for disableStreaming
|
||||
const endpointOption = {
|
||||
|
|
@ -683,9 +706,6 @@ const executeOpenAIChatCompletion = async (envelope, { req, res }) => {
|
|||
: null;
|
||||
|
||||
const collectedUsage = [];
|
||||
/** @type {Promise<import('librechat-data-provider').TAttachment | null>[]} */
|
||||
const artifactPromises = [];
|
||||
|
||||
const toolEndCallback = createToolEndCallback({ req, res, artifactPromises, streamId: null });
|
||||
|
||||
/* Stable for the turn: the primary prime list is fixed once
|
||||
|
|
@ -706,7 +726,7 @@ const executeOpenAIChatCompletion = async (envelope, { req, res }) => {
|
|||
requestBody: mcpRequestBody,
|
||||
toolNames,
|
||||
agent: ctx.agent ?? agent,
|
||||
signal: abortController.signal,
|
||||
signal: execution.signal,
|
||||
toolRegistry: ctx.toolRegistry,
|
||||
callerCapabilityProjection,
|
||||
backgroundToolNames: ctx.backgroundToolNames,
|
||||
|
|
@ -988,7 +1008,7 @@ const executeOpenAIChatCompletion = async (envelope, { req, res }) => {
|
|||
runId: responseId,
|
||||
summarizationConfig,
|
||||
appConfig,
|
||||
signal: abortController.signal,
|
||||
signal: execution.signal,
|
||||
customHandlers: handlers,
|
||||
requestBody: mcpRequestBody,
|
||||
user: { id: userId },
|
||||
|
|
@ -1012,7 +1032,7 @@ const executeOpenAIChatCompletion = async (envelope, { req, res }) => {
|
|||
...(userMCPAuthMap != null && { userMCPAuthMap }),
|
||||
},
|
||||
recursionLimit: resolveRecursionLimit(agentsEConfig, agent),
|
||||
signal: abortController.signal,
|
||||
signal: execution.signal,
|
||||
streamMode: 'values',
|
||||
version: 'v2',
|
||||
};
|
||||
|
|
@ -1028,28 +1048,30 @@ const executeOpenAIChatCompletion = async (envelope, { req, res }) => {
|
|||
// Record token usage against balance
|
||||
const balanceConfig = getBalanceConfig(appConfig);
|
||||
const transactionsConfig = getTransactionsConfig(appConfig);
|
||||
recordCollectedUsage(
|
||||
{
|
||||
spendTokens: db.spendTokens,
|
||||
spendStructuredTokens: db.spendStructuredTokens,
|
||||
pricing: { getMultiplier: db.getMultiplier, getCacheMultiplier: db.getCacheMultiplier },
|
||||
bulkWriteOps: { insertMany: db.bulkInsertTransactions, updateBalance: db.updateBalance },
|
||||
},
|
||||
{
|
||||
user: userId,
|
||||
conversationId,
|
||||
collectedUsage,
|
||||
context: 'message',
|
||||
messageId: responseId,
|
||||
balance: balanceConfig,
|
||||
transactions: transactionsConfig,
|
||||
model: primaryConfig.model || agent.model_parameters?.model,
|
||||
endpointTokenConfig: primaryConfig.endpointTokenConfig,
|
||||
resolveEndpointTokenConfig,
|
||||
},
|
||||
).catch((err) => {
|
||||
logger.error('[OpenAI API] Error recording usage:', getSafeErrorMetadata(err));
|
||||
});
|
||||
execution.track(
|
||||
recordCollectedUsage(
|
||||
{
|
||||
spendTokens: db.spendTokens,
|
||||
spendStructuredTokens: db.spendStructuredTokens,
|
||||
pricing: { getMultiplier: db.getMultiplier, getCacheMultiplier: db.getCacheMultiplier },
|
||||
bulkWriteOps: { insertMany: db.bulkInsertTransactions, updateBalance: db.updateBalance },
|
||||
},
|
||||
{
|
||||
user: userId,
|
||||
conversationId,
|
||||
collectedUsage,
|
||||
context: 'message',
|
||||
messageId: responseId,
|
||||
balance: balanceConfig,
|
||||
transactions: transactionsConfig,
|
||||
model: primaryConfig.model || agent.model_parameters?.model,
|
||||
endpointTokenConfig: primaryConfig.endpointTokenConfig,
|
||||
resolveEndpointTokenConfig,
|
||||
},
|
||||
).catch((err) => {
|
||||
logger.error('[OpenAI API] Error recording usage:', getSafeErrorMetadata(err));
|
||||
}),
|
||||
);
|
||||
|
||||
const usage = buildCompletionUsage(collectedUsage);
|
||||
|
||||
|
|
@ -1060,26 +1082,30 @@ const executeOpenAIChatCompletion = async (envelope, { req, res }) => {
|
|||
res.end();
|
||||
logger.debug(`[OpenAI API] Response ${responseId} completed in ${duration}ms (streaming)`);
|
||||
|
||||
// Wait for artifact processing after response ends (non-blocking)
|
||||
// The HTTP response is complete, while destructive cleanup still waits for artifacts.
|
||||
if (artifactPromises.length > 0) {
|
||||
Promise.all(artifactPromises).catch((artifactError) => {
|
||||
logger.warn(
|
||||
'[OpenAI API] Error processing artifacts:',
|
||||
getSafeErrorMetadata(artifactError),
|
||||
);
|
||||
});
|
||||
execution.track(
|
||||
waitForAgentExecutionWrites(artifactPromises).catch((artifactError) => {
|
||||
logger.warn(
|
||||
'[OpenAI API] Error processing artifacts:',
|
||||
getSafeErrorMetadata(artifactError),
|
||||
);
|
||||
}),
|
||||
);
|
||||
artifactWritesCovered = true;
|
||||
}
|
||||
} else {
|
||||
// For non-streaming, wait for artifacts before sending response
|
||||
if (artifactPromises.length > 0) {
|
||||
try {
|
||||
await Promise.all(artifactPromises);
|
||||
await waitForAgentExecutionWrites(artifactPromises);
|
||||
} catch (artifactError) {
|
||||
logger.warn(
|
||||
'[OpenAI API] Error processing artifacts:',
|
||||
getSafeErrorMetadata(artifactError),
|
||||
);
|
||||
}
|
||||
artifactWritesCovered = true;
|
||||
}
|
||||
|
||||
const response = buildNonStreamingResponse(
|
||||
|
|
@ -1095,6 +1121,7 @@ const executeOpenAIChatCompletion = async (envelope, { req, res }) => {
|
|||
);
|
||||
}
|
||||
} catch (error) {
|
||||
executionError = error;
|
||||
logger.error('[OpenAI API] Error:', getSafeErrorMetadata(error));
|
||||
const protectionEnabled = hasModelBoundContentProtection(
|
||||
appConfig?.filters,
|
||||
|
|
@ -1129,6 +1156,23 @@ const executeOpenAIChatCompletion = async (envelope, { req, res }) => {
|
|||
const errorCode = !protectionEnabled && typeof error?.code === 'string' ? error.code : null;
|
||||
sendErrorResponse(res, statusCode, errorMessage, errorType, errorCode);
|
||||
}
|
||||
} finally {
|
||||
res.off('close', abortOnResponseClose);
|
||||
if (execution) {
|
||||
if (!artifactWritesCovered && artifactPromises.length > 0) {
|
||||
execution.track(
|
||||
waitForAgentExecutionWrites(artifactPromises).catch((artifactError) => {
|
||||
logger.warn(
|
||||
'[OpenAI API] Error processing artifacts:',
|
||||
getSafeErrorMetadata(artifactError),
|
||||
);
|
||||
}),
|
||||
);
|
||||
}
|
||||
await execution.settle(executionError).catch((error) => {
|
||||
logger.error('[OpenAI API] Failed to settle execution:', getSafeErrorMetadata(error));
|
||||
});
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -74,6 +74,8 @@ const {
|
|||
getLangfuseTraceMessageFields,
|
||||
stripActivityLabelParts,
|
||||
CHILD_THREAD_READ_ONLY_ERROR,
|
||||
enrollAgentExecution,
|
||||
waitForAgentExecutionWrites,
|
||||
} = require('@librechat/api');
|
||||
const {
|
||||
createResponsesToolEndCallback,
|
||||
|
|
@ -573,18 +575,40 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
`[Responses API] Request ${responseId} started for agent ${agentId}, stream: ${isStreaming}`,
|
||||
);
|
||||
|
||||
// Set up abort controller
|
||||
const abortController = new AbortController();
|
||||
|
||||
// Handle client disconnect
|
||||
req.on('close', () => {
|
||||
if (!abortController.signal.aborted) {
|
||||
abortController.abort();
|
||||
const conversationId = request.previous_response_id ?? uuidv4();
|
||||
let execution;
|
||||
let executionError;
|
||||
let responseClosed = res.destroyed === true && res.writableEnded !== true;
|
||||
/** @type {Promise<import('librechat-data-provider').TAttachment | null>[]} */
|
||||
const artifactPromises = [];
|
||||
let artifactWritesCovered = false;
|
||||
const abortOnResponseClose = () => {
|
||||
if (res.writableEnded === true) {
|
||||
return;
|
||||
}
|
||||
responseClosed = true;
|
||||
if (execution && !execution.signal.aborted) {
|
||||
execution.abort();
|
||||
logger.debug('[Responses API] Client disconnected, aborting');
|
||||
}
|
||||
});
|
||||
|
||||
};
|
||||
res.once('close', abortOnResponseClose);
|
||||
try {
|
||||
execution = await enrollAgentExecution({
|
||||
runId: responseId,
|
||||
userId: principal.userId,
|
||||
conversationId,
|
||||
agentId,
|
||||
protocol: 'responses',
|
||||
/** Conversation delete-all uses the shared owner-admission fence. Remote
|
||||
* execution must observe it after durable enrollment and before provider work. */
|
||||
isPrincipalActive: db.isSubagentOwnerAdmissible,
|
||||
});
|
||||
if (responseClosed || (res.destroyed === true && res.writableEnded !== true)) {
|
||||
execution.abort();
|
||||
}
|
||||
await execution.beginProviderExecution();
|
||||
|
||||
if (request.previous_response_id != null) {
|
||||
if (typeof request.previous_response_id !== 'string') {
|
||||
return sendResponsesErrorResponse(
|
||||
|
|
@ -612,7 +636,6 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
}
|
||||
}
|
||||
|
||||
const conversationId = request.previous_response_id ?? uuidv4();
|
||||
const parentMessageId = null;
|
||||
const mcpRequestBody = createMCPRuntimeRequestBody({ messageId: responseId, conversationId });
|
||||
const agentsEConfig = appConfig?.endpoints?.[EModelEndpoint.agents];
|
||||
|
|
@ -631,7 +654,7 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
const allowedProviders = new Set(agentsEConfig?.allowedProviders);
|
||||
|
||||
// Create tool loader
|
||||
const loadTools = createToolLoader(abortController.signal);
|
||||
const loadTools = createToolLoader(execution.signal);
|
||||
const skillDbMethods = getSkillDbMethods();
|
||||
|
||||
// Initialize the agent first to check for disableStreaming
|
||||
|
|
@ -1013,8 +1036,6 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
const collectedUsage = [];
|
||||
|
||||
// Artifact promises for processing tool outputs
|
||||
/** @type {Promise<import('librechat-data-provider').TAttachment | null>[]} */
|
||||
const artifactPromises = [];
|
||||
// Use Responses API-specific callback that emits librechat:attachment events
|
||||
const toolEndCallback = createResponsesToolEndCallback({
|
||||
req,
|
||||
|
|
@ -1036,7 +1057,7 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
requestBody: mcpRequestBody,
|
||||
toolNames,
|
||||
agent: ctx.agent ?? agent,
|
||||
signal: abortController.signal,
|
||||
signal: execution.signal,
|
||||
toolRegistry: ctx.toolRegistry,
|
||||
callerCapabilityProjection,
|
||||
backgroundToolNames: ctx.backgroundToolNames,
|
||||
|
|
@ -1100,7 +1121,7 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
runId: responseId,
|
||||
summarizationConfig,
|
||||
appConfig,
|
||||
signal: abortController.signal,
|
||||
signal: execution.signal,
|
||||
customHandlers: handlers,
|
||||
initialSessions,
|
||||
requestBody: mcpRequestBody,
|
||||
|
|
@ -1125,7 +1146,7 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
requestBody: mcpRequestBody,
|
||||
...(userMCPAuthMap != null && { userMCPAuthMap }),
|
||||
},
|
||||
signal: abortController.signal,
|
||||
signal: execution.signal,
|
||||
streamMode: 'values',
|
||||
version: 'v2',
|
||||
};
|
||||
|
|
@ -1141,28 +1162,33 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
// Record token usage against balance
|
||||
const balanceConfig = getBalanceConfig(appConfig);
|
||||
const transactionsConfig = getTransactionsConfig(appConfig);
|
||||
recordCollectedUsage(
|
||||
{
|
||||
spendTokens: db.spendTokens,
|
||||
spendStructuredTokens: db.spendStructuredTokens,
|
||||
pricing: { getMultiplier: db.getMultiplier, getCacheMultiplier: db.getCacheMultiplier },
|
||||
bulkWriteOps: { insertMany: db.bulkInsertTransactions, updateBalance: db.updateBalance },
|
||||
},
|
||||
{
|
||||
user: userId,
|
||||
conversationId,
|
||||
collectedUsage,
|
||||
context: 'message',
|
||||
messageId: responseId,
|
||||
balance: balanceConfig,
|
||||
transactions: transactionsConfig,
|
||||
model: primaryConfig.model || agent.model_parameters?.model,
|
||||
endpointTokenConfig: primaryConfig.endpointTokenConfig,
|
||||
resolveEndpointTokenConfig,
|
||||
},
|
||||
).catch((err) => {
|
||||
logger.error('[Responses API] Error recording usage:', getSafeErrorMetadata(err));
|
||||
});
|
||||
execution.track(
|
||||
recordCollectedUsage(
|
||||
{
|
||||
spendTokens: db.spendTokens,
|
||||
spendStructuredTokens: db.spendStructuredTokens,
|
||||
pricing: { getMultiplier: db.getMultiplier, getCacheMultiplier: db.getCacheMultiplier },
|
||||
bulkWriteOps: {
|
||||
insertMany: db.bulkInsertTransactions,
|
||||
updateBalance: db.updateBalance,
|
||||
},
|
||||
},
|
||||
{
|
||||
user: userId,
|
||||
conversationId,
|
||||
collectedUsage,
|
||||
context: 'message',
|
||||
messageId: responseId,
|
||||
balance: balanceConfig,
|
||||
transactions: transactionsConfig,
|
||||
model: primaryConfig.model || agent.model_parameters?.model,
|
||||
endpointTokenConfig: primaryConfig.endpointTokenConfig,
|
||||
resolveEndpointTokenConfig,
|
||||
},
|
||||
).catch((err) => {
|
||||
logger.error('[Responses API] Error recording usage:', getSafeErrorMetadata(err));
|
||||
}),
|
||||
);
|
||||
|
||||
const usage = buildResponsesUsage(collectedUsage);
|
||||
|
||||
|
|
@ -1202,14 +1228,17 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
}
|
||||
}
|
||||
|
||||
// Wait for artifact processing after response ends (non-blocking)
|
||||
// The HTTP response is complete, while destructive cleanup still waits for artifacts.
|
||||
if (artifactPromises.length > 0) {
|
||||
Promise.all(artifactPromises).catch((artifactError) => {
|
||||
logger.warn(
|
||||
'[Responses API] Error processing artifacts:',
|
||||
getSafeErrorMetadata(artifactError),
|
||||
);
|
||||
});
|
||||
execution.track(
|
||||
waitForAgentExecutionWrites(artifactPromises).catch((artifactError) => {
|
||||
logger.warn(
|
||||
'[Responses API] Error processing artifacts:',
|
||||
getSafeErrorMetadata(artifactError),
|
||||
);
|
||||
}),
|
||||
);
|
||||
artifactWritesCovered = true;
|
||||
}
|
||||
} else {
|
||||
const aggregatorHandlers = createAggregatorEventHandlers(aggregator);
|
||||
|
|
@ -1217,8 +1246,6 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
// Collect usage for balance tracking
|
||||
const collectedUsage = [];
|
||||
|
||||
/** @type {Promise<import('librechat-data-provider').TAttachment | null>[]} */
|
||||
const artifactPromises = [];
|
||||
const toolEndCallback = createToolEndCallback({ req, res, artifactPromises, streamId: null });
|
||||
|
||||
const toolExecuteOptions = {
|
||||
|
|
@ -1233,7 +1260,7 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
requestBody: mcpRequestBody,
|
||||
toolNames,
|
||||
agent: ctx.agent ?? agent,
|
||||
signal: abortController.signal,
|
||||
signal: execution.signal,
|
||||
toolRegistry: ctx.toolRegistry,
|
||||
callerCapabilityProjection,
|
||||
backgroundToolNames: ctx.backgroundToolNames,
|
||||
|
|
@ -1295,7 +1322,7 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
runId: responseId,
|
||||
summarizationConfig,
|
||||
appConfig,
|
||||
signal: abortController.signal,
|
||||
signal: execution.signal,
|
||||
customHandlers: handlers,
|
||||
initialSessions,
|
||||
requestBody: mcpRequestBody,
|
||||
|
|
@ -1319,7 +1346,7 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
requestBody: mcpRequestBody,
|
||||
...(userMCPAuthMap != null && { userMCPAuthMap }),
|
||||
},
|
||||
signal: abortController.signal,
|
||||
signal: execution.signal,
|
||||
streamMode: 'values',
|
||||
version: 'v2',
|
||||
};
|
||||
|
|
@ -1335,38 +1362,44 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
// Record token usage against balance
|
||||
const balanceConfig = getBalanceConfig(appConfig);
|
||||
const transactionsConfig = getTransactionsConfig(appConfig);
|
||||
recordCollectedUsage(
|
||||
{
|
||||
spendTokens: db.spendTokens,
|
||||
spendStructuredTokens: db.spendStructuredTokens,
|
||||
pricing: { getMultiplier: db.getMultiplier, getCacheMultiplier: db.getCacheMultiplier },
|
||||
bulkWriteOps: { insertMany: db.bulkInsertTransactions, updateBalance: db.updateBalance },
|
||||
},
|
||||
{
|
||||
user: userId,
|
||||
conversationId,
|
||||
collectedUsage,
|
||||
context: 'message',
|
||||
messageId: responseId,
|
||||
balance: balanceConfig,
|
||||
transactions: transactionsConfig,
|
||||
model: primaryConfig.model || agent.model_parameters?.model,
|
||||
endpointTokenConfig: primaryConfig.endpointTokenConfig,
|
||||
resolveEndpointTokenConfig,
|
||||
},
|
||||
).catch((err) => {
|
||||
logger.error('[Responses API] Error recording usage:', getSafeErrorMetadata(err));
|
||||
});
|
||||
execution.track(
|
||||
recordCollectedUsage(
|
||||
{
|
||||
spendTokens: db.spendTokens,
|
||||
spendStructuredTokens: db.spendStructuredTokens,
|
||||
pricing: { getMultiplier: db.getMultiplier, getCacheMultiplier: db.getCacheMultiplier },
|
||||
bulkWriteOps: {
|
||||
insertMany: db.bulkInsertTransactions,
|
||||
updateBalance: db.updateBalance,
|
||||
},
|
||||
},
|
||||
{
|
||||
user: userId,
|
||||
conversationId,
|
||||
collectedUsage,
|
||||
context: 'message',
|
||||
messageId: responseId,
|
||||
balance: balanceConfig,
|
||||
transactions: transactionsConfig,
|
||||
model: primaryConfig.model || agent.model_parameters?.model,
|
||||
endpointTokenConfig: primaryConfig.endpointTokenConfig,
|
||||
resolveEndpointTokenConfig,
|
||||
},
|
||||
).catch((err) => {
|
||||
logger.error('[Responses API] Error recording usage:', getSafeErrorMetadata(err));
|
||||
}),
|
||||
);
|
||||
|
||||
if (artifactPromises.length > 0) {
|
||||
try {
|
||||
await Promise.all(artifactPromises);
|
||||
await waitForAgentExecutionWrites(artifactPromises);
|
||||
} catch (artifactError) {
|
||||
logger.warn(
|
||||
'[Responses API] Error processing artifacts:',
|
||||
getSafeErrorMetadata(artifactError),
|
||||
);
|
||||
}
|
||||
artifactWritesCovered = true;
|
||||
}
|
||||
|
||||
const response = buildAggregatedResponse(
|
||||
|
|
@ -1407,6 +1440,7 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
);
|
||||
}
|
||||
} catch (error) {
|
||||
executionError = error;
|
||||
logger.error('[Responses API] Error:', getSafeErrorMetadata(error));
|
||||
const protectionEnabled = hasModelBoundContentProtection(
|
||||
appConfig?.filters,
|
||||
|
|
@ -1443,6 +1477,23 @@ const executeResponse = async (envelope, { req, res }) => {
|
|||
sendResponsesErrorResponse(res, statusCode, errorMessage, errorType, errorCode);
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
res.off('close', abortOnResponseClose);
|
||||
if (execution) {
|
||||
if (!artifactWritesCovered && artifactPromises.length > 0) {
|
||||
execution.track(
|
||||
waitForAgentExecutionWrites(artifactPromises).catch((artifactError) => {
|
||||
logger.warn(
|
||||
'[Responses API] Error processing artifacts:',
|
||||
getSafeErrorMetadata(artifactError),
|
||||
);
|
||||
}),
|
||||
);
|
||||
}
|
||||
await execution.settle(executionError).catch((error) => {
|
||||
logger.error('[Responses API] Failed to settle execution:', getSafeErrorMetadata(error));
|
||||
});
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
|
|
|
|||
|
|
@ -2,6 +2,8 @@ const archiveAllHandler = jest.fn();
|
|||
const generationJobManager = {
|
||||
getJob: jest.fn().mockResolvedValue(null),
|
||||
abortJob: jest.fn().mockResolvedValue({ success: true }),
|
||||
getCleanupBlockingJobIdsForUser: jest.fn().mockResolvedValue([]),
|
||||
getCleanupBlockingJobIdsForConversations: jest.fn().mockResolvedValue([]),
|
||||
};
|
||||
const subagentActivityHandlerInputs = [];
|
||||
const moderatedTexts = [];
|
||||
|
|
|
|||
|
|
@ -109,6 +109,8 @@ describe('Convos Routes', () => {
|
|||
moderatedTexts.length = 0;
|
||||
generationJobManager.getJob.mockResolvedValue(null);
|
||||
generationJobManager.abortJob.mockResolvedValue({ success: true });
|
||||
generationJobManager.getCleanupBlockingJobIdsForUser.mockResolvedValue([]);
|
||||
generationJobManager.getCleanupBlockingJobIdsForConversations.mockResolvedValue([]);
|
||||
});
|
||||
|
||||
it('binds the activity subscription adapter to the subagent task store', () => {
|
||||
|
|
@ -406,6 +408,9 @@ describe('Convos Routes', () => {
|
|||
|
||||
it('drains a paused event actor after owner-wide deletion removes its conversation', async () => {
|
||||
const createdAt = Date.now();
|
||||
generationJobManager.getCleanupBlockingJobIdsForUser.mockResolvedValue([
|
||||
'paused-event-child',
|
||||
]);
|
||||
deleteConvos.mockResolvedValue({
|
||||
deletedCount: 1,
|
||||
conversationIds: ['paused-event-child'],
|
||||
|
|
@ -429,6 +434,108 @@ describe('Convos Routes', () => {
|
|||
});
|
||||
});
|
||||
|
||||
it('drains every owner remote run before selecting the delete-all snapshot', async () => {
|
||||
const createdAt = Date.now();
|
||||
generationJobManager.getCleanupBlockingJobIdsForUser.mockResolvedValue([
|
||||
'resp-new-conversation',
|
||||
]);
|
||||
generationJobManager.getJob.mockImplementation(async (streamId) =>
|
||||
streamId === 'resp-new-conversation'
|
||||
? {
|
||||
conversationId: 'conversation-not-yet-persisted',
|
||||
metadata: { userId: 'test-user-123' },
|
||||
status: 'running',
|
||||
createdAt,
|
||||
}
|
||||
: null,
|
||||
);
|
||||
deleteConvos.mockResolvedValue({
|
||||
deletedCount: 1,
|
||||
conversationIds: ['conversation-not-yet-persisted'],
|
||||
});
|
||||
|
||||
const response = await request(app).delete('/api/convos/all');
|
||||
|
||||
expect(response.status).toBe(201);
|
||||
expect(generationJobManager.getCleanupBlockingJobIdsForUser).toHaveBeenCalledWith(
|
||||
'test-user-123',
|
||||
undefined,
|
||||
);
|
||||
expect(generationJobManager.abortJob).toHaveBeenCalledWith('resp-new-conversation', {
|
||||
expectedCreatedAt: createdAt,
|
||||
awaitProviderDrain: true,
|
||||
});
|
||||
expect(generationJobManager.abortJob.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
deleteConvos.mock.invocationCallOrder[0],
|
||||
);
|
||||
});
|
||||
|
||||
it('re-drains and removes persistence created during a recovered owner-fence gap', async () => {
|
||||
const createdAt = Date.now();
|
||||
generationJobManager.getCleanupBlockingJobIdsForUser
|
||||
.mockResolvedValueOnce([])
|
||||
.mockResolvedValueOnce(['resp-gap']);
|
||||
generationJobManager.getJob.mockImplementation(async (streamId) =>
|
||||
streamId === 'resp-gap'
|
||||
? { metadata: { userId: 'test-user-123' }, status: 'running', createdAt }
|
||||
: null,
|
||||
);
|
||||
deleteConvos
|
||||
.mockResolvedValueOnce({ deletedCount: 1, conversationIds: ['original'] })
|
||||
.mockImplementationOnce(async (_userId, _filter, options) => {
|
||||
expect(options).toEqual(
|
||||
expect.objectContaining({ allowEmpty: true, beforeDelete: expect.any(Function) }),
|
||||
);
|
||||
await options.beforeDelete(['gap-conversation']);
|
||||
return { deletedCount: 1, conversationIds: ['gap-conversation'] };
|
||||
});
|
||||
subagentThreadStore.withOwnerDeletionFence.mockImplementationOnce(
|
||||
async (_userId, _tenantId, deletion, recover) => {
|
||||
const result = await deletion();
|
||||
await recover();
|
||||
return result;
|
||||
},
|
||||
);
|
||||
|
||||
const response = await request(app).delete('/api/convos/all');
|
||||
|
||||
expect(response.status).toBe(201);
|
||||
expect(generationJobManager.abortJob).toHaveBeenCalledWith('resp-gap', {
|
||||
expectedCreatedAt: createdAt,
|
||||
awaitProviderDrain: true,
|
||||
});
|
||||
expect(deleteConvos).toHaveBeenCalledTimes(2);
|
||||
expect(deleteMessages).toHaveBeenCalledWith({ user: 'test-user-123' });
|
||||
expect(deleteAgentCheckpoints.mock.calls.map((call) => call[0])).toEqual([
|
||||
['original'],
|
||||
['gap-conversation'],
|
||||
]);
|
||||
});
|
||||
|
||||
it('retries owner message cleanup after conversations were already removed', async () => {
|
||||
deleteConvos
|
||||
.mockResolvedValueOnce({ deletedCount: 1, conversationIds: ['original'] })
|
||||
.mockResolvedValueOnce({ deletedCount: 0, conversationIds: [] });
|
||||
deleteMessages
|
||||
.mockRejectedValueOnce(new Error('message database unavailable'))
|
||||
.mockResolvedValueOnce({ deletedCount: 1 });
|
||||
|
||||
const first = await request(app).delete('/api/convos/all');
|
||||
const retry = await request(app).delete('/api/convos/all');
|
||||
|
||||
expect(first.status).toBe(500);
|
||||
expect(retry.status).toBe(201);
|
||||
expect(deleteAgentCheckpoints.mock.calls[0][0]).toEqual(['original']);
|
||||
expect(deleteConvos).toHaveBeenNthCalledWith(
|
||||
2,
|
||||
'test-user-123',
|
||||
{},
|
||||
expect.objectContaining({ allowEmpty: true }),
|
||||
);
|
||||
expect(deleteMessages).toHaveBeenCalledTimes(2);
|
||||
expect(deleteMessages).toHaveBeenLastCalledWith({ user: 'test-user-123' });
|
||||
});
|
||||
|
||||
it('should delete all conversations, tool calls, and shared links for a user', async () => {
|
||||
const mockDbResponse = {
|
||||
deletedCount: 5,
|
||||
|
|
@ -636,6 +743,36 @@ describe('Convos Routes', () => {
|
|||
{},
|
||||
expect.objectContaining({ beforeDelete: expect.any(Function) }),
|
||||
);
|
||||
expect(subagentThreadStore.withOwnerDeletionFence.mock.calls[0][3]).toEqual(
|
||||
expect.any(Function),
|
||||
);
|
||||
});
|
||||
|
||||
it('drains owner remote runs before the empty-filter deletion snapshot', async () => {
|
||||
const createdAt = Date.now();
|
||||
generationJobManager.getCleanupBlockingJobIdsForUser.mockResolvedValue(['resp-new']);
|
||||
generationJobManager.getJob.mockImplementation(async (streamId) =>
|
||||
streamId === 'resp-new'
|
||||
? { metadata: { userId: 'test-user-123' }, status: 'running', createdAt }
|
||||
: null,
|
||||
);
|
||||
deleteConvos.mockResolvedValue({
|
||||
deletedCount: 1,
|
||||
conversationIds: ['new-conversation'],
|
||||
});
|
||||
|
||||
const response = await request(app)
|
||||
.delete('/api/convos')
|
||||
.send({ arg: { thread_id: 'thread-abc' } });
|
||||
|
||||
expect(response.status).toBe(201);
|
||||
expect(generationJobManager.abortJob).toHaveBeenCalledWith('resp-new', {
|
||||
expectedCreatedAt: createdAt,
|
||||
awaitProviderDrain: true,
|
||||
});
|
||||
expect(generationJobManager.abortJob.mock.invocationCallOrder[0]).toBeLessThan(
|
||||
deleteConvos.mock.invocationCallOrder[0],
|
||||
);
|
||||
});
|
||||
|
||||
it('drains a paused event actor after an empty-filter deletion removes it', async () => {
|
||||
|
|
@ -666,9 +803,12 @@ describe('Convos Routes', () => {
|
|||
});
|
||||
|
||||
it('fails closed before checkpoint pruning when generation lookup stays unavailable', async () => {
|
||||
deleteConvos.mockResolvedValue({
|
||||
deletedCount: 1,
|
||||
conversationIds: ['paused-event-child'],
|
||||
deleteConvos.mockImplementation(async (_userId, _filter, options) => {
|
||||
await options.beforeDelete(['paused-event-child']);
|
||||
return {
|
||||
deletedCount: 1,
|
||||
conversationIds: ['paused-event-child'],
|
||||
};
|
||||
});
|
||||
generationJobManager.getJob.mockRejectedValue(new Error('generation store unavailable'));
|
||||
|
||||
|
|
@ -743,15 +883,172 @@ describe('Convos Routes', () => {
|
|||
expectedCreatedAt: createdAt,
|
||||
awaitProviderDrain: true,
|
||||
});
|
||||
expect(deleteConvos).toHaveBeenNthCalledWith(2, 'test-user-123', {
|
||||
conversationId: { $in: ['parent-conversation', 'child-conversation'] },
|
||||
});
|
||||
expect(deleteConvos).toHaveBeenNthCalledWith(
|
||||
2,
|
||||
'test-user-123',
|
||||
{ conversationId: { $in: ['parent-conversation', 'child-conversation'] } },
|
||||
{ allowEmpty: true },
|
||||
);
|
||||
expect(deleteMessages).toHaveBeenCalledWith({
|
||||
user: 'test-user-123',
|
||||
conversationId: { $in: ['parent-conversation', 'child-conversation'] },
|
||||
});
|
||||
});
|
||||
|
||||
it('drains response-id runs indexed under a deleted conversation', async () => {
|
||||
const createdAt = Date.now();
|
||||
generationJobManager.getCleanupBlockingJobIdsForConversations.mockResolvedValue([
|
||||
'resp_remote-run',
|
||||
]);
|
||||
generationJobManager.getJob.mockImplementation(async (streamId) =>
|
||||
streamId === 'resp_remote-run'
|
||||
? { metadata: { userId: 'test-user-123' }, status: 'running', createdAt }
|
||||
: null,
|
||||
);
|
||||
deleteConvos.mockImplementationOnce(async (_userId, _filter, options) => {
|
||||
await options.beforeDelete(['conversation-1']);
|
||||
return { deletedCount: 1, conversationIds: ['conversation-1'] };
|
||||
});
|
||||
|
||||
const response = await request(app)
|
||||
.delete('/api/convos')
|
||||
.send({ arg: { conversationId: 'conversation-1' } });
|
||||
|
||||
expect(response.status).toBe(201);
|
||||
expect(generationJobManager.getCleanupBlockingJobIdsForConversations).toHaveBeenCalledWith(
|
||||
'test-user-123',
|
||||
['conversation-1'],
|
||||
undefined,
|
||||
);
|
||||
expect(generationJobManager.abortJob).toHaveBeenCalledWith('resp_remote-run', {
|
||||
expectedCreatedAt: createdAt,
|
||||
awaitProviderDrain: true,
|
||||
});
|
||||
});
|
||||
|
||||
it('waits for terminal response-id runs whose provider writes are undrained', async () => {
|
||||
const createdAt = Date.now();
|
||||
generationJobManager.getCleanupBlockingJobIdsForConversations.mockResolvedValue([
|
||||
'resp_terminal-run',
|
||||
]);
|
||||
generationJobManager.getJob.mockImplementation(async (streamId) =>
|
||||
streamId === 'resp_terminal-run'
|
||||
? {
|
||||
metadata: { userId: 'test-user-123', providerDrained: false },
|
||||
status: 'complete',
|
||||
createdAt,
|
||||
}
|
||||
: null,
|
||||
);
|
||||
deleteConvos.mockImplementationOnce(async (_userId, _filter, options) => {
|
||||
await options.beforeDelete(['conversation-1']);
|
||||
return { deletedCount: 1, conversationIds: ['conversation-1'] };
|
||||
});
|
||||
|
||||
const response = await request(app)
|
||||
.delete('/api/convos')
|
||||
.send({ arg: { conversationId: 'conversation-1' } });
|
||||
|
||||
expect(response.status).toBe(201);
|
||||
expect(generationJobManager.abortJob).toHaveBeenCalledWith('resp_terminal-run', {
|
||||
expectedCreatedAt: createdAt,
|
||||
awaitProviderDrain: true,
|
||||
});
|
||||
});
|
||||
|
||||
it('catches a response-id run admitted after the pre-delete snapshot', async () => {
|
||||
const createdAt = Date.now();
|
||||
let deletionCommitted = false;
|
||||
generationJobManager.getCleanupBlockingJobIdsForConversations.mockImplementation(async () =>
|
||||
deletionCommitted ? ['resp_late-run'] : [],
|
||||
);
|
||||
generationJobManager.getJob.mockImplementation(async (streamId) =>
|
||||
streamId === 'resp_late-run'
|
||||
? { metadata: { userId: 'test-user-123' }, status: 'running', createdAt }
|
||||
: null,
|
||||
);
|
||||
deleteConvos.mockImplementationOnce(async (_userId, _filter, options) => {
|
||||
await options.beforeDelete(['conversation-1']);
|
||||
deletionCommitted = true;
|
||||
return { deletedCount: 1, conversationIds: ['conversation-1'] };
|
||||
});
|
||||
|
||||
const response = await request(app)
|
||||
.delete('/api/convos')
|
||||
.send({ arg: { conversationId: 'conversation-1' } });
|
||||
|
||||
expect(response.status).toBe(201);
|
||||
expect(generationJobManager.getCleanupBlockingJobIdsForConversations).toHaveBeenCalledTimes(
|
||||
2,
|
||||
);
|
||||
expect(generationJobManager.abortJob).toHaveBeenCalledWith('resp_late-run', {
|
||||
expectedCreatedAt: createdAt,
|
||||
awaitProviderDrain: true,
|
||||
});
|
||||
});
|
||||
|
||||
it('removes persistence even when a racing response drains before post-delete discovery', async () => {
|
||||
deleteConvos
|
||||
.mockImplementationOnce(async (_userId, _filter, options) => {
|
||||
await options.beforeDelete(['conversation-1']);
|
||||
return { deletedCount: 1, conversationIds: ['conversation-1'] };
|
||||
})
|
||||
.mockResolvedValueOnce({ deletedCount: 1, conversationIds: ['conversation-1'] });
|
||||
generationJobManager.getCleanupBlockingJobIdsForConversations.mockResolvedValue([]);
|
||||
|
||||
const response = await request(app)
|
||||
.delete('/api/convos')
|
||||
.send({ arg: { conversationId: 'conversation-1' } });
|
||||
|
||||
expect(response.status).toBe(201);
|
||||
expect(generationJobManager.abortJob).not.toHaveBeenCalled();
|
||||
expect(deleteConvos).toHaveBeenNthCalledWith(
|
||||
2,
|
||||
'test-user-123',
|
||||
{ conversationId: { $in: ['conversation-1'] } },
|
||||
{ allowEmpty: true },
|
||||
);
|
||||
expect(deleteMessages).toHaveBeenCalledWith({
|
||||
user: 'test-user-123',
|
||||
conversationId: { $in: ['conversation-1'] },
|
||||
});
|
||||
});
|
||||
|
||||
it('fails closed when the idempotent remnant sweep hits a storage failure', async () => {
|
||||
deleteConvos
|
||||
.mockImplementationOnce(async (_userId, _filter, options) => {
|
||||
await options.beforeDelete(['conversation-1']);
|
||||
return { deletedCount: 1, conversationIds: ['conversation-1'] };
|
||||
})
|
||||
.mockRejectedValueOnce(new Error('remnant database unavailable'));
|
||||
|
||||
const response = await request(app)
|
||||
.delete('/api/convos')
|
||||
.send({ arg: { conversationId: 'conversation-1' } });
|
||||
|
||||
expect(response.status).toBe(500);
|
||||
expect(response.text).toBe('Error clearing conversations');
|
||||
expect(deleteAgentCheckpoints).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('fails closed when remnant message cleanup is unavailable', async () => {
|
||||
deleteConvos
|
||||
.mockImplementationOnce(async (_userId, _filter, options) => {
|
||||
await options.beforeDelete(['conversation-1']);
|
||||
return { deletedCount: 1, conversationIds: ['conversation-1'] };
|
||||
})
|
||||
.mockResolvedValueOnce({ deletedCount: 0, conversationIds: [] });
|
||||
deleteMessages.mockRejectedValueOnce(new Error('message database unavailable'));
|
||||
|
||||
const response = await request(app)
|
||||
.delete('/api/convos')
|
||||
.send({ arg: { conversationId: 'conversation-1' } });
|
||||
|
||||
expect(response.status).toBe(500);
|
||||
expect(response.text).toBe('Error clearing conversations');
|
||||
expect(deleteAgentCheckpoints).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('does not prune generation persistence when provider stop is unconfirmed', async () => {
|
||||
const createdAt = Date.now();
|
||||
let deletionCommitted = false;
|
||||
|
|
@ -1010,6 +1307,7 @@ describe('Convos Routes', () => {
|
|||
});
|
||||
|
||||
expect(executionOrder).toEqual([
|
||||
'deleteConvos',
|
||||
'deleteConvos',
|
||||
'deleteToolCalls',
|
||||
'deleteConvoSharedLinksWithCleanup',
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ const db = require('~/models');
|
|||
const apiKeyMiddleware = createRequireApiKeyAuth({
|
||||
validateAgentApiKey: db.validateAgentApiKey,
|
||||
findUser: db.findUser,
|
||||
isPrincipalActive: db.isAgentTriggerPrincipalActive,
|
||||
});
|
||||
|
||||
const requireRemoteAgentAuth = createRemoteAgentAuth({
|
||||
|
|
@ -21,6 +22,7 @@ const requireRemoteAgentAuth = createRemoteAgentAuth({
|
|||
findUser: db.findUser,
|
||||
getRolesByNames: db.findRolesByNames,
|
||||
updateUser: db.updateUser,
|
||||
isPrincipalActive: db.isAgentTriggerPrincipalActive,
|
||||
getAppConfig,
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -275,10 +275,29 @@ async function retryPostDeleteCancellation(cancellationPlan, deletedConversation
|
|||
}
|
||||
|
||||
/** Confirms every exact generation is stopped before its conversation wave is removed. */
|
||||
async function confirmAgentGenerationsDrained(userId, conversationIds, leaseTaskIds = []) {
|
||||
async function confirmAgentGenerationsDrained(
|
||||
userId,
|
||||
conversationIds,
|
||||
leaseTaskIds = [],
|
||||
tenantId,
|
||||
ownerWide = false,
|
||||
) {
|
||||
let foundActiveGeneration = false;
|
||||
const drainErrors = [];
|
||||
const generationIds = [...new Set([...conversationIds, ...leaseTaskIds])];
|
||||
let conversationRunIds;
|
||||
try {
|
||||
conversationRunIds = ownerWide
|
||||
? await GenerationJobManager.getCleanupBlockingJobIdsForUser(userId, tenantId)
|
||||
: await GenerationJobManager.getCleanupBlockingJobIdsForConversations(
|
||||
userId,
|
||||
conversationIds,
|
||||
tenantId,
|
||||
);
|
||||
} catch (error) {
|
||||
logger.warn('Conversation generation index lookup failed', error);
|
||||
throw new Error('Conversation generations could not be confirmed drained.');
|
||||
}
|
||||
const generationIds = [...new Set([...conversationIds, ...leaseTaskIds, ...conversationRunIds])];
|
||||
await Promise.all(
|
||||
generationIds.map(async (conversationId) => {
|
||||
let job;
|
||||
|
|
@ -296,6 +315,7 @@ async function confirmAgentGenerationsDrained(userId, conversationIds, leaseTask
|
|||
const needsDrain =
|
||||
job.status === 'running' ||
|
||||
job.status === 'requires_action' ||
|
||||
job.metadata?.providerDrained === false ||
|
||||
job.metadata?.terminalPersistencePending === true;
|
||||
if (!needsDrain) return;
|
||||
foundActiveGeneration = true;
|
||||
|
|
@ -339,25 +359,53 @@ async function confirmAgentGenerationsDrained(userId, conversationIds, leaseTask
|
|||
return true;
|
||||
}
|
||||
|
||||
/** Stops event-bound child generations on their owning replica and then removes
|
||||
* persistence that raced the first conversation cascade. */
|
||||
async function drainDeletedAgentGenerations(userId, conversationIds, leaseTaskIds = []) {
|
||||
const foundActiveGeneration = await confirmAgentGenerationsDrained(
|
||||
/** Repeats generation discovery after the conversation wave is gone, then always
|
||||
* removes remnants for that immutable deletion set. A remote run may settle and
|
||||
* leave the cleanup index between persisting and this lookup; absence from the
|
||||
* index is therefore not evidence that the second persistence sweep is unnecessary. */
|
||||
async function drainDeletedAgentGenerations(userId, conversationIds, leaseTaskIds = [], tenantId) {
|
||||
await confirmAgentGenerationsDrained(userId, conversationIds, leaseTaskIds, tenantId);
|
||||
await db.deleteConvos(userId, { conversationId: { $in: conversationIds } }, { allowEmpty: true });
|
||||
await db.deleteMessages({ user: userId, conversationId: { $in: conversationIds } });
|
||||
}
|
||||
|
||||
/** Orders every owner-scoped agent execution against a delete-all persistence
|
||||
* snapshot. The recovery callback repeats the non-subagent drain if the durable
|
||||
* fence ever lapses and must be reacquired after deletion has started. */
|
||||
async function withAgentOwnerDeletionFence(userId, tenantId, deletion, recoverPersistence) {
|
||||
const drainRemoteRuns = () => confirmAgentGenerationsDrained(userId, [], [], tenantId, true);
|
||||
let recoveryConversationIds = [];
|
||||
const result = await subagentThreadTaskStore.withOwnerDeletionFence(
|
||||
userId,
|
||||
conversationIds,
|
||||
leaseTaskIds,
|
||||
tenantId,
|
||||
async () => {
|
||||
await drainRemoteRuns();
|
||||
return deletion();
|
||||
},
|
||||
async () => {
|
||||
await drainRemoteRuns();
|
||||
/** Runs only after the fence was restored. No new provider may enter while
|
||||
* persistence created during the gap is removed idempotently. */
|
||||
const recovery = await recoverPersistence();
|
||||
recoveryConversationIds = recovery.conversationIds ?? [];
|
||||
},
|
||||
);
|
||||
if (!foundActiveGeneration) {
|
||||
return;
|
||||
}
|
||||
try {
|
||||
await db.deleteConvos(userId, { conversationId: { $in: conversationIds } });
|
||||
} catch {
|
||||
// Expected when no generation raced the first cascade.
|
||||
}
|
||||
await db
|
||||
.deleteMessages({ user: userId, conversationId: { $in: conversationIds } })
|
||||
.catch((error) => logger.warn('Deleted child message remnant cleanup failed', error));
|
||||
return { result, recoveryConversationIds };
|
||||
}
|
||||
|
||||
async function deleteOwnerConversationPersistence(userId, filter, tenantId, checkpointer) {
|
||||
const result = await db.deleteConvos(userId, filter, {
|
||||
allowEmpty: true,
|
||||
beforeDelete: (conversationIds) =>
|
||||
confirmAgentGenerationsDrained(userId, conversationIds, [], tenantId),
|
||||
});
|
||||
/** Consume the deletion receipt before the fallible message sweep. A retry after
|
||||
* conversations are gone cannot reconstruct these checkpoint identities. */
|
||||
await deleteAgentCheckpoints(result.conversationIds ?? [], checkpointer);
|
||||
/** Always runs, including an empty conversation retry, so an interrupted writer
|
||||
* that persisted messages first cannot make its cleanup permanently unreachable. */
|
||||
await db.deleteMessages({ user: userId });
|
||||
return result;
|
||||
}
|
||||
|
||||
router.delete('/', configMiddleware, async (req, res) => {
|
||||
|
|
@ -396,8 +444,10 @@ router.delete('/', configMiddleware, async (req, res) => {
|
|||
typeof req.user.tenantId === 'string' && req.user.tenantId !== ''
|
||||
? req.user.tenantId
|
||||
: undefined;
|
||||
const checkpointer = req.config?.endpoints?.[EModelEndpoint.agents]?.checkpointer;
|
||||
let cancellationPlan;
|
||||
let dbResponse;
|
||||
let recoveryConversationIds = [];
|
||||
if (filter.conversationId) {
|
||||
/** Resolve the targets while the conversations still exist: the second pass
|
||||
* runs after their rows are gone and can only reach registered owners. */
|
||||
|
|
@ -409,20 +459,26 @@ router.delete('/', configMiddleware, async (req, res) => {
|
|||
await subagentThreadTaskStore.cancelPlan(cancellationPlan);
|
||||
dbResponse = await db.deleteConvos(req.user.id, filter, {
|
||||
beforeDelete: (conversationIds) =>
|
||||
confirmAgentGenerationsDrained(req.user.id, conversationIds),
|
||||
confirmAgentGenerationsDrained(req.user.id, conversationIds, [], tenantId),
|
||||
});
|
||||
} else {
|
||||
/** An empty filter deletes every conversation this owner has, so it runs behind
|
||||
* the same admission fence as `DELETE /all` rather than a bare drain. */
|
||||
dbResponse = await subagentThreadTaskStore.withOwnerDeletionFence(req.user.id, tenantId, () =>
|
||||
db.deleteConvos(req.user.id, filter, {
|
||||
beforeDelete: (conversationIds) =>
|
||||
confirmAgentGenerationsDrained(req.user.id, conversationIds),
|
||||
}),
|
||||
const fencedDeletion = await withAgentOwnerDeletionFence(
|
||||
req.user.id,
|
||||
tenantId,
|
||||
() => deleteOwnerConversationPersistence(req.user.id, filter, tenantId, checkpointer),
|
||||
() => deleteOwnerConversationPersistence(req.user.id, filter, tenantId, checkpointer),
|
||||
);
|
||||
dbResponse = fencedDeletion.result;
|
||||
recoveryConversationIds = fencedDeletion.recoveryConversationIds;
|
||||
}
|
||||
const deletedConversationIds =
|
||||
dbResponse.conversationIds ?? (filter.conversationId ? [filter.conversationId] : []);
|
||||
const deletedConversationIds = [
|
||||
...new Set([
|
||||
...(dbResponse.conversationIds ?? (filter.conversationId ? [filter.conversationId] : [])),
|
||||
...recoveryConversationIds,
|
||||
]),
|
||||
];
|
||||
/** Root deletion closes new child admission. Replay the plan to catch a task
|
||||
* admitted after the first pass but before that fence, extended with the cascade
|
||||
* this deletion reported. */
|
||||
|
|
@ -441,20 +497,20 @@ router.delete('/', configMiddleware, async (req, res) => {
|
|||
deletedConversationIds.includes(lease.conversationId),
|
||||
)
|
||||
.map((lease) => lease.taskId),
|
||||
tenantId,
|
||||
);
|
||||
} else if (deletedConversationIds.length > 0) {
|
||||
/** Owner-wide deletion drains lease-backed tasks before the cascade, but a
|
||||
* requires_action event actor has intentionally released its lease. Its durable
|
||||
* generation is still addressable by the deleted conversation id and must be
|
||||
* terminalized before its checkpoint is pruned. */
|
||||
await drainDeletedAgentGenerations(req.user.id, deletedConversationIds);
|
||||
await drainDeletedAgentGenerations(req.user.id, deletedConversationIds, [], tenantId);
|
||||
}
|
||||
// HITL: prune the deleted conversations' durable checkpoints — a paused run's
|
||||
// checkpoint would otherwise persist until the Mongo TTL. Never throws.
|
||||
await deleteAgentCheckpoints(
|
||||
deletedConversationIds,
|
||||
req.config?.endpoints?.[EModelEndpoint.agents]?.checkpointer,
|
||||
);
|
||||
if (filter.conversationId) {
|
||||
await deleteAgentCheckpoints(deletedConversationIds, checkpointer);
|
||||
}
|
||||
if (filter.conversationId) {
|
||||
await Promise.all(deletedConversationIds.map((id) => db.deleteToolCalls(req.user.id, id)));
|
||||
await Promise.all(
|
||||
|
|
@ -474,28 +530,17 @@ router.delete('/all', configMiddleware, async (req, res) => {
|
|||
typeof req.user.tenantId === 'string' && req.user.tenantId !== ''
|
||||
? req.user.tenantId
|
||||
: undefined;
|
||||
const checkpointer = req.config?.endpoints?.[EModelEndpoint.agents]?.checkpointer;
|
||||
/** Fences new child admission for this owner, drains the live ones, and deletes
|
||||
* inside that fence: a child admitted on another replica mid-deletion would
|
||||
* otherwise keep running against conversations that no longer exist. */
|
||||
const dbResponse = await subagentThreadTaskStore.withOwnerDeletionFence(
|
||||
const fencedDeletion = await withAgentOwnerDeletionFence(
|
||||
req.user.id,
|
||||
tenantId,
|
||||
() =>
|
||||
db.deleteConvos(
|
||||
req.user.id,
|
||||
{},
|
||||
{
|
||||
beforeDelete: (conversationIds) =>
|
||||
confirmAgentGenerationsDrained(req.user.id, conversationIds),
|
||||
},
|
||||
),
|
||||
);
|
||||
await drainDeletedAgentGenerations(req.user.id, dbResponse.conversationIds ?? []);
|
||||
// HITL: prune ALL the deleted conversations' durable checkpoints in one bulk pass.
|
||||
await deleteAgentCheckpoints(
|
||||
dbResponse.conversationIds,
|
||||
req.config?.endpoints?.[EModelEndpoint.agents]?.checkpointer,
|
||||
() => deleteOwnerConversationPersistence(req.user.id, {}, tenantId, checkpointer),
|
||||
() => deleteOwnerConversationPersistence(req.user.id, {}, tenantId, checkpointer),
|
||||
);
|
||||
const dbResponse = fencedDeletion.result;
|
||||
await db.deleteToolCalls(req.user.id);
|
||||
await deleteAllSharedLinksWithCleanup(req.user.id);
|
||||
res.status(201).json(dbResponse);
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue