// 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); }), ); }), ),);
// 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);
}),
);
}),
),
);