From 70fc0782b1a118991c947db745f1f6d57f4e2242 Mon Sep 17 00:00:00 2001 From: Danny Avila Date: Wed, 2 Sep 2026 12:40:43 -0400 Subject: [PATCH] =?UTF-8?q?=F0=9F=A7=B7=20fix:=20Detach=20the=20Stopped=20?= =?UTF-8?q?OAuth=20Waiter=20Instead=20of=20Dropping=20Its=20Signal?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- api/server/services/ActionService.js | 56 +++++++++------ api/server/services/ActionService.spec.js | 88 +++++++++++++++++++++++ packages/api/src/utils/index.ts | 1 + packages/api/src/utils/promises.spec.ts | 70 ++++++++++++++++++ packages/api/src/utils/promises.ts | 39 ++++++++++ 5 files changed, 231 insertions(+), 23 deletions(-) create mode 100644 packages/api/src/utils/promises.spec.ts create mode 100644 packages/api/src/utils/promises.ts 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); + }, + ); + }); +}