diff --git a/api/server/services/ActionService.js b/api/server/services/ActionService.js index 84f0a6dfb4..cd81d76082 100644 --- a/api/server/services/ActionService.js +++ b/api/server/services/ActionService.js @@ -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) { diff --git a/api/server/services/ActionService.spec.js b/api/server/services/ActionService.spec.js index 8771eeb4b5..8f6f9ae83e 100644 --- a/api/server/services/ActionService.spec.js +++ b/api/server/services/ActionService.spec.js @@ -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' } }; diff --git a/packages/api/src/utils/index.ts b/packages/api/src/utils/index.ts index 06e63a62eb..8a12471b3f 100644 --- a/packages/api/src/utils/index.ts +++ b/packages/api/src/utils/index.ts @@ -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'; diff --git a/packages/api/src/utils/promises.spec.ts b/packages/api/src/utils/promises.spec.ts new file mode 100644 index 0000000000..cc3ddf2481 --- /dev/null +++ b/packages/api/src/utils/promises.spec.ts @@ -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((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((_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); + } + }); +}); diff --git a/packages/api/src/utils/promises.ts b/packages/api/src/utils/promises.ts new file mode 100644 index 0000000000..b51a8a9d24 --- /dev/null +++ b/packages/api/src/utils/promises.ts @@ -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(promise: Promise, signal?: AbortSignal | null): Promise { + if (signal == null) { + return promise; + } + + if (signal.aborted) { + promise.catch(() => undefined); + return Promise.reject(signal.reason); + } + + return new Promise((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); + }, + ); + }); +}