diff --git a/.changeset/oauth-refresh-cross-session.md b/.changeset/oauth-refresh-cross-session.md new file mode 100644 index 000000000..45b8ed041 --- /dev/null +++ b/.changeset/oauth-refresh-cross-session.md @@ -0,0 +1,5 @@ +--- +"@executor-js/sdk": patch +--- + +Share the OAuth refresh gate across execution stacks so rotating refresh tokens are not reused. The in-flight gate lived inside a single scoped executor, but a self-host builds a fresh scoped executor per MCP session, so two sessions could each redeem the same stored refresh token. Providers that rotate refresh tokens reject the second redemption and may revoke the whole token family, forcing reauthorization. The gate now hangs off the shared root database handle and its key includes the tenant, so concurrent sessions join one grant. The grant runs on its own fiber that callers join, so a cancelled session no longer interrupts peers waiting on the same refresh. Dedup covers one database handle in one process; multi-replica deployments still need database-backed coordination. diff --git a/e2e/selfhost/oauth-refresh-cross-session.test.ts b/e2e/selfhost/oauth-refresh-cross-session.test.ts new file mode 100644 index 000000000..2f6d57a7b --- /dev/null +++ b/e2e/selfhost/oauth-refresh-cross-session.test.ts @@ -0,0 +1,262 @@ +// Selfhost-only: two MCP sessions that hit an expired token at the same moment +// must share ONE refresh-token grant, never race two. +// +// Issue #1520: the in-flight refresh gate lived inside a single scoped +// executor, but the self-host builds a fresh scoped executor per MCP session, +// so each session believed it was the refresh winner and redeemed the same +// stored refresh token. Providers that rotate refresh tokens answer the second +// redemption with `invalid_grant: refresh token reuse detected` and may revoke +// the whole token family — the connection dies and the user must reauthorize. +// The first refresh cycle succeeds, so the bug stays invisible until a later +// expiry. +// +// The journey: an OpenAPI integration completes a real authorization-code flow +// against a live test AS, the upstream then rejects both sessions' first call +// with a 401 at the same instant (it holds both requests until both arrive), and +// the AS's own request ledger proves exactly one refresh grant was issued and +// both retries carried the same new bearer. +import { randomBytes } from "node:crypto"; +import { createServer, type ServerResponse } from "node:http"; + +import { expect } from "@effect/vitest"; +import { Effect } from "effect"; +import { composePluginApi } from "@executor-js/api/server"; +import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; +import { + AuthTemplateSlug, + ConnectionName, + IntegrationSlug, + OAuthClientSlug, +} from "@executor-js/sdk/shared"; +import { serveOAuthTestServer } from "@executor-js/sdk/testing"; + +import { scenario } from "../src/scenario"; +import { Api, Mcp, Target } from "../src/services"; + +const api = composePluginApi([openApiHttpPlugin()] as const); + +const unique = (prefix: string) => `${prefix}_${randomBytes(4).toString("hex")}`; + +type UpstreamHandle = { + readonly url: string; + readonly bearers: () => readonly string[]; + readonly close: () => void; +}; + +const serveUpstream = () => + Effect.acquireRelease( + Effect.callback((resume) => { + const bearers: string[] = []; + const initialResponses: ServerResponse[] = []; + const server = createServer((request, response) => { + if (request.method === "GET" && (request.url ?? "").startsWith("/issues")) { + bearers.push((request.headers.authorization ?? "").replace(/^Bearer\s+/i, "")); + if (initialResponses.length < 2) { + initialResponses.push(response); + if (initialResponses.length === 2) { + for (const initialResponse of initialResponses) { + initialResponse.writeHead(401, { "content-type": "application/json" }); + initialResponse.end(JSON.stringify({ error: "invalid_token" })); + } + } + return; + } + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ issues: [] })); + return; + } + response.writeHead(404, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: "not_found" })); + }); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + const port = typeof address === "object" && address ? address.port : 0; + resume( + Effect.succeed({ + url: `http://127.0.0.1:${port}`, + bearers: () => [...bearers], + close: () => { + server.close(); + server.closeAllConnections(); + }, + }), + ); + }); + }), + (server) => Effect.sync(server.close), + ); + +const spec = ( + baseUrl: string, + oauth: { readonly authorizationEndpoint: string; readonly tokenEndpoint: string }, +): string => + JSON.stringify({ + openapi: "3.0.3", + info: { title: "Issues API", version: "1.0.0" }, + servers: [{ url: baseUrl }], + paths: { + "/issues": { + get: { + operationId: "listIssues", + security: [{ oauth: ["issues.read"] }], + responses: { "200": { description: "issues" } }, + }, + }, + }, + components: { + securitySchemes: { + oauth: { + type: "oauth2", + flows: { + authorizationCode: { + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + scopes: { "issues.read": "Read issues" }, + }, + }, + }, + }, + }, + }); + +const invokeByAddressCode = (address: string) => ` +const segments = ${JSON.stringify(address)}.split(".").slice(1); +let node = tools; +for (const segment of segments) node = node[segment]; +const result = await node({}); +return JSON.stringify(result); +`; + +const completeAuthorization = (authorizationUrl: string) => + Effect.promise(async () => { + const authorize = await fetch(authorizationUrl, { redirect: "manual" }); + const loginUrl = authorize.headers.get("location"); + if (!loginUrl) throw new Error(`authorize did not redirect: ${authorize.status}`); + const login = await fetch(loginUrl, { + method: "POST", + headers: { authorization: `Basic ${Buffer.from("alice:password").toString("base64")}` }, + redirect: "manual", + }); + const callbackUrl = login.headers.get("location"); + if (!callbackUrl) throw new Error(`login did not redirect: ${login.status}`); + const code = new URL(callbackUrl).searchParams.get("code"); + if (!code) throw new Error("callback carried no authorization code"); + return code; + }); + +scenario( + "OAuth refresh · separate MCP sessions share one rotating-token refresh grant", + { timeout: 180_000 }, + Effect.scoped( + Effect.gen(function* () { + const target = yield* Target; + const { client: makeClient } = yield* Api; + const mcp = yield* Mcp; + const identity = yield* target.newIdentity(); + const client = yield* makeClient(api, identity); + const upstream = yield* serveUpstream(); + const oauth = yield* serveOAuthTestServer({ scopes: ["issues.read"] }); + const slug = unique("refreshcrosssession"); + const clientSlug = OAuthClientSlug.make(unique("refreshcrosssessionc")); + + yield* Effect.ensuring( + Effect.gen(function* () { + yield* client.openapi.addSpec({ + payload: { + spec: { kind: "blob", value: spec(upstream.url, oauth) }, + slug, + baseUrl: upstream.url, + authenticationTemplate: [ + { + slug: "oauth", + kind: "oauth2", + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + scopes: ["issues.read"], + }, + ], + }, + }); + yield* client.oauth.createClient({ + payload: { + owner: "org", + slug: clientSlug, + grant: "authorization_code", + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + clientId: "test-client", + clientSecret: "test-secret", + originIntegration: IntegrationSlug.make(slug), + }, + }); + const started = yield* client.oauth.start({ + payload: { + client: clientSlug, + clientOwner: "org", + owner: "org", + name: ConnectionName.make("main"), + integration: IntegrationSlug.make(slug), + template: AuthTemplateSlug.make("oauth"), + }, + }); + expect(started.status, "oauth.start redirects to the authorization server").toBe( + "redirect", + ); + if (started.status !== "redirect") return yield* Effect.die("no redirect"); + const code = yield* completeAuthorization(started.authorizationUrl); + yield* client.oauth.complete({ payload: { state: started.state, code } }); + + const address = (yield* client.tools.list({ query: {} })) + .filter((tool) => String(tool.integration) === slug) + .map((tool) => String(tool.address)) + .find((tool) => tool.endsWith("listIssues")); + expect(address, "the OAuth-protected tool is in the catalog").toBeDefined(); + if (!address) return yield* Effect.die("no listIssues tool"); + yield* oauth.clearRequests; + + const firstSession = mcp.session(identity); + const secondSession = mcp.session(identity); + const call = (session: typeof firstSession) => + Effect.gen(function* () { + let result = yield* session.call("execute", { code: invokeByAddressCode(address) }); + let approvals = 0; + while (result.text.includes("executionId:") && approvals < 10) { + result = yield* session.approvePaused(result.text); + approvals += 1; + } + expect(result.ok, `MCP execute completed: ${result.text.slice(0, 400)}`).toBe(true); + }); + + yield* Effect.all([call(firstSession), call(secondSession)], { + concurrency: "unbounded", + }); + + const refreshGrants = (yield* oauth.requests).filter( + (request) => + request.path === "/token" && request.body.includes("grant_type=refresh_token"), + ); + expect(refreshGrants, "both sessions joined one refresh grant").toHaveLength(1); + const bearers = upstream.bearers(); + expect(bearers, "both rejected calls retried after the refresh").toHaveLength(4); + expect(bearers[2], "the first retry used a new bearer").not.toBe(bearers[0]); + expect(bearers[3], "both retries used the refreshed bearer").toBe(bearers[2]); + }), + Effect.gen(function* () { + yield* client.connections + .remove({ + params: { + owner: "org", + integration: IntegrationSlug.make(slug), + name: ConnectionName.make("main"), + }, + }) + .pipe(Effect.ignore); + yield* client.oauth + .removeClient({ params: { slug: clientSlug }, payload: { owner: "org" } }) + .pipe(Effect.ignore); + yield* client.openapi.removeSpec({ params: { slug } }).pipe(Effect.ignore); + }), + ); + }), + ), +); diff --git a/e2e/selfhost/oauth-refresh-session-stress.test.ts b/e2e/selfhost/oauth-refresh-session-stress.test.ts new file mode 100644 index 000000000..7a7b105d4 --- /dev/null +++ b/e2e/selfhost/oauth-refresh-session-stress.test.ts @@ -0,0 +1,315 @@ +// Selfhost-only: the shared refresh gate must hold under real session +// concurrency, and must RELEASE once a grant settles. +// +// The cross-session scenario next door races two sessions. This one races eight +// through a barrier upstream that holds every first call until all have arrived, +// so the contention is forced rather than left to the scheduler — without a +// shared gate the refresh-grant count scales 1:1 with the session count, which +// is precisely what makes a rotating-token provider revoke the connection. +// +// The second wave then proves the gate is CLEARED after a grant settles rather +// than latched. A latched gate would replay a retired token; a gate that never +// released would deadlock every later refresh. Both failure modes are invisible +// to a single-wave test. +import { randomBytes } from "node:crypto"; +import { createServer, type ServerResponse } from "node:http"; + +import { expect } from "@effect/vitest"; +import { Effect } from "effect"; +import { composePluginApi } from "@executor-js/api/server"; +import { openApiHttpPlugin } from "@executor-js/plugin-openapi/api"; +import { + AuthTemplateSlug, + ConnectionName, + IntegrationSlug, + OAuthClientSlug, +} from "@executor-js/sdk/shared"; +import { serveOAuthTestServer } from "@executor-js/sdk/testing"; + +import { scenario } from "../src/scenario"; +import { Api, Mcp, Target } from "../src/services"; + +const api = composePluginApi([openApiHttpPlugin()] as const); + +const unique = (prefix: string) => `${prefix}_${randomBytes(4).toString("hex")}`; + +/** Concurrent MCP sessions racing one connection's rotating refresh token. */ +const SESSIONS = 8; +/** Upstream hits per session per wave: the rejected call plus its retry. */ +const HITS_PER_SESSION_PER_WAVE = 2; +/** A partial wave means a session never reached upstream — fail loudly, don't hang. */ +const WAVE_BARRIER_TIMEOUT_MS = 20_000; + +type UpstreamHandle = { + readonly url: string; + readonly bearers: () => readonly string[]; + readonly close: () => void; +}; + +/** + * Upstream that rejects a whole wave at once. + * + * The barrier is the point: it holds every session's first call until all + * `waveSize` have arrived, then 401s them together. That forces the sessions + * into a real simultaneous refresh rather than hoping the scheduler interleaves + * them. `POST /_rearm` opens the next wave, so a second round proves the + * in-flight gate is released after a grant settles rather than latched. + */ +const serveUpstream = (waveSize: number) => + Effect.acquireRelease( + Effect.callback((resume) => { + const bearers: string[] = []; + let held: ServerResponse[] = []; + let rejecting = true; + let barrier: ReturnType | null = null; + + const flush = () => { + if (barrier) { + clearTimeout(barrier); + barrier = null; + } + const batch = held; + held = []; + rejecting = false; + for (const response of batch) { + response.writeHead(401, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: "invalid_token" })); + } + }; + + const server = createServer((request, response) => { + const url = request.url ?? ""; + if (request.method === "POST" && url.startsWith("/_rearm")) { + rejecting = true; + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ rearmed: true })); + return; + } + if (request.method === "GET" && url.startsWith("/issues")) { + bearers.push((request.headers.authorization ?? "").replace(/^Bearer\s+/i, "")); + if (rejecting) { + held.push(response); + if (held.length === waveSize) flush(); + else if (!barrier) barrier = setTimeout(flush, WAVE_BARRIER_TIMEOUT_MS); + return; + } + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ issues: [] })); + return; + } + response.writeHead(404, { "content-type": "application/json" }); + response.end(JSON.stringify({ error: "not_found" })); + }); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + const port = typeof address === "object" && address ? address.port : 0; + resume( + Effect.succeed({ + url: `http://127.0.0.1:${port}`, + bearers: () => [...bearers], + close: () => { + if (barrier) clearTimeout(barrier); + server.close(); + server.closeAllConnections(); + }, + }), + ); + }); + }), + (server) => Effect.sync(server.close), + ); + +const spec = ( + baseUrl: string, + oauth: { readonly authorizationEndpoint: string; readonly tokenEndpoint: string }, +): string => + JSON.stringify({ + openapi: "3.0.3", + info: { title: "Issues API", version: "1.0.0" }, + servers: [{ url: baseUrl }], + paths: { + "/issues": { + get: { + operationId: "listIssues", + security: [{ oauth: ["issues.read"] }], + responses: { "200": { description: "issues" } }, + }, + }, + }, + components: { + securitySchemes: { + oauth: { + type: "oauth2", + flows: { + authorizationCode: { + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + scopes: { "issues.read": "Read issues" }, + }, + }, + }, + }, + }, + }); + +const invokeByAddressCode = (address: string) => ` +const segments = ${JSON.stringify(address)}.split(".").slice(1); +let node = tools; +for (const segment of segments) node = node[segment]; +const result = await node({}); +return JSON.stringify(result); +`; + +const completeAuthorization = (authorizationUrl: string) => + Effect.promise(async () => { + const authorize = await fetch(authorizationUrl, { redirect: "manual" }); + const loginUrl = authorize.headers.get("location"); + if (!loginUrl) throw new Error(`authorize did not redirect: ${authorize.status}`); + const login = await fetch(loginUrl, { + method: "POST", + headers: { authorization: `Basic ${Buffer.from("alice:password").toString("base64")}` }, + redirect: "manual", + }); + const callbackUrl = login.headers.get("location"); + if (!callbackUrl) throw new Error(`login did not redirect: ${login.status}`); + const code = new URL(callbackUrl).searchParams.get("code"); + if (!code) throw new Error("callback carried no authorization code"); + return code; + }); + +const distinct = (values: readonly string[]) => [...new Set(values)]; + +scenario( + `OAuth refresh · ${SESSIONS} concurrent MCP sessions survive two rotating-token waves`, + { timeout: 300_000 }, + Effect.scoped( + Effect.gen(function* () { + const target = yield* Target; + const { client: makeClient } = yield* Api; + const mcp = yield* Mcp; + const identity = yield* target.newIdentity(); + const client = yield* makeClient(api, identity); + const upstream = yield* serveUpstream(SESSIONS); + const oauth = yield* serveOAuthTestServer({ scopes: ["issues.read"] }); + const slug = unique("refreshstress"); + const clientSlug = OAuthClientSlug.make(unique("refreshstressc")); + + yield* Effect.ensuring( + Effect.gen(function* () { + yield* client.openapi.addSpec({ + payload: { + spec: { kind: "blob", value: spec(upstream.url, oauth) }, + slug, + baseUrl: upstream.url, + authenticationTemplate: [ + { + slug: "oauth", + kind: "oauth2", + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + scopes: ["issues.read"], + }, + ], + }, + }); + yield* client.oauth.createClient({ + payload: { + owner: "org", + slug: clientSlug, + grant: "authorization_code", + authorizationUrl: oauth.authorizationEndpoint, + tokenUrl: oauth.tokenEndpoint, + clientId: "test-client", + clientSecret: "test-secret", + originIntegration: IntegrationSlug.make(slug), + }, + }); + const started = yield* client.oauth.start({ + payload: { + client: clientSlug, + clientOwner: "org", + owner: "org", + name: ConnectionName.make("main"), + integration: IntegrationSlug.make(slug), + template: AuthTemplateSlug.make("oauth"), + }, + }); + expect(started.status, "oauth.start redirects to the authorization server").toBe( + "redirect", + ); + if (started.status !== "redirect") return yield* Effect.die("no redirect"); + const code = yield* completeAuthorization(started.authorizationUrl); + yield* client.oauth.complete({ payload: { state: started.state, code } }); + + const address = (yield* client.tools.list({ query: {} })) + .filter((tool) => String(tool.integration) === slug) + .map((tool) => String(tool.address)) + .find((tool) => tool.endsWith("listIssues")); + expect(address, "the OAuth-protected tool is in the catalog").toBeDefined(); + if (!address) return yield* Effect.die("no listIssues tool"); + yield* oauth.clearRequests; + + const sessions = Array.from({ length: SESSIONS }, () => mcp.session(identity)); + const call = (session: (typeof sessions)[number]) => + Effect.gen(function* () { + let result = yield* session.call("execute", { code: invokeByAddressCode(address) }); + let approvals = 0; + while (result.text.includes("executionId:") && approvals < 10) { + result = yield* session.approvePaused(result.text); + approvals += 1; + } + // Without a shared gate the loser redeems a retired refresh token + // and the authorization server answers invalid_grant, so this is + // the assertion that carries the user-visible failure. + expect(result.ok, `MCP execute completed: ${result.text.slice(0, 400)}`).toBe(true); + }); + + const wave = () => Effect.all(sessions.map(call), { concurrency: "unbounded" }); + + yield* wave(); + yield* Effect.promise(() => + fetch(`${upstream.url}/_rearm`, { method: "POST" }).then((response) => response.text()), + ); + yield* wave(); + + const refreshGrants = (yield* oauth.requests).filter( + (request) => + request.path === "/token" && request.body.includes("grant_type=refresh_token"), + ); + expect( + refreshGrants, + `${SESSIONS} sessions per wave joined one grant per wave`, + ).toHaveLength(2); + + const bearers = upstream.bearers(); + expect(bearers, "every session called and retried in both waves").toHaveLength( + SESSIONS * HITS_PER_SESSION_PER_WAVE * 2, + ); + const firstAttempts = bearers.slice(0, SESSIONS); + const firstRetries = bearers.slice(SESSIONS, SESSIONS * 2); + const secondRetries = bearers.slice(SESSIONS * 3); + expect(distinct(firstAttempts), "wave one started on one shared token").toHaveLength(1); + expect(distinct(firstRetries), "wave one retried on one shared token").toHaveLength(1); + expect(distinct(secondRetries), "wave two retried on one shared token").toHaveLength(1); + expect(firstRetries[0], "wave one minted a new token").not.toBe(firstAttempts[0]); + expect(secondRetries[0], "wave two minted another new token").not.toBe(firstRetries[0]); + }), + Effect.gen(function* () { + yield* client.connections + .remove({ + params: { + owner: "org", + integration: IntegrationSlug.make(slug), + name: ConnectionName.make("main"), + }, + }) + .pipe(Effect.ignore); + yield* client.oauth + .removeClient({ params: { slug: clientSlug }, payload: { owner: "org" } }) + .pipe(Effect.ignore); + yield* client.openapi.removeSpec({ params: { slug } }).pipe(Effect.ignore); + }), + ); + }), + ), +); diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index fb831260b..8683e7947 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -1,4 +1,4 @@ -import { Effect, Inspectable, Layer, Option, Predicate, Schema } from "effect"; +import { Effect, Fiber, Inspectable, Layer, Option, Predicate, Schema } from "effect"; import { FetchHttpClient, type HttpClient } from "effect/unstable/http"; import { fumadb } from "@executor-js/fumadb"; import { memoryAdapter } from "@executor-js/fumadb/adapters/memory"; @@ -174,6 +174,30 @@ const PLUGIN_STORAGE_DELETE_KEY_BATCH_SIZE = 90; const PLUGIN_STORAGE_CREATE_ROW_BATCH_SIZE = 90; const MAX_APPROVAL_ARGUMENT_PREVIEW_CHARS = 4_000; +type RefreshInFlight = Effect.Effect; + +// Scoped executors are rebuilt for each MCP session, while a self-host shares +// its root DB handle. Keep the rotating-token gate on that shared handle so +// separate sessions cannot redeem the same refresh token concurrently. +// +// SCOPE OF THE GUARANTEE: dedup reaches exactly as far as one root DB handle in +// one process. A host that hands every scoped executor a FRESH handle (a `db` +// factory, or a per-request handle like Cloud's) keys a different map each time +// and gets no dedup at all — silently, because an unshared gate still behaves +// correctly for the one caller holding it. Multi-replica deployments are outside +// it for the same reason: a process-local map cannot see a peer replica. Both +// need database-backed coordination (compare-and-swap on the stored refresh +// token) rather than a wider map. +const refreshInFlightByDb = new WeakMap>(); + +const refreshInFlightFor = (db: object): Map => { + const existing = refreshInFlightByDb.get(db); + if (existing) return existing; + const created = new Map(); + refreshInFlightByDb.set(db, created); + return created; +}; + // --------------------------------------------------------------------------- // Elicitation handler — resolved once at `createExecutor({ onElicitation })` // and overridable per `execute`. A tool that requests user input mid-execution @@ -1598,6 +1622,7 @@ export const createExecutor = { validateExecutorDbTables(tables, rootDbUntyped.internal.tables); @@ -1733,17 +1758,8 @@ export const createExecutor = - >(); - const connectionKey = (row: ConnectionRow): string => - `${row.owner}:${row.subject}:${row.integration}:${row.name}`; + `${tenant}:${row.owner}:${row.subject}:${row.integration}:${row.name}`; const loadOAuthClientRow = ( owner: Owner, @@ -2043,17 +2059,27 @@ export const createExecutor = refreshInFlight.delete(key))), + // The grant runs on its OWN fiber and callers only JOIN it. This entry is + // shared by every execution stack on this database, so the fiber that + // registers it is merely the first arrival, not an owner: if the grant + // inherited that caller's interruption (a disconnected MCP client, an + // execution deadline, a cancelled tool call) every peer awaiting the same + // entry would fail with an interrupt none of them caused and none can act + // on. Joining is per-caller, so a cancelled peer detaches without + // touching the grant or its siblings. A grant nobody is left waiting on + // still settles — token requests are bounded by `AbortSignal.timeout` — + // and still persists the rotated token, which is what keeps the next + // caller off a consumed one. + const running = Effect.runFork( + performTokenRefresh(row, provider, trigger).pipe( + Effect.ensuring(Effect.sync(() => refreshInFlight.delete(key))), + ), ); - // Re-check after building (a peer fiber may have registered first while - // we built ours) so everyone converges on the same shared grant. - const winner = refreshInFlight.get(key) ?? gated; - if (winner === gated) refreshInFlight.set(key, gated); - return yield* winner; + // No `yield*` between the lookup above and this registration, so + // check-and-set is atomic against peer fibers and cannot double-fire. + const shared = Fiber.join(running); + refreshInFlight.set(key, shared); + return yield* shared; }); // Resolve every named input of a connection (`variable → value`). A diff --git a/packages/core/sdk/src/oauth-flow.test.ts b/packages/core/sdk/src/oauth-flow.test.ts index 51de072d8..f80fdc4cc 100644 --- a/packages/core/sdk/src/oauth-flow.test.ts +++ b/packages/core/sdk/src/oauth-flow.test.ts @@ -1,5 +1,5 @@ import { describe, expect, it } from "@effect/vitest"; -import { Effect, Predicate } from "effect"; +import { Effect, Exit, Fiber, Predicate } from "effect"; import { AuthTemplateSlug, @@ -14,6 +14,7 @@ import { authToolFailure } from "./auth-tool-failure"; import { decodeOAuthCallbackState } from "./oauth"; import { OAuthStartError } from "./oauth-client"; import { missingGrantedOAuthScopes } from "./oauth-service"; +import { createExecutor } from "./executor"; import { definePlugin } from "./plugin"; import { makeTestWorkspaceHarness, memoryCredentialsPlugin } from "./test-config"; import { ToolResult } from "./tool-result"; @@ -821,6 +822,163 @@ describe("oauth token refresh in resolveConnectionValue", () => { ), ); + it.effect("shares one refresh grant across executor stacks for the same connection", () => + Effect.scoped( + Effect.gen(function* () { + const server = yield* serveOAuthTestServer({ scopes: ["read"] }); + const harness = yield* makeTestWorkspaceHarness({ plugins }); + const { executor, config } = harness; + yield* executor.acme.seed(); + + yield* executor.oauth.createClient({ + owner: "org", + slug: CLIENT, + authorizationUrl: server.authorizationEndpoint, + tokenUrl: server.tokenEndpoint, + grant: "authorization_code", + clientId: "test-client", + clientSecret: "test-secret", + resource: server.mcpResourceUrl, + }); + + const started = yield* executor.oauth.start({ + owner: "org", + client: CLIENT, + clientOwner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + }); + expect(started.status).toBe("redirect"); + if (started.status !== "redirect") return; + const callback = yield* server.completeAuthorizationCodeFlow({ + authorizationUrl: started.authorizationUrl, + }); + yield* executor.oauth.complete({ state: started.state, code: callback.code }); + + const original = (yield* executor.execute( + ToolAddress.make("tools.acme.org.main.whoami"), + {}, + )) as { token: string }; + yield* Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b("name", "=", "main"), + set: { expires_at: Date.now() - 60_000 }, + }), + ); + yield* server.clearRequests; + + const peer = yield* createExecutor(config); + yield* Effect.addFinalizer(() => peer.close().pipe(Effect.ignore)); + const [first, second] = yield* Effect.all( + [ + executor.execute(ToolAddress.make("tools.acme.org.main.whoami"), {}), + peer.execute(ToolAddress.make("tools.acme.org.main.whoami"), {}), + ], + { concurrency: "unbounded" }, + ); + + const firstToken = (first as { token: string }).token; + const secondToken = (second as { token: string }).token; + expect(firstToken).not.toBe(original.token); + expect(secondToken).toBe(firstToken); + const refreshGrants = (yield* server.requests).filter( + (request) => + request.path === "/token" && request.body.includes("grant_type=refresh_token"), + ); + expect(refreshGrants).toHaveLength(1); + }), + ), + ); + + it.effect("a joined stack survives the owning stack being interrupted mid-refresh", () => + Effect.scoped( + Effect.gen(function* () { + const server = yield* serveOAuthTestServer({ scopes: ["read"] }); + const harness = yield* makeTestWorkspaceHarness({ plugins }); + const { executor, config } = harness; + yield* executor.acme.seed(); + + yield* executor.oauth.createClient({ + owner: "org", + slug: CLIENT, + authorizationUrl: server.authorizationEndpoint, + tokenUrl: server.tokenEndpoint, + grant: "authorization_code", + clientId: "test-client", + clientSecret: "test-secret", + resource: server.mcpResourceUrl, + }); + const started = yield* executor.oauth.start({ + owner: "org", + client: CLIENT, + clientOwner: "org", + name: ConnectionName.make("main"), + integration: INTEG, + template: TEMPLATE, + }); + expect(started.status).toBe("redirect"); + if (started.status !== "redirect") return; + const callback = yield* server.completeAuthorizationCodeFlow({ + authorizationUrl: started.authorizationUrl, + }); + yield* executor.oauth.complete({ state: started.state, code: callback.code }); + + const address = ToolAddress.make("tools.acme.org.main.whoami"); + yield* executor.execute(address, {}); + yield* Effect.promise(() => + config.db.updateMany("connection", { + where: (b) => b("name", "=", "main"), + set: { expires_at: Date.now() - 60_000 }, + }), + ); + + // Park the owning stack inside the token request so the second stack + // has a window to join the in-flight grant before the interrupt lands. + let release: (() => void) | null = null; + const parked = new Promise((resolve) => { + release = resolve; + }); + let sawTokenRequest = false; + // oxlint-disable-next-line executor/no-raw-fetch -- test seam: this wrapper replaces the platform fetch and must delegate back to it once unparked + const passthrough: typeof globalThis.fetch = globalThis.fetch; + const parkingFetch: typeof globalThis.fetch = async (input, init) => { + if (String(input).includes("/token")) { + sawTokenRequest = true; + await parked; + } + return passthrough(input, init); + }; + + const owner = yield* createExecutor({ ...config, fetch: parkingFetch }); + const joiner = yield* createExecutor(config); + yield* Effect.addFinalizer(() => owner.close().pipe(Effect.ignore)); + yield* Effect.addFinalizer(() => joiner.close().pipe(Effect.ignore)); + + const outcome = yield* Effect.promise(async () => { + const ownerFiber = Effect.runFork(owner.execute(address, {})); + for (let attempt = 0; attempt < 300 && !sawTokenRequest; attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 10)); + } + const joined = Effect.runPromise(Effect.exit(joiner.execute(address, {}))); + await new Promise((resolve) => setTimeout(resolve, 200)); + // The owning stack's MCP client disconnects mid-refresh. Its + // interruption only lands once the shared grant settles, so unpark the + // token request rather than awaiting the interrupt first. + const interrupted = Effect.runPromise(Fiber.interrupt(ownerFiber)); + release?.(); + await interrupted; + return await joined; + }); + + expect(sawTokenRequest).toBe(true); + expect(Exit.isSuccess(outcome), "the joined stack completed on the shared grant").toBe( + true, + ); + }), + ), + ); + it.effect( "refreshes a Personal (user) connection minted through a Workspace (org) app — own→shared client resolution", () =>