CERBERUS/LibreChat
1
0
Fork 0
mirror of https://github.com/danny-avila/LibreChat.git synced 2026-10-22 01:53:33 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

🎯 fix: Require an Abort-Shaped Error and Spare Shared OAuth Flows

Browse source 
Addresses both findings from the Codex review of b3c67082cd.

An aborted run signal proves the turn is over, not that the rejection in
hand was the cancellation — a permission, OAuth, or upstream failure can
reject in the same tick a user presses Stop. Both quiet-log branches now
require the error to look like an abort as well, so a real failure racing
the Stop stays at error level and visible to operational alerts.
`isAbortError` moves to `@librechat/api` and the copy in
`abortMiddleware` is retired, keeping one implementation. Its spec pulls
the real function through its partial package mock rather than asserting
a stub.

Forwarding the run signal also reached ActionService, where two of the
three OAuth flows are keyed `userId:action_id` and so outlive the run
that opened them: a second run for the same action joins the very same
record, and the browser's OAuth callback reads its metadata to exchange
the code. `monitorFlow` deletes the key when a waiter's signal aborts, so
one Stop could strand a concurrent run and discard an authorization the
user had already granted. Those two flows no longer take the signal; the
run-scoped login flow, keyed by thread and run, still does.

Every guard is covered by a test that fails when the guard is removed.
This commit is contained in:
Danny Avila 2026-09-02 12:30:31 -04:00
parent b3c67082cd
commit d53ff170ab
10 changed files with 346 additions and 132 deletions
Download patch file Download diff file Expand all files Collapse all files

31
api/server/middleware/abortMiddleware.js
View file

@ -4,6 +4,7 @@ const {
isEnabled,
sendEvent,
countTokens,
isAbortError,
GenerationJobManager,
recordCollectedUsage,
getTransactionsConfig,
@ -16,36 +17,6 @@ const { sendError } = require('~/server/middleware/error');
const { abortRun } = require('./abortRun');
const db = require('~/models');
/**
* @param {Error | unknown} error
* @returns {boolean}
*/
const isAbortError = (error) => {
const visited = new Set();
let current = error;
while (current && typeof current === 'object' && !visited.has(current)) {
visited.add(current);
const errorName = current.name;
const errorCode = current.code;
const errorMessage = typeof current.message === 'string' ? current.message : '';
if (
errorName === 'AbortError' ||
errorCode === 'ABORT_ERR' ||
errorMessage.includes('AbortError') ||
/(?:operation|request|stream) was aborted/i.test(errorMessage)
) {
return true;
}
current = current.cause;
}
return false;
};
/**
* Spend tokens for all models from collected usage.
* This handles both sequential and parallel agent execution.

6
api/server/middleware/abortMiddleware.spec.js
View file

@ -31,6 +31,9 @@ jest.mock('@librechat/data-schemas', () => ({
}));
jest.mock('@librechat/api', () => ({
/** Real implementation: these tests exist to verify abort classification
* itself, so mocking it would assert the mock rather than the behavior. */
isAbortError: jest.requireActual('@librechat/api').isAbortError,
countTokens: jest.fn().mockResolvedValue(100),
isEnabled: jest.fn().mockReturnValue(false),
sendEvent: jest.fn(),
@ -44,6 +47,9 @@ jest.mock('@librechat/api', () => ({
}));
jest.mock('librechat-data-provider', () => ({
/** Keep the module real: `@librechat/api` is partially un-mocked above and
* reads constants (`CacheKeys`, ...) from it at import time. */
...jest.requireActual('librechat-data-provider'),
isAssistantsEndpoint: jest.fn().mockReturnValue(false),
ErrorTypes: { INVALID_REQUEST: 'INVALID_REQUEST', NO_SYSTEM_MESSAGES: 'NO_SYSTEM_MESSAGES' },
}));

37
api/server/services/ActionService.js
View file

@ -267,22 +267,24 @@ async function createActionTool({
config?.signal,
);
logger.debug('Waiting for OAuth Authorization response', { action_id, identifier });
const result = await flowManager.createFlow(
identifier,
'oauth',
{
state: stateToken,
userId: userId,
client_url: metadata.auth.client_url,
redirect_uri: `${process.env.DOMAIN_SERVER}/api/actions/${action_id}/oauth/callback`,
token_exchange_method: metadata.auth.token_exchange_method,
allowedAddresses,
/** Encrypted values */
encrypted_oauth_client_id: encrypted.oauth_client_id,
encrypted_oauth_client_secret: encrypted.oauth_client_secret,
},
config?.signal,
);
/** Deliberately unsignalled. This key is `userId:action_id`, so a
* second run for the same action joins this very flow, and the
* browser's OAuth callback reads its metadata to exchange the
* code. `monitorFlow` deletes the key when a waiter's signal
* aborts, so forwarding the run signal here would let one Stop
* strand the other run and discard an authorization the user
* already granted. Run-scoped flows below still take it. */
const result = await flowManager.createFlow(identifier, 'oauth', {
state: stateToken,
userId: userId,
client_url: metadata.auth.client_url,
redirect_uri: `${process.env.DOMAIN_SERVER}/api/actions/${action_id}/oauth/callback`,
token_exchange_method: metadata.auth.token_exchange_method,
allowedAddresses,
/** Encrypted values */
encrypted_oauth_client_id: encrypted.oauth_client_id,
encrypted_oauth_client_secret: encrypted.oauth_client_secret,
});
logger.debug('Received OAuth Authorization response', { action_id, identifier });
data.delta.auth = undefined;
data.delta.expires_at = undefined;
@ -351,11 +353,12 @@ async function createActionTool({
);
const flowsCache = getLogStores(CacheKeys.FLOWS);
const flowManager = getActionFlowStateManager(flowsCache);
/** Also shared across this user's runs for the action; see the
* authorization flow above for why it stays unsignalled. */
const refreshData = await flowManager.createFlowWithHandler(
`${identifier}:refresh`,
'oauth_refresh',
refreshTokens,
config?.signal,
);
metadata.oauth_access_token = refreshData.access_token;
if (refreshData.refresh_token) {

82
api/server/services/ActionService.spec.js
View file

@ -315,6 +315,88 @@ describe('createActionTool OAuth events', () => {
});
});
describe('createActionTool OAuth flow cancellation', () => {
/**
* `monitorFlow` deletes the flow key when a waiter's signal aborts. The
* authorization and refresh flows are keyed `userId:action_id`, so a second
* run for the same action joins the very same record and the browser's OAuth
* callback reads its metadata to exchange the code — one Stop must not strand
* either of them. Only the run-scoped login flow may carry the run signal.
*/
it('withholds the run signal from flows shared beyond this run', async () => {
const preparedExecutor = {
setAuth: jest.fn().mockResolvedValue(undefined),
execute: jest.fn().mockResolvedValue({ data: { ok: true } }),
};
const requestBuilder = {
createExecutor: jest.fn(() => ({
setParams: jest.fn(() => preparedExecutor),
})),
};
mockFindToken.mockResolvedValue(null);
mockActionFlowManager.createFlowWithHandler.mockImplementation(
async (_flowId, _type, handler) => handler(),
);
mockActionFlowManager.createFlow.mockResolvedValue({
access_token: 'access-token',
refresh_token: 'refresh-token',
expires_in: 3600,
});
const actionTool = await createActionTool({
userId: 'action-user',
res: {},
action: {
action_id: 'action-1',
metadata: {
domain: 'https://api.example.com',
oauth_client_id: 'client-id',
auth: {
type: 'oauth',
authorization_url: 'https://auth.example.com/authorize',
client_url: 'https://auth.example.com/token',
scope: 'read',
},
},
},
requestBuilder,
encrypted: {
oauth_client_id: 'encrypted-client-id',
oauth_client_secret: 'encrypted-client-secret',
},
streamId: 'action-oauth-stream',
jobCreatedAt: 1234,
});
const signal = new AbortController().signal;
await actionTool._call(
{},
{
signal,
metadata: { thread_id: 'thread-1', run_id: 'run-1' },
toolCall: {
id: 'tool-call-1',
stepId: 'step-1',
name: 'action-tool',
type: 'tool_call',
},
},
);
const [sharedFlowId, sharedType, , sharedSignal] =
mockActionFlowManager.createFlow.mock.calls[0];
expect(sharedFlowId).toBe('action-user:action-1');
expect(sharedType).toBe('oauth');
expect(sharedSignal).toBeUndefined();
const loginCall = mockActionFlowManager.createFlowWithHandler.mock.calls.find(
([, type]) => type === 'oauth_login',
);
expect(loginCall[0]).toBe('action-user:action-1:oauth_login:thread-1:run-1');
expect(loginCall[3]).toBe(signal);
});
});
describe('validateAndUpdateTool', () => {
const mockReq = { user: { id: 'user123' } };

11
api/server/services/MCP.js
View file

@ -40,6 +40,7 @@ const {
containsGraphTokenPlaceholder,
createAuthIdentityContext,
isOAuthServer,
isAbortError,
OpenIDReauthRequiredError,
} = require('@librechat/api');
const {
@ -1308,10 +1309,12 @@ function createToolInstance({
}
return result;
} catch (error) {
/** A user Stop aborts every in-flight call at once. The rejection that
* follows is the cancellation working, so it must not reach error-level
* operational alerts; the wrapping below still reports it to the turn. */
if (config?.signal?.aborted === true) {
/** A user Stop aborts every in-flight call at once, and that rejection is
* the cancellation working, so it must not reach error-level operational
* alerts; the wrapping below still reports it to the turn. The error has
* to look like an abort as well: a permission, OAuth, or upstream failure
* can reject in the same tick as the Stop and must stay visible. */
if (config?.signal?.aborted === true && isAbortError(error)) {
logger.debug(
`[MCP][${serverName}][${toolName}][User: ${userId}] Tool call cancelled by user abort`,
);

70
api/server/services/MCP.spec.js
View file

@ -1627,13 +1627,7 @@ describe('User parameter passing tests', () => {
({ options }) =>
new Promise((_resolve, reject) => {
const signal = options?.signal;
signal?.addEventListener(
'abort',
() => reject(new Error('This operation was aborted')),
{
once: true,
},
);
signal?.addEventListener('abort', () => reject(signal.reason), { once: true });
}),
),
});
@ -1680,6 +1674,68 @@ describe('User parameter passing tests', () => {
);
});
it('keeps a real failure racing the Stop at error level', async () => {
const mockUser = { id: 'race-user', role: 'USER' };
const mockRes = { write: jest.fn(), flush: jest.fn() };
const abortController = new AbortController();
const { getRoleByName } = require('~/models');
getRoleByName.mockResolvedValue({
permissions: {
[PermissionTypes.MCP_SERVERS]: {
[Permissions.USE]: true,
},
},
});
mockGetMCPManager.mockReturnValue({
callTool: jest.fn(
({ options }) =>
new Promise((_resolve, reject) => {
options?.signal?.addEventListener(
'abort',
() => reject(new Error('upstream 503 from the MCP server')),
{ once: true },
);
}),
),
});
const mcpTool = await createMCPTool({
res: mockRes,
user: mockUser,
config: { url: 'https://race.example.com/mcp' },
toolKey: `test-tool${D}test-server`,
provider: 'openai',
userMCPAuthMap: {},
availableTools: {
[`test-tool${D}test-server`]: {
function: {
description: 'Cached tool',
parameters: { type: 'object', properties: {} },
},
},
},
});
const call = mcpTool.invoke(
{},
{
signal: abortController.signal,
configurable: { user: mockUser },
metadata: { provider: 'openai', thread_id: 'thread-1', run_id: 'run-1' },
toolCall: {},
},
);
await new Promise((resolve) => setImmediate(resolve));
abortController.abort();
await expect(call).rejects.toThrow();
await new Promise((resolve) => setImmediate(resolve));
expect(logger.error).toHaveBeenCalledWith(
expect.stringContaining('Error calling MCP tool'),
expect.anything(),
);
});
it.each(['OAuth flow initiated - return early', 'Pending OAuth flow reused - return early'])(
'preserves runtime-detected OAuth for the internal signal: %s',
async (oauthSignal) => {

143
packages/api/src/agents/handlers.spec.ts
View file

@ -310,41 +310,53 @@ describe('createToolExecuteHandler', () => {
});
describe('run cancellation', () => {
it('forwards the batch abort signal into foreground tool invocations', async () => {
const capturedSignals: (AbortSignal | undefined)[] = [];
const controller = new AbortController();
const tool = {
name: 'slow_tool',
invoke: jest.fn(async (_args: unknown, config: Record<string, unknown>) => {
capturedSignals.push(config.signal as AbortSignal | undefined);
await new Promise<void>((resolve, reject) => {
const signal = config.signal as AbortSignal | undefined;
if (signal == null) {
setTimeout(resolve, 50);
return;
}
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
});
return { content: 'never' };
}),
/** Production aborts with no reason, yielding a DOMException named
* `AbortError` — the shape every cancellation check downstream keys on. */
function abortingTool(name = 'slow_tool') {
return {
name,
invoke: jest.fn(
(_args: unknown, config: Record<string, unknown>) =>
new Promise((_resolve, reject) => {
const signal = config.signal as AbortSignal | undefined;
if (signal == null) {
setTimeout(() => reject(new Error('never aborted')), 50);
return;
}
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
}),
),
};
}
function runBatch(
tool: { name: string; invoke: jest.Mock },
request: Partial<ToolExecuteBatchRequest>,
controller: AbortController,
): Promise<ToolExecuteResult[]> {
const loadTools: ToolExecuteOptions['loadTools'] = jest.fn(async () => ({
loadedTools: [tool] as never[],
}));
const handler = createToolExecuteHandler({ loadTools });
const results = await new Promise<ToolExecuteResult[]>((resolve, reject) => {
const request: ToolExecuteBatchRequest = {
toolCalls: [{ id: 'call-1', name: 'slow_tool', args: {} }] as ToolCallRequest[],
return new Promise<ToolExecuteResult[]>((resolve, reject) => {
handler.handle('on_tool_execute', {
toolCalls: [{ id: 'call-1', name: tool.name, args: {} }] as ToolCallRequest[],
signal: controller.signal,
...request,
resolve,
reject,
};
handler.handle('on_tool_execute', request);
setTimeout(() => controller.abort(new Error('Operation aborted by user')), 10);
} as ToolExecuteBatchRequest);
setTimeout(() => controller.abort(), 10);
});
}
expect(capturedSignals[0]).toBe(controller.signal);
it('forwards the batch abort signal into foreground tool invocations', async () => {
const controller = new AbortController();
const tool = abortingTool();
const results = await runBatch(tool, {}, controller);
expect(tool.invoke.mock.calls[0][1].signal).toBe(controller.signal);
expect(results).toHaveLength(1);
expect(results[0].status).toBe('error');
});
@ -352,31 +364,8 @@ describe('createToolExecuteHandler', () => {
it('logs a cancelled tool call as debug rather than a tool error', async () => {
const errorSpy = jest.spyOn(logger, 'error').mockReturnValue(logger);
const controller = new AbortController();
const tool = {
name: 'slow_tool',
invoke: jest.fn(
(_args: unknown, config: Record<string, unknown>) =>
new Promise((_resolve, reject) => {
const signal = config.signal as AbortSignal;
signal.addEventListener('abort', () => reject(signal.reason), { once: true });
}),
),
};
const loadTools: ToolExecuteOptions['loadTools'] = jest.fn(async () => ({
loadedTools: [tool] as never[],
}));
const handler = createToolExecuteHandler({ loadTools });
await new Promise<ToolExecuteResult[]>((resolve, reject) => {
const request: ToolExecuteBatchRequest = {
toolCalls: [{ id: 'call-1', name: 'slow_tool', args: {} }] as ToolCallRequest[],
signal: controller.signal,
resolve,
reject,
};
handler.handle('on_tool_execute', request);
setTimeout(() => controller.abort(new Error('Operation aborted by user')), 10);
});
await runBatch(abortingTool(), {}, controller);
expect(
errorSpy.mock.calls.filter(([message]) =>
@ -386,9 +375,39 @@ describe('createToolExecuteHandler', () => {
});
/**
* An aborted run means the turn is over, not that this particular rejection
* was the cancellation — an unrelated failure can always race the Stop. The
* quiet-log branch must therefore never become a way around output filtering.
* An aborted run says the turn is over, not that this rejection was the
* cancellation. A genuine failure that lands in the same tick as the Stop
* must stay visible to operational logging.
*/
it('keeps an unrelated failure racing the Stop at error level', async () => {
const errorSpy = jest.spyOn(logger, 'error').mockReturnValue(logger);
const controller = new AbortController();
const tool = {
name: 'slow_tool',
invoke: jest.fn(
(_args: unknown, config: Record<string, unknown>) =>
new Promise((_resolve, reject) => {
const signal = config.signal as AbortSignal;
signal.addEventListener(
'abort',
() => reject(new Error('upstream 503 from the tool backend')),
{ once: true },
);
}),
),
};
await runBatch(tool, {}, controller);
expect(
errorSpy.mock.calls.filter(([message]) =>
String(message).includes('[ON_TOOL_EXECUTE] Tool slow_tool error'),
),
).toHaveLength(1);
});
/**
* The quiet-log branch must never double as a way around output filtering.
*/
it('still filters a tool failure that rejects after the run was aborted', async () => {
const protectedValue = 'PROTECTED-CANCELLED-TOOL-OUTPUT';
@ -405,22 +424,12 @@ describe('createToolExecuteHandler', () => {
}),
),
};
const loadTools: ToolExecuteOptions['loadTools'] = jest.fn(async () => ({
loadedTools: [tool] as never[],
}));
const handler = createToolExecuteHandler({ loadTools });
const [result] = await new Promise<ToolExecuteResult[]>((resolve, reject) => {
const request: ToolExecuteBatchRequest = {
toolCalls: [{ id: 'call-1', name: 'slow_tool', args: {} }] as ToolCallRequest[],
configurable: { req: protectedToolOutputRequest() },
signal: controller.signal,
resolve,
reject,
};
handler.handle('on_tool_execute', request);
setTimeout(() => controller.abort(new Error('Operation aborted by user')), 10);
});
const [result] = await runBatch(
tool,
{ configurable: { req: protectedToolOutputRequest() } },
controller,
);
expect(result.status).toBe('error');
expect(result.errorMessage).toContain('content_filter_block');

19
packages/api/src/agents/handlers.ts
View file

@ -78,7 +78,13 @@ import {
stripIntentLabelsFromToolDefinitions,
INTENT_ARG,
} from './intent';
import { getSafeErrorMetadata, logAxiosError, runOutsideTracing, truncateMiddle } from '~/utils';
import {
isAbortError,
logAxiosError,
truncateMiddle,
runOutsideTracing,
getSafeErrorMetadata,
} from '~/utils';
import { buildSkillPrimeMessage, isSkillFilePath, SKILL_FILE_PREFIX } from './skills';
import { resolveCallerCapabilityProjectionSnapshot } from './callerCapabilities';
import { createSkillContentDigest } from './compatibility';
@ -5998,12 +6004,13 @@ export function createToolExecuteHandler(options: ToolExecuteOptions): EventHand
const { message, logContext } = getSafeToolError(toolError);
/** A user Stop rejects every in-flight call at once. That is
* the abort working, not a fault, so it is logged at debug.
* It only ever changes the log level: an aborted run says
* the turn is over, not that this particular rejection was
* the cancellation, so an unrelated failure racing the Stop
* must still take the same filtering and result path. */
* An aborted run says the turn is over, not that THIS
* rejection was the cancellation, so the error must look
* like one too; an unrelated failure racing the Stop stays
* at error level. Either way the level is all that changes
* — filtering and the result shape are identical. */
const logToolFailure = (context: Record<string, unknown>): void => {
if (runSignal?.aborted === true) {
if (runSignal?.aborted === true && isAbortError(toolError)) {
logger.debug(
`[ON_TOOL_EXECUTE] Tool ${tc.name} cancelled by run abort`,
context,

43
packages/api/src/utils/errors.spec.ts
View file

@ -1,4 +1,4 @@
import { getSafeErrorMetadata } from './errors';
import { getSafeErrorMetadata, isAbortError } from './errors';
describe('getSafeErrorMetadata', () => {
it('keeps bounded diagnostic fields without serializing raw provider data', () => {
@ -51,3 +51,44 @@ describe('getSafeErrorMetadata', () => {
).toEqual({ type: 'UnknownError' });
});
});
describe('isAbortError', () => {
it('recognizes the DOMException a bare abort() produces', () => {
const controller = new AbortController();
controller.abort();
expect(isAbortError(controller.signal.reason)).toBe(true);
});
it('recognizes an abort rewrapped by an intermediate layer', () => {
const controller = new AbortController();
controller.abort();
const wrapped = new Error('[MCP][server][tool] tool call failed', {
cause: controller.signal.reason,
});
expect(isAbortError(wrapped)).toBe(true);
});
it('recognizes the SDK message shape that only stringifies the reason', () => {
expect(
isAbortError(new Error('MCP error -32001: AbortError: This operation was aborted')),
).toBe(true);
});
it('does not treat an unrelated failure as a cancellation', () => {
expect(isAbortError(new Error('upstream 503 from the tool backend'))).toBe(false);
expect(isAbortError(Object.assign(new Error('forbidden'), { code: 'EPERM' }))).toBe(false);
});
it('terminates on a cyclic cause chain', () => {
const first = new Error('first') as Error & { cause?: unknown };
const second = new Error('second') as Error & { cause?: unknown };
first.cause = second;
second.cause = first;
expect(isAbortError(first)).toBe(false);
});
it('tolerates non-error values', () => {
expect(isAbortError(undefined)).toBe(false);
expect(isAbortError('aborted')).toBe(false);
});
});

36
packages/api/src/utils/errors.ts
View file

@ -38,3 +38,39 @@ export function getSafeErrorMetadata(error: unknown): SafeErrorMetadata {
...(status !== undefined && { status }),
};
}
/**
* Whether a caught error is a cancellation rather than a genuine failure.
*
* An aborted signal on its own proves only that the run is over: a permission,
* OAuth, or upstream error can reject in the same tick a user presses Stop, and
* treating those as cancellations hides real faults from operational alerts.
* Callers that want to quiet a cancellation should require this as well as the
* signal state. Walks `cause` because intermediate layers rewrap the rejection.
*/
export function isAbortError(error: unknown): boolean {
const visited = new Set<unknown>();
let current: unknown = error;
while (current != null && typeof current === 'object' && !visited.has(current)) {
visited.add(current);
const name = readProperty(current, 'name');
const code = readProperty(current, 'code');
const messageValue = readProperty(current, 'message');
const message = typeof messageValue === 'string' ? messageValue : '';
if (
name === 'AbortError' ||
code === 'ABORT_ERR' ||
message.includes('AbortError') ||
/(?:operation|request|stream) was aborted/i.test(message)
) {
return true;
}
current = readProperty(current, 'cause');
}
return false;
}