fix(oauth): complete browser callbacks before tool sync

e2e/selfhost/mcp-oauth-callback-background-sync.test.ts

@@ -0,0 +1,170 @@
+// An OAuth callback commits the fresh grant before it synchronizes a remote
+// MCP catalog. A slow tools/list response must not keep the popup request open;
+// the host keeps catalog work alive and the tools converge afterward.
+import { randomBytes } from "node:crypto";
+
+import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
+import { ListToolsRequestSchema } from "@modelcontextprotocol/sdk/types.js";
+
+import { expect } from "@effect/vitest";
+import { Effect, Schedule } from "effect";
+import { composePluginApi } from "@executor-js/api/server";
+import { deriveMcpNamespace } from "@executor-js/plugin-mcp";
+import { mcpHttpPlugin } from "@executor-js/plugin-mcp/api";
+import { serveMcpServerWithOAuth } from "@executor-js/plugin-mcp/testing";
+import { IntegrationSlug } from "@executor-js/sdk/shared";
+import { OAuthTestServer } from "@executor-js/sdk/testing";
+
+import { scenario } from "../src/scenario";
+import { Api, Browser, Target } from "../src/services";
+import { visit } from "../src/surfaces/browser";
+
+const api = composePluginApi([mcpHttpPlugin()] as const);
+
+const submitProviderLogin = async (loginUrl: string): Promise<string> => {
+  const response = await fetch(loginUrl, {
+    method: "POST",
+    redirect: "manual",
+    headers: { authorization: `Basic ${Buffer.from("alice:password").toString("base64")}` },
+  });
+  const location = response.headers.get("location");
+  if (response.status !== 302 || !location) {
+    throw new Error(`provider login did not redirect (${response.status})`);
+  }
+  return new URL(location, loginUrl).toString();
+};
+
+for (const failsFirst of [false, true]) {
+  scenario(
+    `MCP OAuth · callback closes before ${failsFirst ? "failing" : "blocked"} catalog discovery and preserves the grant`,
+    { timeout: 240_000 },
+    Effect.scoped(
+      Effect.gen(function* () {
+        const target = yield* Target;
+        const browser = yield* Browser;
+        const { client: makeApiClient } = yield* Api;
+        const gate = Promise.withResolvers<void>();
+        const listing = Promise.withResolvers<void>();
+        let failListing = failsFirst;
+        const server = yield* serveMcpServerWithOAuth(
+          () => {
+            const mcp = new McpServer(
+              { name: "callback-mcp", version: "1.0.0" },
+              { capabilities: { tools: {} } },
+            );
+            mcp.server.setRequestHandler(ListToolsRequestSchema, async () => {
+              listing.resolve();
+              if (failListing) throw new Error("Temporary catalog outage");
+              return { tools: [{ name: "simple_echo", inputSchema: { type: "object" as const } }] };
+            });
+            return mcp;
+          },
+          { path: "/mcp", beforeAuthenticatedRequest: () => gate.promise },
+        );
+        const identity = yield* target.newIdentity();
+        const client = yield* makeApiClient(api, identity);
+        const displayName = `Slow callback MCP ${randomBytes(3).toString("hex")}`;
+        const slug = IntegrationSlug.make(deriveMcpNamespace({ name: displayName }));
+        const clientsBefore = new Set((yield* client.oauth.listClients()).map((item) => item.slug));
+
+        yield* Effect.gen(function* () {
+          yield* browser.session(identity, async ({ page, step }) => {
+            await step("Add an OAuth-protected MCP integration", async () => {
+              const addUrl = new URL("/integrations/add/mcp", target.baseUrl);
+              addUrl.searchParams.set("url", server.endpoint);
+              await visit(page, addUrl.toString());
+              await page
+                .getByText("How does this server authenticate?")
+                .waitFor({ timeout: 30_000 });
+              await page.getByPlaceholder("e.g. Linear").fill(displayName);
+              await page.getByRole("button", { name: "Add integration" }).click();
+              await page.waitForURL(/\/integrations\/(?!add\b)[^/?]+$/, { timeout: 30_000 });
+            });
+
+            await step("Authorize while the MCP catalog is deliberately slow", async () => {
+              await page.getByRole("button", { name: "Add connection" }).first().click();
+              await page.getByRole("heading", { name: /Add connection/ }).waitFor();
+
+              const popupPromise = page.waitForEvent("popup", { timeout: 30_000 });
+              await page.getByRole("button", { name: "Connect", exact: true }).click();
+              const popup = await popupPromise;
+              await popup.waitForURL(/\/login\?/, { timeout: 30_000 });
+              const callbackUrl = await submitProviderLogin(popup.url());
+
+              // No authenticated MCP request can complete before we release
+              // the gate. Callback success therefore proves ordering without
+              // racing a timer against a cold browser or a loaded host.
+              await popup.goto(callbackUrl, { waitUntil: "domcontentloaded", timeout: 30_000 });
+              const committed = await Effect.runPromise(
+                client.connections.list({ query: { integration: slug } }),
+              );
+              expect(
+                committed.length,
+                "the callback persisted the connection before discovery",
+              ).toBe(1);
+              // Release before the opener's separate health probe, which also
+              // uses this transport, after proving the callback has returned.
+              gate.resolve();
+              await page
+                .getByText("Connection added", { exact: true })
+                .waitFor({ timeout: 30_000 });
+            });
+          });
+
+          yield* Effect.promise(() => listing.promise);
+          const afterListing = yield* client.connections.list({ query: { integration: slug } });
+          expect(
+            afterListing.length,
+            "a failed remote listing cannot remove the durable grant",
+          ).toBe(1);
+          const connection = afterListing[0];
+          if (connection === undefined) return yield* Effect.die("Missing committed connection");
+          expect(
+            connection.lastHealth?.status,
+            "discovery reports the actual upstream health",
+          ).toBe(failsFirst ? "degraded" : "healthy");
+          failListing = false;
+          if (failsFirst) {
+            // Failed discovery intentionally backs off until the catalog TTL.
+            // An explicit refresh is the supported immediate recovery action.
+            const params = { owner: connection.owner, integration: slug, name: connection.name };
+            yield* client.connections.refresh({ params });
+          }
+
+          const tools = yield* client.tools.list({ query: { integration: slug } }).pipe(
+            Effect.filterOrFail(
+              (items) => items.some((tool) => String(tool.name) === "simple_echo"),
+              () => "slow_mcp_catalog_pending" as const,
+            ),
+            Effect.retry(Schedule.both(Schedule.spaced("1 second"), Schedule.recurs(20))),
+          );
+          expect(
+            tools.map((tool) => String(tool.name)),
+            "the host-kept background sync eventually publishes the remote tool",
+          ).toContain("simple_echo");
+          const healthy = yield* client.connections.checkHealth({
+            params: { owner: connection.owner, integration: slug, name: connection.name },
+            query: {},
+          });
+          expect(healthy.status, "the recovered server accepts the existing grant").toBe("healthy");
+        }).pipe(
+          Effect.ensuring(
+            Effect.gen(function* () {
+              gate.resolve();
+              const clientsAfter = yield* client.oauth.listClients();
+              for (const oauthClient of clientsAfter) {
+                if (!clientsBefore.has(oauthClient.slug)) {
+                  yield* client.oauth.removeClient({
+                    params: { slug: oauthClient.slug },
+                    payload: { owner: oauthClient.owner },
+                  });
+                }
+              }
+              yield* client.mcp.removeServer({ params: { slug } });
+            }).pipe(Effect.ignore),
+          ),
+        );
+      }),
+    ).pipe(Effect.provide(OAuthTestServer.layer())),
+  );
+}

packages/core/api/src/handlers/oauth.ts

@@ -212,13 +212,16 @@ export const OAuthHandlers = HttpApiBuilder.group(ExecutorApi, "oauth", (handler
           const html = yield* runOAuthCallback({
             complete: ({ state, code, callbackDomain }) =>
               executor.oauth
-                .complete({
-                  // `runOAuthCallback`'s `state` is a raw string from the URL;
-                  // the SDK speaks the branded `OAuthState` (nominal brand).
-                  state: OAuthState.make(state),
-                  code: code ?? "",
-                  callbackDomain,
-                })
+                .complete(
+                  {
+                    // `runOAuthCallback`'s `state` is a raw string from the URL;
+                    // the SDK speaks the branded `OAuthState` (nominal brand).
+                    state: OAuthState.make(state),
+                    code: code ?? "",
+                    callbackDomain,
+                  },
+                  { toolSync: "background" },
+                )
                 .pipe(
                   Effect.tapError((cause: unknown) =>
                     Effect.logError("OAuth callback completion failed", cause),

packages/core/sdk/src/executor.ts

@@ -4453,6 +4453,12 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
               // pre-reconnect "expired" outlive the reconnect; the next health
               // check writes the verdict for the new grant.
               last_health: null,
+              // A fresh grant invalidates the catalog's freshness even when
+              // its remote rebuild runs after the OAuth callback responds.
+              // If that background task is interrupted, the next tools read
+              // sees this marker and converges it through the normal stale
+              // catalog path.
+              tools_synced_at: null,
               updated_at: now,
             };
             if (existing) {
@@ -4599,13 +4605,44 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
               );
             }
 
-            // Produce + persist tools for the minted connection (same path
-            // connections.create uses).
-            yield* produceConnectionTools(integrationRow, ref).pipe(
+            // The connection row and credential are already durable. Interactive
+            // OAuth callbacks return at this boundary and let remote discovery run
+            // under the host's keep-alive; otherwise a slow MCP listTools call can
+            // keep the popup open until the Worker request is cancelled. Explicit
+            // mints (for example client_credentials) retain the original contract.
+            const syncTools = produceConnectionTools(
+              integrationRow,
+              ref,
+              input.toolSync ?? "explicit",
+            ).pipe(
               Effect.catchTag("IntegrationNotFoundError", () =>
                 Effect.succeed([] as readonly Tool[]),
               ),
             );
+            if (input.toolSync === "background") {
+              const fiber = yield* Effect.forkDetach(
+                syncTools.pipe(
+                  Effect.catch((error) =>
+                    Effect.logWarning("executor OAuth tool sync failed", {
+                      integration: String(ref.integration),
+                      connection: String(ref.name),
+                      error: describeSyncFailure(error),
+                    }),
+                  ),
+                  Effect.withSpan("executor.oauth.tools.sync", {
+                    attributes: {
+                      "executor.integration": String(ref.integration),
+                      "executor.connection": String(ref.name),
+                    },
+                  }),
+                ),
+              );
+              config.waitUntil?.(
+                new Promise<void>((resolve) => fiber.addObserver(() => resolve(undefined))),
+              );
+            } else {
+              yield* syncTools;
+            }
           }),
         );
 

packages/core/sdk/src/oauth-client.ts

@@ -376,6 +376,13 @@ export interface OAuthCompleteInput {
   readonly callbackDomain?: string | null;
 }
 
+/** Host-lifecycle behavior for OAuth completion. The HTTP popup uses
+ *  background tool synchronization so it can close after the durable grant;
+ *  programmatic callers keep the default explicit catalog guarantee. */
+export interface OAuthCompleteOptions {
+  readonly toolSync?: "explicit" | "background";
+}
+
 /** Probe a base/issuer URL for OAuth 2.1 authorization-server metadata so the
  *  onboarding UI can pre-fill a client's endpoints. */
 export interface OAuthProbeInput {
@@ -535,6 +542,7 @@ export interface OAuthService {
   ) => Effect.Effect<ConnectResult, OAuthStartError | OrgWriteDeniedError | StorageFailure>;
   readonly complete: (
     input: OAuthCompleteInput,
+    options?: OAuthCompleteOptions,
   ) => Effect.Effect<
     Connection,
     OAuthCompleteError | OAuthSessionNotFoundError | OrgWriteDeniedError | StorageFailure

packages/core/sdk/src/oauth-flow.test.ts

@@ -1,5 +1,5 @@
 import { describe, expect, it } from "@effect/vitest";
-import { Deferred, Effect, Fiber, Predicate } from "effect";
+import { Deferred, Effect, Fiber, Option, Predicate } from "effect";
 import { withQueryContext } from "@executor-js/fumadb/query";
 
 import {
@@ -264,6 +264,98 @@ describe("oauth.start / oauth.complete", () => {
       ),
   );
 
+  it.effect("complete returns after the durable grant while remote tool discovery continues", () =>
+    Effect.scoped(
+      Effect.gen(function* () {
+        const discoveryStarted = yield* Deferred.make<void>();
+        const releaseDiscovery = yield* Deferred.make<void>();
+        const keptAlive: Promise<unknown>[] = [];
+        const slowOAuthPlugin = definePlugin(() => ({
+          id: "acme" as const,
+          storage: () => ({}),
+          resolveTools: () =>
+            Effect.gen(function* () {
+              yield* Deferred.succeed(discoveryStarted, undefined);
+              yield* Deferred.await(releaseDiscovery);
+              return {
+                tools: [{ name: ToolName.make("whoami"), description: "whoami" }],
+              };
+            }),
+          describeAuthMethods: () => [
+            {
+              id: "oauth",
+              label: "OAuth2",
+              kind: "oauth" as const,
+              template: String(TEMPLATE),
+              oauth: { scopes: ["read"] },
+            },
+          ],
+          invokeTool: ({ credential }) => Effect.succeed({ token: credential.value }),
+          extension: (ctx) => ({
+            seed: () =>
+              ctx.core.integrations.register({
+                slug: INTEG,
+                description: "Slow Acme",
+                config: {},
+              }),
+          }),
+        }))();
+        const server = yield* serveOAuthTestServer({ scopes: ["read"] });
+        const { executor } = yield* makeTestWorkspaceHarness({
+          plugins: [memoryCredentialsPlugin(), slowOAuthPlugin] as const,
+          waitUntil: (promise) => keptAlive.push(promise),
+        });
+        yield* Effect.addFinalizer(() =>
+          Deferred.succeed(releaseDiscovery, undefined).pipe(
+            Effect.andThen(Effect.promise(() => Promise.all(keptAlive))),
+          ),
+        );
+        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",
+        });
+        const started = yield* executor.oauth.start({
+          owner: "org",
+          client: CLIENT,
+          clientOwner: "org",
+          name: ConnectionName.make("main-account"),
+          integration: INTEG,
+          template: TEMPLATE,
+        });
+        expect(started.status).toBe("redirect");
+        if (started.status !== "redirect") return;
+        const callback = yield* server.completeAuthorizationCodeFlow({
+          authorizationUrl: started.authorizationUrl,
+        });
+
+        const completed = yield* executor.oauth
+          .complete({ state: started.state, code: callback.code }, { toolSync: "background" })
+          .pipe(Effect.timeoutOption("1 second"));
+        expect(
+          Option.isSome(completed),
+          "the callback returns while listTools remains deliberately blocked",
+        ).toBe(true);
+        expect(keptAlive).toHaveLength(1);
+        yield* Deferred.await(discoveryStarted);
+
+        const connections = yield* executor.connections.list({ integration: INTEG });
+        expect(connections.map((connection) => String(connection.name))).toEqual(["mainAccount"]);
+
+        yield* Deferred.succeed(releaseDiscovery, undefined);
+        yield* Effect.promise(() => Promise.all(keptAlive));
+        const tools = yield* executor.tools.list({ integration: INTEG });
+        expect(tools.map((tool) => String(tool.name))).toEqual(["whoami"]);
+      }),
+    ),
+  );
+
   it.effect("persists HTTP Basic client auth for code exchange and refresh", () =>
     Effect.scoped(
       Effect.gen(function* () {