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: Detach the Stopped OAuth Waiter Instead of Dropping Its Signal

Browse source 
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.
This commit is contained in:
Danny Avila 2026-09-02 12:40:43 -04:00
parent d53ff170ab
commit 70fc0782b1
5 changed files with 231 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -6,6 +6,7 @@ const { logger, decryptV2 } = require('@librechat/data-schemas');
const {
sendEvent,
logAxiosError,
detachOnAbort,
getTokenExpiresAt,
refreshAccessToken,
GenerationJobManager,
@ -267,24 +268,29 @@ async function createActionTool({
config?.signal,
);
logger.debug('Waiting for OAuth Authorization response', { action_id, identifier });
/** 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,
});
/** Detached rather than signalled. 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, which would strand the other run and discard an
* authorization the user already granted. Stopping only this
* waiter leaves the flow to finish for whoever else needs it,
* while keeping a late authorization from resuming this call
* into the API request below. */
const result = await detachOnAbort(
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,
);
logger.debug('Received OAuth Authorization response', { action_id, identifier });
data.delta.auth = undefined;
data.delta.expires_at = undefined;
@ -354,11 +360,15 @@ 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,
* authorization flow above for why it is detached, not
* signalled. */
const refreshData = await detachOnAbort(
flowManager.createFlowWithHandler(
`${identifier}:refresh`,
'oauth_refresh',
refreshTokens,
),
config?.signal,
);
metadata.oauth_access_token = refreshData.access_token;
if (refreshData.refresh_token) {

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

@ -397,6 +397,94 @@ describe('createActionTool OAuth flow cancellation', () => {
});
});
describe('createActionTool OAuth stop behavior', () => {
/**
* Detaching is not the same as ignoring the Stop: the shared flow survives for
* other waiters and the browser callback, but a late authorization must not
* resume this call into the API request it was gating.
*/
it('does not run the API request when the run is stopped mid-OAuth', 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(),
);
/** The shared flow completes only after the user has pressed Stop, exactly
* as it would when the browser callback lands late. */
let completeFlow;
mockActionFlowManager.createFlow.mockImplementation(
() =>
new Promise((resolve) => {
completeFlow = resolve;
}),
);
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 controller = new AbortController();
const call = actionTool._call(
{},
{
signal: controller.signal,
metadata: { thread_id: 'thread-1', run_id: 'run-1' },
toolCall: {
id: 'tool-call-1',
stepId: 'step-1',
name: 'action-tool',
type: 'tool_call',
},
},
);
await new Promise((resolve) => setImmediate(resolve));
controller.abort();
/** `_call` reports failures through logAxiosError rather than throwing. */
await call;
completeFlow({
access_token: 'access-token',
refresh_token: 'refresh-token',
expires_in: 3600,
});
await new Promise((resolve) => setImmediate(resolve));
expect(preparedExecutor.execute).not.toHaveBeenCalled();
});
});
describe('validateAndUpdateTool', () => {
const mockReq = { user: { id: 'user123' } };

1
packages/api/src/utils/index.ts
View file

@ -6,6 +6,7 @@ export * from './content';
export * from './email';
export * from './env';
export * from './errors';
export * from './promises';
export * from './events';
export * from './files';
export * from './import';

70
packages/api/src/utils/promises.spec.ts Normal file
View file

@ -0,0 +1,70 @@
import { detachOnAbort } from './promises';
describe('detachOnAbort', () => {
it('passes the promise through when no signal is given', async () => {
await expect(detachOnAbort(Promise.resolve('value'))).resolves.toBe('value');
});
it('resolves normally when the signal never aborts', async () => {
const controller = new AbortController();
await expect(detachOnAbort(Promise.resolve('value'), controller.signal)).resolves.toBe('value');
});
it('propagates the underlying rejection when the signal never aborts', async () => {
const controller = new AbortController();
await expect(
detachOnAbort(Promise.reject(new Error('upstream failed')), controller.signal),
).rejects.toThrow('upstream failed');
});
it('rejects with the abort reason and leaves the work running', async () => {
const controller = new AbortController();
let settle!: (value: string) => void;
const shared = new Promise<string>((resolve) => (settle = resolve));
let sharedSettled = false;
shared.then(() => (sharedSettled = true));
const detached = detachOnAbort(shared, controller.signal);
controller.abort();
await expect(detached).rejects.toBe(controller.signal.reason);
expect(sharedSettled).toBe(false);
/** The shared work still completes for whoever else is waiting on it. */
settle('completed after the abort');
await shared;
expect(sharedSettled).toBe(true);
});
it('rejects immediately when the signal is already aborted', async () => {
const controller = new AbortController();
controller.abort();
await expect(detachOnAbort(Promise.resolve('value'), controller.signal)).rejects.toBe(
controller.signal.reason,
);
});
it('does not surface a late rejection of detached work', async () => {
const unhandled: unknown[] = [];
const onUnhandled = (reason: unknown): void => {
unhandled.push(reason);
};
process.on('unhandledRejection', onUnhandled);
try {
const controller = new AbortController();
let fail!: (error: Error) => void;
const shared = new Promise<string>((_resolve, reject) => (fail = reject));
const detached = detachOnAbort(shared, controller.signal);
controller.abort();
await expect(detached).rejects.toBe(controller.signal.reason);
fail(new Error('shared work failed later'));
await new Promise((resolve) => setTimeout(resolve, 10));
expect(unhandled).toHaveLength(0);
} finally {
process.off('unhandledRejection', onUnhandled);
}
});
});

39
packages/api/src/utils/promises.ts Normal file
View file

@ -0,0 +1,39 @@
/**
* Resolves with `promise`, but rejects as soon as `signal` aborts.
*
* Unlike handing the signal to the work itself, the underlying promise keeps
* running. Use it for shared work whose lifetime outlives this caller — an OAuth
* flow keyed by user and action that a concurrent run may also be awaiting and
* that a browser callback will complete. Cancelling such work on one waiter's
* abort strands the others and discards an authorization the user has already
* granted; leaving the caller attached is equally wrong, because the value
* arrives after the user pressed Stop and the caller acts on it.
*
* The detached promise's later settlement is swallowed rather than surfacing as
* an unhandled rejection.
*/
export function detachOnAbort<T>(promise: Promise<T>, signal?: AbortSignal | null): Promise<T> {
if (signal == null) {
return promise;
}
if (signal.aborted) {
promise.catch(() => undefined);
return Promise.reject(signal.reason);
}
return new Promise<T>((resolve, reject) => {
const onAbort = (): void => reject(signal.reason);
signal.addEventListener('abort', onAbort, { once: true });
promise.then(
(value) => {
signal.removeEventListener('abort', onAbort);
resolve(value);
},
(error) => {
signal.removeEventListener('abort', onAbort);
reject(error);
},
);
});
}