LibreChat/api/server/middleware/abortMiddleware.spec.js
Danny Avila 0219513115
🛑 fix: Forward Run Abort Signal to Foreground Tool Calls (#15517)
* 🛑 fix: Forward Run Abort Signal to Foreground Tool Calls

Stopping a generation left in-flight tool calls running on the far side.
For MCP over streamable-http the external server kept burning CPU until
the tool finished or timed out, with no `notifications/cancelled` ever
sent.

The MCP layer was never the problem: `MCPManager.callTool` already
spreads `options.signal` into `client.request`, and the SDK's
`Protocol.request` sends `notifications/cancelled` the moment that signal
aborts. The signal simply never arrived.

Agent tools do not execute inside LangGraph's ToolNode — `eventDrivenMode`
dispatches `on_tool_execute` and `createToolExecuteHandler` invokes them.
The agents SDK puts the run's abort signal on the batch request
(`ToolExecuteBatchRequest.signal`, documented as one handlers should
forward), but the handler destructured every field except `signal` and
built its invoke config without it. `config.signal` was therefore
`undefined` in every foreground tool, so `createMCPTool` derived no
signal and the SDK registered no abort listener.

Forward it. This restores cancellation for all signal-aware foreground
tools, not just MCP. The detached background invoke keeps its own
controller, since that work deliberately outlives the turn.

An aborted call now rejects promptly and resolves an error result rather
than hanging, so the abort is logged at debug instead of as a tool error
— a user pressing Stop should not spray the error log.

* 🔇 fix: Keep Cancelled Tool Calls on the Filtered Path and Out of Error Logs

Addresses both findings from the Codex review of 48fded564a.

P1 — the cancellation branch returned before `filteredToolOutputResult`,
so with tool-output filtering configured an aborted call's error text
reached the turn uninspected. Worse, `runSignal.aborted` says the run is
over, not that this rejection was the cancellation: an unrelated failure
racing the Stop took the same unfiltered exit. Cancellation now only
selects the log level; filtering and the result shape are unchanged.

P2 — `createMCPTool` logs its own error before wrapping and rethrowing,
so every Stop during an MCP call still emitted an error-level MCP entry
and could pollute operational alerts. That catch now recognizes the abort
and logs at debug; the wrapped message still reaches the turn.

Both are covered by tests that fail against the previous commit.

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

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.

* 🧷 fix: Detach the Stopped OAuth Waiter Instead of Dropping Its Signal

Addresses the Codex review of d53ff170ab.

Withholding the run signal from the shared Action OAuth flows protected
concurrent runs and the browser callback, but left this invocation
attached to the flow after the Stop. A callback landing late then resumed
`_call` straight into `preparedExecutor.execute` — a consequential API
request running after the user had already stopped the turn. Handing the
signal back is not the answer either, since `monitorFlow` deletes the
shared `userId:action_id` key on abort.

`detachOnAbort` separates the two: the caller stops waiting the moment
the signal aborts, while the shared flow runs on for whoever else needs
it — another run, or the callback exchanging its code. Late settlement of
the detached work is swallowed rather than surfacing as an unhandled
rejection.

Both shared flows now use it. The run-scoped login flow keeps the signal
directly, as nothing outside its run observes it.

* 🪪 fix: Keep an Action OAuth Abort an Abort

Addresses two of the three findings from the Codex review of 70fc0782b1.

Detaching the stopped waiter left the rejection to be misread on the way
out. The refresh catch treated it as a failed refresh and called
`requestLogin()`, emitting an OAuth prompt and opening pending
authorization state for a turn that had already ended. `requestLogin` and
the surrounding auth catch relabelled it `Failed to authenticate OAuth
tool` / `Authentication failed`, and the outer catch handed that to
`logAxiosError` — so an ordinary Stop produced two error-level entries
and returned failure text as the tool's result. Each boundary now lets an
abort through unchanged, and the outer catch logs it at debug and
rethrows so it stays a cancellation end to end.

Deferred: waiter-only cancellation in `FlowStateManager`, so a detached
`monitorFlow` stops polling instead of running out the flow's TTL. That
poller is bounded at three minutes and lives exactly as long as it does
for a waiter that never left, so it changes no lifetime this branch
introduced; giving `FlowStateManager` a non-destructive abort mode is its
own change, shared with MCP and indexSync.

* 🤫 fix: Classify the Abort at the MCPManager Boundary Too

Addresses the Codex review of 9bcf1aa57e.

`MCPManager.callTool` logs every rejection at error level before
rethrowing to `createMCPTool`, so a user Stop still emitted an MCP error
from the inner boundary even after the outer one learned to recognize it.
The same signal-plus-abort-shape classification now applies here.

The review also noted, correctly, that the existing test could not have
caught this: it replaces `getMCPManager` wholesale, so it only ever
exercised the outer logger. The new cases drive the real manager with a
connection whose `client.request` rejects the way the SDK does once a
request signal aborts, and cover both the cancellation and a genuine
failure racing the Stop.

* 🧹 chore: Restore Import Order in the Agents Tool Handler
2026-09-02 13:32:33 -04:00

427 lines
14 KiB
JavaScript

/**
* Tests for abortMiddleware - spendCollectedUsage function
*
* This tests the token spending logic for abort scenarios,
* particularly for parallel agents (addedConvo) where multiple
* models need their tokens spent.
*
* spendCollectedUsage delegates to recordCollectedUsage from @librechat/api,
* passing pricing + bulkWriteOps deps, with context: 'abort'.
* After spending, it clears the collectedUsage array to prevent double-spending
* from the AgentClient finally block (which shares the same array reference).
*/
const mockSpendTokens = jest.fn().mockResolvedValue();
const mockSpendStructuredTokens = jest.fn().mockResolvedValue();
const mockRecordCollectedUsage = jest
.fn()
.mockResolvedValue({ input_tokens: 100, output_tokens: 50 });
const mockGetMultiplier = jest.fn().mockReturnValue(1);
const mockGetCacheMultiplier = jest.fn().mockReturnValue(null);
const mockGetTransactionsConfig = jest.fn().mockReturnValue({ enabled: false });
jest.mock('@librechat/data-schemas', () => ({
logger: {
debug: jest.fn(),
error: jest.fn(),
warn: jest.fn(),
info: jest.fn(),
},
}));
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(),
GenerationJobManager: {
abortJob: jest.fn(),
},
recordCollectedUsage: mockRecordCollectedUsage,
getTransactionsConfig: (...args) => mockGetTransactionsConfig(...args),
sanitizeMessageForTransmit: jest.fn((msg) => msg),
buildAbortedResponseMetadata: jest.fn().mockReturnValue(null),
}));
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' },
}));
jest.mock('~/app/clients/prompts', () => ({
truncateText: jest.fn((text) => text),
smartTruncateText: jest.fn((text) => text),
}));
jest.mock('~/cache/clearPendingReq', () => jest.fn().mockResolvedValue());
jest.mock('~/server/middleware/error', () => ({
sendError: jest.fn(),
}));
const mockUpdateBalance = jest.fn().mockResolvedValue({});
const mockBulkInsertTransactions = jest.fn().mockResolvedValue(undefined);
jest.mock('~/models', () => ({
saveMessage: jest.fn().mockResolvedValue(),
getConvo: jest.fn().mockResolvedValue({ title: 'Test Chat' }),
updateBalance: mockUpdateBalance,
bulkInsertTransactions: mockBulkInsertTransactions,
spendTokens: (...args) => mockSpendTokens(...args),
spendStructuredTokens: (...args) => mockSpendStructuredTokens(...args),
getMultiplier: mockGetMultiplier,
getCacheMultiplier: mockGetCacheMultiplier,
}));
jest.mock('./abortRun', () => ({
abortRun: jest.fn(),
}));
const { logger } = require('@librechat/data-schemas');
const { sendError } = require('~/server/middleware/error');
const { GenerationJobManager } = require('@librechat/api');
const db = require('~/models');
const { handleAbort, handleAbortError, spendCollectedUsage } = require('./abortMiddleware');
const buildAbortRequest = () => ({
body: {
model: 'gpt-4',
},
user: {
id: 'user-123',
},
});
describe('abortMiddleware - spendCollectedUsage', () => {
beforeEach(() => {
jest.clearAllMocks();
});
describe('spendCollectedUsage delegation', () => {
it('should return early if collectedUsage is empty', async () => {
await spendCollectedUsage({
userId: 'user-123',
conversationId: 'convo-123',
collectedUsage: [],
fallbackModel: 'gpt-4',
});
expect(mockRecordCollectedUsage).not.toHaveBeenCalled();
});
it('should return early if collectedUsage is null', async () => {
await spendCollectedUsage({
userId: 'user-123',
conversationId: 'convo-123',
collectedUsage: null,
fallbackModel: 'gpt-4',
});
expect(mockRecordCollectedUsage).not.toHaveBeenCalled();
});
it('should call recordCollectedUsage with abort context and full deps', async () => {
const collectedUsage = [{ input_tokens: 100, output_tokens: 50, model: 'gpt-4' }];
await spendCollectedUsage({
userId: 'user-123',
conversationId: 'convo-123',
collectedUsage,
fallbackModel: 'gpt-4',
messageId: 'msg-123',
});
expect(mockRecordCollectedUsage).toHaveBeenCalledTimes(1);
expect(mockRecordCollectedUsage).toHaveBeenCalledWith(
{
spendTokens: expect.any(Function),
spendStructuredTokens: expect.any(Function),
pricing: {
getMultiplier: mockGetMultiplier,
getCacheMultiplier: mockGetCacheMultiplier,
},
bulkWriteOps: {
insertMany: mockBulkInsertTransactions,
updateBalance: mockUpdateBalance,
},
},
{
user: 'user-123',
conversationId: 'convo-123',
collectedUsage,
context: 'abort',
messageId: 'msg-123',
model: 'gpt-4',
},
);
});
it('should pass context abort for multiple models (parallel agents)', async () => {
const collectedUsage = [
{ input_tokens: 100, output_tokens: 50, model: 'gpt-4' },
{ input_tokens: 80, output_tokens: 40, model: 'claude-3' },
{ input_tokens: 120, output_tokens: 60, model: 'gemini-pro' },
];
await spendCollectedUsage({
userId: 'user-123',
conversationId: 'convo-123',
collectedUsage,
fallbackModel: 'gpt-4',
});
expect(mockRecordCollectedUsage).toHaveBeenCalledTimes(1);
expect(mockRecordCollectedUsage).toHaveBeenCalledWith(
expect.any(Object),
expect.objectContaining({
context: 'abort',
collectedUsage,
}),
);
});
it('should handle real-world parallel agent abort scenario', async () => {
const collectedUsage = [
{ input_tokens: 31596, output_tokens: 151, model: 'gemini-3-flash-preview' },
{ input_tokens: 28000, output_tokens: 120, model: 'gpt-5.2' },
];
await spendCollectedUsage({
userId: 'user-123',
conversationId: 'convo-123',
collectedUsage,
fallbackModel: 'gemini-3-flash-preview',
});
expect(mockRecordCollectedUsage).toHaveBeenCalledTimes(1);
expect(mockRecordCollectedUsage).toHaveBeenCalledWith(
expect.any(Object),
expect.objectContaining({
user: 'user-123',
conversationId: 'convo-123',
context: 'abort',
model: 'gemini-3-flash-preview',
}),
);
});
/**
* Race condition prevention: after abort middleware spends tokens,
* the collectedUsage array is cleared so AgentClient.recordCollectedUsage()
* (which shares the same array reference) sees an empty array and returns early.
*/
it('should clear collectedUsage array after spending to prevent double-spending', async () => {
const collectedUsage = [
{ input_tokens: 100, output_tokens: 50, model: 'gpt-4' },
{ input_tokens: 80, output_tokens: 40, model: 'claude-3' },
];
expect(collectedUsage.length).toBe(2);
await spendCollectedUsage({
userId: 'user-123',
conversationId: 'convo-123',
collectedUsage,
fallbackModel: 'gpt-4',
});
expect(mockRecordCollectedUsage).toHaveBeenCalledTimes(1);
expect(collectedUsage.length).toBe(0);
});
it('should await recordCollectedUsage before clearing array', async () => {
let resolved = false;
mockRecordCollectedUsage.mockImplementation(async () => {
await new Promise((resolve) => setTimeout(resolve, 10));
resolved = true;
return { input_tokens: 100, output_tokens: 50 };
});
const collectedUsage = [
{ input_tokens: 100, output_tokens: 50, model: 'gpt-4' },
{ input_tokens: 80, output_tokens: 40, model: 'claude-3' },
];
await spendCollectedUsage({
userId: 'user-123',
conversationId: 'convo-123',
collectedUsage,
fallbackModel: 'gpt-4',
});
expect(resolved).toBe(true);
expect(collectedUsage.length).toBe(0);
});
});
});
describe('abortMiddleware - handleAbortError', () => {
beforeEach(() => {
jest.clearAllMocks();
});
it.each([
[
'native DOMException AbortError',
new DOMException('The operation was aborted', 'AbortError'),
'AbortError',
],
[
'wrapped AbortError message',
new Error('SSE stream disconnected: AbortError: The operation was aborted'),
'Error',
],
[
'cause-nested AbortError',
new Error('Request failed', {
cause: new DOMException('The operation was aborted', 'AbortError'),
}),
'Error',
],
])('logs a %s as a debug event instead of an error', async (_label, error, name) => {
await handleAbortError({}, buildAbortRequest(), error, {
sender: 'AI',
conversationId: 'convo-123',
messageId: 'message-123',
parentMessageId: 'parent-123',
userMessageId: 'user-message-123',
});
expect(logger.error).not.toHaveBeenCalled();
expect(logger.debug).toHaveBeenCalledWith('[handleAbortError] AI response aborted by user', {
conversationId: 'convo-123',
code: error.code,
name,
message: error.message,
});
expect(sendError).toHaveBeenCalledTimes(1);
});
it('keeps unexpected generation errors classified as errors', async () => {
const error = new Error('Provider failed');
await handleAbortError({}, buildAbortRequest(), error, {
sender: 'AI',
conversationId: 'convo-123',
messageId: 'message-123',
parentMessageId: 'parent-123',
userMessageId: 'user-message-123',
});
expect(logger.error).toHaveBeenCalledWith(
'[handleAbortError] AI response error; aborting request:',
error,
);
expect(logger.debug).not.toHaveBeenCalled();
expect(sendError).toHaveBeenCalledTimes(1);
});
});
/**
* The transactions config is resolved from the request's app config and must reach
* every write path in this file. `createTransaction` reads `transactions` from the
* caller-supplied data, so an omitted value is indistinguishable from enabled and
* the write proceeds even when `transactions.enabled` is false.
*/
describe('abortMiddleware - transactions config', () => {
const buildJobData = () => ({
model: 'gpt-4',
responseMessageId: 'msg-123',
conversationId: 'convo-123',
endpoint: 'agents',
sender: 'AI',
promptTokens: 25,
userMessage: {
messageId: 'user-msg-123',
parentMessageId: 'parent-123',
conversationId: 'convo-123',
text: 'hello',
},
});
const buildReq = () => ({
body: { abortKey: 'convo-123:1', endpoint: 'agents' },
user: { id: 'user-123', email: 'user@example.com' },
config: { transactions: { enabled: false } },
});
const buildRes = () => ({
headersSent: false,
setHeader: jest.fn(),
send: jest.fn(),
});
beforeEach(() => {
jest.clearAllMocks();
mockGetTransactionsConfig.mockReturnValue({ enabled: false });
mockRecordCollectedUsage.mockResolvedValue({ input_tokens: 100, output_tokens: 50 });
db.getConvo.mockResolvedValue({ title: 'Test Chat' });
});
it('forwards transactions through spendCollectedUsage to recordCollectedUsage', async () => {
const collectedUsage = [{ input_tokens: 100, output_tokens: 50, model: 'gpt-4' }];
await spendCollectedUsage({
userId: 'user-123',
conversationId: 'convo-123',
collectedUsage,
fallbackModel: 'gpt-4',
transactions: { enabled: false },
});
expect(mockRecordCollectedUsage).toHaveBeenCalledTimes(1);
expect(mockRecordCollectedUsage).toHaveBeenCalledWith(
expect.any(Object),
expect.objectContaining({ context: 'abort', transactions: { enabled: false } }),
);
});
it('resolves the config from req and forwards it on the collected-usage path', async () => {
const collectedUsage = [{ input_tokens: 100, output_tokens: 50, model: 'gpt-4' }];
GenerationJobManager.abortJob.mockResolvedValue({
success: true,
jobData: buildJobData(),
content: [],
text: 'partial',
collectedUsage,
});
const req = buildReq();
await handleAbort()(req, buildRes());
expect(logger.error).not.toHaveBeenCalled();
expect(mockGetTransactionsConfig).toHaveBeenCalledWith(req.config);
expect(mockRecordCollectedUsage).toHaveBeenCalledTimes(1);
expect(mockRecordCollectedUsage).toHaveBeenCalledWith(
expect.any(Object),
expect.objectContaining({ context: 'abort', transactions: { enabled: false } }),
);
});
it('resolves the config from req and forwards it on the token-count fallback path', async () => {
GenerationJobManager.abortJob.mockResolvedValue({
success: true,
jobData: buildJobData(),
content: [],
text: 'partial',
collectedUsage: [],
});
const req = buildReq();
await handleAbort()(req, buildRes());
expect(logger.error).not.toHaveBeenCalled();
expect(mockGetTransactionsConfig).toHaveBeenCalledWith(req.config);
expect(mockRecordCollectedUsage).not.toHaveBeenCalled();
expect(mockSpendTokens).toHaveBeenCalledTimes(1);
expect(mockSpendTokens).toHaveBeenCalledWith(
expect.objectContaining({ context: 'incomplete', transactions: { enabled: false } }),
expect.any(Object),
);
});
});