Share the OAuth refresh gate across execution stacks

.changeset/oauth-refresh-cross-session.md

@@ -0,0 +1,13 @@
+---
+"@executor-js/sdk": patch
+---
+
+Share the OAuth refresh gate across execution stacks so a rotating refresh token is redeemed once.
+
+The in-flight refresh gate was built inside `createExecutor`, so it only covered one execution stack. A host builds a fresh stack per MCP session, and now per request, so two sessions resolving the same connection each read the same stored refresh token and each believed they were the refresh winner. Against a provider that rotates refresh tokens, the loser redeems a token the winner already spent, and a provider that detects reuse revokes the whole token family: the connection dies and the user has to reauthorize. The first refresh always succeeds, so the fault stayed invisible until a later expiry.
+
+The gate now hangs off the root database handle, which is the object hosts already share across sessions and requests, so every stack over one handle converges on one gate. Its key includes the tenant, because a gate that spans tenants would otherwise let two tenants collide on one entry.
+
+The grant also runs on its own detached fiber that callers await, rather than on whichever caller registered it. Sharing an entry across stacks would otherwise share the first caller's cancellation: a disconnected MCP client or an execution deadline would fail every peer waiting on that entry, and would abandon a refresh token the authorization server had already rotated, which is itself a dead connection. A cancelled peer now detaches without touching the grant, and a grant nobody is left waiting on still settles and still persists the rotated token.
+
+Deduplication covers one database handle in one process. A host that builds a fresh handle per request or per session, and any multi-instance or multi-replica deployment, is out of scope here and still needs database-backed coordination, such as a compare-and-swap on the stored refresh token.

e2e/selfhost/oauth-refresh-cross-session.test.ts

@@ -0,0 +1,275 @@
+// 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 execution
+// stack, but the self-host builds a fresh stack 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 authorization server; the upstream then rejects both
+// sessions' first call with a 401 at the same instant (it holds both requests
+// until both have arrived, so the contention is forced rather than left to the
+// scheduler); and the authorization server'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")}`;
+
+/** Both sessions call once, are rejected together, then retry once. */
+const SESSIONS = 2;
+
+type UpstreamHandle = {
+  readonly url: string;
+  readonly bearers: () => readonly string[];
+  readonly close: () => void;
+};
+
+/**
+ * Upstream that rejects the whole first wave at once.
+ *
+ * The barrier is the point: it holds every session's first call until all have
+ * arrived, then 401s them together, which forces both sessions into a genuinely
+ * simultaneous refresh instead of hoping the scheduler interleaves them.
+ */
+const serveUpstream = () =>
+  Effect.acquireRelease(
+    Effect.callback<UpstreamHandle>((resume) => {
+      const bearers: string[] = [];
+      const held: ServerResponse[] = [];
+      const server = createServer((request, response) => {
+        if (request.method === "GET" && (request.url ?? "").startsWith("/issues")) {
+          bearers.push((request.headers.authorization ?? "").replace(/^Bearer\s+/i, ""));
+          if (held.length < SESSIONS) {
+            held.push(response);
+            if (held.length === SESSIONS) {
+              for (const rejected of held) {
+                rejected.writeHead(401, { "content-type": "application/json" });
+                rejected.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) return null;
+    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) return null;
+    return new URL(callbackUrl).searchParams.get("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);
+          expect(code, "the authorization server issued a callback code").toBeDefined();
+          if (code == null) return yield* Effect.die("no authorization code");
+          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);
+            });
+
+          yield* Effect.all(sessions.map(call), { 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(
+            SESSIONS * 2,
+          );
+          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);
+        }),
+      );
+    }),
+  ),
+);