Share the OAuth refresh gate across execution stacks

.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.

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<UpstreamHandle>((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);
+        }),
+      );
+    }),
+  ),
+);