fix(core): serialize MCP OAuth refreshes across processes

packages/core/src/mcp/index.ts

@@ -8,6 +8,7 @@ import { createHash } from "node:crypto"
 import { isDeepStrictEqual } from "node:util"
 import { Cause, Context, Effect, Exit, FiberSet, Latch, Layer, Schema, Scope, Semaphore, Stream, Types } from "effect"
 import { makeLocationNode } from "@opencode/util/effect/app-node"
+import { EffectFlock } from "@opencode/util/effect-flock"
 import { Credential } from "../credential.js"
 import { Bus } from "../bus.js"
 import { Environment } from "../environment/index.js"
@@ -149,6 +150,7 @@ export const layer = (options?: Options) =>
       const forms = yield* Form.Service
       const integration = yield* Integration.Service
       const credentials = yield* Credential.Service
+      const flock = yield* EffectFlock.Service
       const root = yield* Effect.scope
       const fork = yield* FiberSet.makeRuntime<never, void, never>()
 
@@ -214,6 +216,7 @@ export const layer = (options?: Options) =>
         const { McpOAuth } = yield* Effect.promise(() => import("./oauth.js"))
         return yield* McpOAuth.connectProvider({ config: entry.config, integrationID: entry.integrationID }).pipe(
           Effect.provideService(Credential.Service, credentials),
+          Effect.provideService(EffectFlock.Service, flock),
         )
       })
 
@@ -759,7 +762,7 @@ export function configured(options?: Options) {
   return makeLocationNode({
     service: Service,
     layer: layer(options),
-    deps: [Location.node, Environment.node, Bus.node, Form.node, Integration.node, Credential.node],
+    deps: [Location.node, Environment.node, Bus.node, Form.node, Integration.node, Credential.node, EffectFlock.node],
   })
 }
 

packages/core/src/mcp/oauth.ts

@@ -18,6 +18,7 @@ import {
 import { OAuthMetadataSchema, OpenIdProviderDiscoveryMetadataSchema } from "@modelcontextprotocol/core"
 import { Cause, Deferred, Effect } from "effect"
 import { ConfigMCP } from "@opencode/schema/config/mcp"
+import { EffectFlock } from "@opencode/util/effect-flock"
 import { Credential } from "../credential.js"
 import { OauthCallbackPage } from "../oauth/page.js"
 import type { Integration } from "../integration.js"
@@ -141,6 +142,7 @@ export interface Options {
   readonly clientMetadataUrl?: string
   readonly discovery?: OAuthDiscoveryState
   readonly invalidate?: OAuthClientProvider["invalidateCredentials"]
+  readonly refreshLock?: OAuthClientProvider["withRefreshLock"]
 }
 
 export const provider = (options: Options): OAuthClientProvider => {
@@ -200,6 +202,7 @@ export const provider = (options: Options): OAuthClientProvider => {
       return redirect.open(url)
     },
     ...(options.invalidate ? { invalidateCredentials: options.invalidate } : {}),
+    ...(options.refreshLock ? { withRefreshLock: options.refreshLock } : {}),
     saveCodeVerifier: (verifier) => options.store.saveCodeVerifier(verifier),
     codeVerifier: async () => {
       const verifier = await options.store.codeVerifier()
@@ -271,6 +274,7 @@ export const connectProvider = Effect.fnUntraced(function* (input: {
   readonly integrationID: Integration.ID
 }) {
   const credentials = yield* Credential.Service
+  const flock = yield* EffectFlock.Service
   const run = Effect.runPromiseWith(yield* Effect.context())
   const found = (yield* credentials.list(input.integrationID)).at(-1)
   if (!found || found.value.type !== "oauth") return provider({ config: input.config, store: memoryStore() })
@@ -291,6 +295,10 @@ export const connectProvider = Effect.fnUntraced(function* (input: {
       await run(Effect.logWarning("mcp oauth credential invalidated", { credentialID: id, scope }))
       await run(credentials.remove(id))
     },
+    // Other processes share the row too. A refresh that waited here re-reads the token the previous
+    // holder saved instead of replaying the one it replaced, which rotating servers reject or revoke.
+    refreshLock: (refresh) =>
+      run(flock.withLock(Effect.tryPromise({ try: refresh, catch: (error) => error }), `mcp-oauth-refresh:${id}`)),
     store: {
       tokens: async () => {
         const oauth = await read()

packages/core/test/mcp-oauth.test.ts

@@ -5,11 +5,24 @@ import { Credential } from "@opencode/core/credential"
 import { Integration } from "@opencode/core/integration"
 import { McpClient } from "@opencode/core/mcp/client"
 import { McpOAuth } from "@opencode/core/mcp/oauth"
+import { EffectFlock } from "@opencode/util/effect-flock"
+import { LayerNode } from "@opencode/util/effect/layer-node"
+import { Global } from "@opencode/util/global"
 import { Cause, Effect, Exit } from "effect"
 import { hostEnvironmentLayer } from "./fixture/environment"
+import { tmpdir } from "./fixture/tmpdir"
 
 const authServer = Bun.serve({ port: 0, fetch: () => new Response(null, { status: 404 }) })
-afterAll(() => authServer.stop(true))
+const state = await tmpdir()
+afterAll(async () => {
+  authServer.stop(true)
+  await state[Symbol.asyncDispose]()
+})
+
+// Every provider locks in one directory, as separate processes on the same machine do.
+const flock = LayerNode.compile(EffectFlock.node, {
+  replacements: [Global.node.replace(Global.layerWith({ state: state.path }))],
+})
 
 const integrationID = Integration.ID.make("mcp_test")
 const methodID = Integration.MethodID.make("oauth")
@@ -54,7 +67,10 @@ const memoryCredentials = (initial: Credential.Info[]) => {
 
 const connectProvider = (config: typeof ConfigMCP.Remote.Type, store: ReturnType<typeof memoryCredentials>) =>
   Effect.runPromise(
-    McpOAuth.connectProvider({ config, integrationID }).pipe(Effect.provideService(Credential.Service, store.service)),
+    McpOAuth.connectProvider({ config, integrationID }).pipe(
+      Effect.provideService(Credential.Service, store.service),
+      Effect.provide(flock),
+    ),
   )
 
 // Serves authorization server metadata with the given capabilities and records DCR + token requests.
@@ -275,6 +291,35 @@ describe("MCP OAuth", () => {
     ])
   })
 
+  test("serializes refreshes of a credential shared across processes", async () => {
+    const presented: string[] = []
+    const server = Bun.serve({
+      port: 0,
+      async fetch(request) {
+        const url = new URL(request.url)
+        if (request.method !== "POST" || url.pathname !== "/token") return new Response(null, { status: 404 })
+        const token = new URLSearchParams(await request.text()).get("refresh_token") ?? ""
+        presented.push(token)
+        // Hold the response so a refresher that did not wait would present the same token.
+        await Bun.sleep(50)
+        return Response.json({ access_token: "next", token_type: "Bearer", refresh_token: `${token}+` })
+      },
+    })
+    const url = server.url.href
+    const store = memoryCredentials([credential({ access: "expired", refresh: "r", url })])
+    const providers = await Promise.all([connectProvider(remote(url), store), connectProvider(remote(url), store)])
+
+    // Plain fetch: processes share no in-memory request, so only the lock can order the refreshes.
+    const results = await Promise.all(
+      providers.map((provider) => auth(provider, { serverUrl: url, fetchFn: (input, init) => fetch(input, init) })),
+    ).finally(() => server.stop(true))
+
+    expect(results).toEqual(["AUTHORIZED", "AUTHORIZED"])
+    expect(presented).toEqual(["r", "r+"])
+    const stored = store.rows.get(Credential.ID.make("cred_test"))?.value
+    expect(stored?.type === "oauth" && stored.refresh).toBe("r++")
+  })
+
   test("generates a loopback redirect URL when none is configured", async () => {
     expect(await authorize()).toMatch(/^http:\/\/127\.0\.0\.1:\d+\/callback$/)
   })

packages/core/test/mcp.test.ts

@@ -17,6 +17,7 @@ import { ConfigMcpPlugin } from "@opencode/core/config/plugin/mcp"
 import { Credential } from "@opencode/core/credential"
 import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder"
 import { LayerNode } from "@opencode/util/effect/layer-node"
+import { EffectFlock } from "@opencode/util/effect-flock"
 import { Bus } from "@opencode/core/bus"
 import { ID, type Payload } from "@opencode/schema/event"
 import { Form } from "@opencode/core/form"
@@ -347,6 +348,7 @@ function resourceMcpLayer(
           },
         }),
         Layer.mock(Credential.Service, {}),
+        Layer.mock(EffectFlock.Service, {}),
         overrides?.environment ?? hostEnvironmentLayer,
       ),
     ),
@@ -1988,7 +1990,15 @@ testEffect(Layer.empty).live("keeps MCP config snapshots stable during an in-fli
 
 const shutdownIt = testEffect(
   AppNodeBuilder.build(
-    LayerNode.group([Bus.node, Integration.node, Credential.node, Form.node, Environment.node, Location.node]),
+    LayerNode.group([
+      Bus.node,
+      Integration.node,
+      Credential.node,
+      EffectFlock.node,
+      Form.node,
+      Environment.node,
+      Location.node,
+    ]),
     [
       Location.node.replace(
         Layer.succeed(

patches/@modelcontextprotocol%2Fclient@2.0.0.patch

@@ -1,8 +1,32 @@
 diff --git a/dist/index.cjs b/dist/index.cjs
-index 635f1c0134274c89e7efbcae2adf3d9ecba575ee..5eaa7b451c8c7e12e51f61c2a835ff6f7a8ea0ae 100644
+index 635f1c0134274c89e7efbcae2adf3d9ecba575ee..a2a7fc2a1b567ee3633392d6c596734d1d00fb90 100644
 --- a/dist/index.cjs
 +++ b/dist/index.cjs
-@@ -977,7 +977,6 @@ async function discoverMetadataWithFallback(serverUrl, wellKnownType, fetchFn, o
+@@ -754,6 +754,7 @@ async function authInternal(provider, { serverUrl, authorizationCode, iss, scope
+ 		}, infoCtx);
+ 		return "AUTHORIZED";
+ 	}
++	const refreshed = await (provider.withRefreshLock ?? ((refresh) => refresh()))(async () => {
+ 	let tokens = discardIfIssuerMismatch(await provider.tokens(infoCtx), issuer);
+ 	if (tokens && tokens.issuer === void 0) {
+ 		tokens = {
+@@ -775,11 +776,14 @@ async function authInternal(provider, { serverUrl, authorizationCode, iss, scope
+ 			...newTokens,
+ 			issuer
+ 		}, infoCtx);
+-		return "AUTHORIZED";
++		return true;
+ 	} catch (error) {
+ 		if (error instanceof InsecureTokenEndpointError) throw error;
+ 		if (!(error instanceof require_src.OAuthError) || error.code === require_src.OAuthErrorCode.ServerError) {} else throw error;
+ 	}
++	return false;
++	});
++	if (refreshed) return "AUTHORIZED";
+ 	const state = provider.state ? await provider.state() : void 0;
+ 	const { authorizationUrl, codeVerifier } = await startAuthorization(authorizationServerUrl, {
+ 		metadata,
+@@ -977,7 +981,6 @@ async function discoverMetadataWithFallback(serverUrl, wellKnownType, fetchFn, o
  	else {
  		const wellKnownPath = buildWellKnownPath(wellKnownType, issuer.pathname);
  		url = new URL(wellKnownPath, opts?.metadataServerUrl ?? issuer);
@@ -10,7 +34,7 @@ index 635f1c0134274c89e7efbcae2adf3d9ecba575ee..5eaa7b451c8c7e12e51f61c2a835ff6f
  	}
  	let response = await tryMetadataDiscovery(url, protocolVersion, fetchFn);
  	if (!opts?.metadataUrl && shouldAttemptFallback(response, issuer.pathname)) response = await tryMetadataDiscovery(new URL(`/.well-known/${wellKnownType}`, issuer), protocolVersion, fetchFn);
-@@ -1158,7 +1157,6 @@ async function startAuthorization(authorizationServerUrl, { metadata, clientInfo
+@@ -1158,7 +1161,6 @@ async function startAuthorization(authorizationServerUrl, { metadata, clientInfo
  	authorizationUrl.searchParams.set("redirect_uri", String(redirectUrl));
  	if (state) authorizationUrl.searchParams.set("state", state);
  	if (scope) authorizationUrl.searchParams.set("scope", scope);
@@ -18,11 +42,69 @@ index 635f1c0134274c89e7efbcae2adf3d9ecba575ee..5eaa7b451c8c7e12e51f61c2a835ff6f
  	if (resource) authorizationUrl.searchParams.set("resource", resource.href);
  	return {
  		authorizationUrl,
+diff --git a/dist/index.d.cts b/dist/index.d.cts
+index 2d74180d5402a97f7460305726d8dc7da6105dcf..97a30ccce75a8135827654375f90d9034f9fb434 100644
+--- a/dist/index.d.cts
++++ b/dist/index.d.cts
+@@ -335,6 +335,12 @@ interface OAuthClientProvider {
+    * This avoids requiring the user to intervene manually.
+    */
+   invalidateCredentials?(scope: 'all' | 'client' | 'tokens' | 'verifier' | 'discovery'): void | Promise<void>;
++  /**
++   * If implemented, wraps reading the stored tokens, refreshing them and saving the result.
++   * Providers whose credentials are shared across processes can hold a lock here, so a
++   * waiting refresher reads the rotated token instead of replaying the one it replaced.
++   */
++  withRefreshLock?<T>(refresh: () => Promise<T>): Promise<T>;
+   /**
+    * Prepares grant-specific parameters for a token request.
+    *
+diff --git a/dist/index.d.mts b/dist/index.d.mts
+index d54447a08f0aa9876dacfe37f50ec1a48a77c132..e26e4dadb4867b80904ae7011cf0d121645a61ae 100644
+--- a/dist/index.d.mts
++++ b/dist/index.d.mts
+@@ -335,6 +335,12 @@ interface OAuthClientProvider {
+    * This avoids requiring the user to intervene manually.
+    */
+   invalidateCredentials?(scope: 'all' | 'client' | 'tokens' | 'verifier' | 'discovery'): void | Promise<void>;
++  /**
++   * If implemented, wraps reading the stored tokens, refreshing them and saving the result.
++   * Providers whose credentials are shared across processes can hold a lock here, so a
++   * waiting refresher reads the rotated token instead of replaying the one it replaced.
++   */
++  withRefreshLock?<T>(refresh: () => Promise<T>): Promise<T>;
+   /**
+    * Prepares grant-specific parameters for a token request.
+    *
 diff --git a/dist/index.mjs b/dist/index.mjs
-index f02ce3ca394e826fc27c9848dbe9a214bf6ba7a9..73a93a066c3540e131fa0770a74b4a7e8d9343a0 100644
+index f02ce3ca394e826fc27c9848dbe9a214bf6ba7a9..5136555235c5b1a30cff4708d396dccfbf8cee00 100644
 --- a/dist/index.mjs
 +++ b/dist/index.mjs
-@@ -974,7 +974,6 @@ async function discoverMetadataWithFallback(serverUrl, wellKnownType, fetchFn, o
+@@ -751,6 +751,7 @@ async function authInternal(provider, { serverUrl, authorizationCode, iss, scope
+ 		}, infoCtx);
+ 		return "AUTHORIZED";
+ 	}
++	const refreshed = await (provider.withRefreshLock ?? ((refresh) => refresh()))(async () => {
+ 	let tokens = discardIfIssuerMismatch(await provider.tokens(infoCtx), issuer);
+ 	if (tokens && tokens.issuer === void 0) {
+ 		tokens = {
+@@ -772,11 +773,14 @@ async function authInternal(provider, { serverUrl, authorizationCode, iss, scope
+ 			...newTokens,
+ 			issuer
+ 		}, infoCtx);
+-		return "AUTHORIZED";
++		return true;
+ 	} catch (error) {
+ 		if (error instanceof InsecureTokenEndpointError) throw error;
+ 		if (!(error instanceof OAuthError) || error.code === OAuthErrorCode.ServerError) {} else throw error;
+ 	}
++	return false;
++	});
++	if (refreshed) return "AUTHORIZED";
+ 	const state = provider.state ? await provider.state() : void 0;
+ 	const { authorizationUrl, codeVerifier } = await startAuthorization(authorizationServerUrl, {
+ 		metadata,
+@@ -974,7 +978,6 @@ async function discoverMetadataWithFallback(serverUrl, wellKnownType, fetchFn, o
  	else {
  		const wellKnownPath = buildWellKnownPath(wellKnownType, issuer.pathname);
  		url = new URL(wellKnownPath, opts?.metadataServerUrl ?? issuer);
@@ -30,7 +112,7 @@ index f02ce3ca394e826fc27c9848dbe9a214bf6ba7a9..73a93a066c3540e131fa0770a74b4a7e
  	}
  	let response = await tryMetadataDiscovery(url, protocolVersion, fetchFn);
  	if (!opts?.metadataUrl && shouldAttemptFallback(response, issuer.pathname)) response = await tryMetadataDiscovery(new URL(`/.well-known/${wellKnownType}`, issuer), protocolVersion, fetchFn);
-@@ -1155,7 +1154,6 @@ async function startAuthorization(authorizationServerUrl, { metadata, clientInfo
+@@ -1155,7 +1158,6 @@ async function startAuthorization(authorizationServerUrl, { metadata, clientInfo
  	authorizationUrl.searchParams.set("redirect_uri", String(redirectUrl));
  	if (state) authorizationUrl.searchParams.set("state", state);
  	if (scope) authorizationUrl.searchParams.set("scope", scope);