Cascade integration removal to every member and stop serving orphaned rows (#1991)

* Cascade integration removal to every member and stop serving orphaned rows

* Register the MCP server before starting OAuth in the local app test

Author: RhysSullivan

.changeset/remove-integration-cascade.md

@@ -0,0 +1,5 @@
+---
+"@executor-js/sdk": patch
+---
+
+Removing an integration now drops every member's connections and tools under it, not only the remover's own. Tool and connection listings no longer serve rows whose integration is gone from the catalog, invoking such a tool reports the missing integration, and `oauth.start` refuses an unknown integration before creating a session.

apps/local/src/mcp-oauth.test.ts

@@ -73,6 +73,10 @@ const TEST_BASE_URL = "http://local.test";
 
 interface Harness {
   readonly fetch: typeof globalThis.fetch;
+  readonly registerRemoteServer: (input: {
+    readonly slug: string;
+    readonly endpoint: string;
+  }) => Effect.Effect<void, unknown>;
   readonly dispose: () => Promise<void>;
 }
 
@@ -132,6 +136,18 @@ const startHarness = async (tmpDir: string): Promise<Harness> => {
       webHandler(
         input instanceof Request ? input : new Request(input, init),
       )) as typeof globalThis.fetch,
+    // `oauth.start` refuses an integration that is not in the catalog, so the
+    // flow under test needs a registered MCP server to mint against.
+    registerRemoteServer: ({ slug, endpoint }) =>
+      executor.mcp
+        .addServer({
+          transport: "remote",
+          name: slug,
+          slug,
+          endpoint,
+          authenticationTemplate: [{ kind: "oauth2", slug: "oauth" }],
+        })
+        .pipe(Effect.asVoid),
     dispose: async () => {
       await Effect.runPromise(Effect.ignore(Effect.tryPromise(() => disposeHandler())));
       await Effect.runPromise(
@@ -191,6 +207,12 @@ describe("local oauth (real OAuth discovery + stubbed start)", () => {
           expect(probed.authorizationUrl).toBe(oauth.authorizationEndpoint);
           expect(probed.tokenUrl).toBe(oauth.tokenEndpoint);
 
+          // The catalog row `start` mints against.
+          yield* harness.registerRemoteServer({
+            slug: "mcp_remote",
+            endpoint: oauth.mcpResourceUrl,
+          });
+
           // createClient — register an owner-scoped OAuth app for the start flow.
           const slug = `mcp-oauth2-${randomBytes(4).toString("hex")}`;
           const created = yield* run((client) =>

packages/core/sdk/src/errors.ts

@@ -290,6 +290,9 @@ export type ExecuteError =
   | PluginNotLoadedError
   | NoHandlerError
   | ConnectionNotFoundError
+  /** The tool row outlived its integration (an orphan the catalog no longer
+   *  lists), so there is no plugin config to invoke it against. */
+  | IntegrationNotFoundError
   | CredentialProviderNotRegisteredError
   | CredentialResolutionError
   | ElicitationDeclinedError
@@ -298,6 +301,5 @@ export type ExecuteError =
 /** Convenience union spanning every typed error the SDK raises. */
 export type ExecutorError =
   | ExecuteError
-  | IntegrationNotFoundError
   | IntegrationRemovalNotAllowedError
   | ArtifactNotFoundError;

packages/core/sdk/src/executor.ts

@@ -1986,6 +1986,22 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
     const healthProbeInFlight = healthProbeGateFor(rootDbUntyped);
     const fuma = makeFumaClient(rootDb);
     const core = makeCoreDb(fuma);
+    // The ONE tenant-wide mutating handle: delete-only, tenant reach. Used
+    // solely by the integration-removal cascade, which must drop EVERY
+    // member's connections and tools under the removed slug — a bound admin
+    // can only reach its own rows, and the rest would survive as orphans that
+    // still list and invoke. The context rebinds inside the removal
+    // transaction, so the cascade commits or rolls back with the catalog row.
+    // Never exposed to plugins or request surfaces.
+    const cascadeCore = makeCoreDb(
+      makeFumaClient(rootDb, {
+        context: {
+          ...ownerContext,
+          reach: "tenant",
+          writes: "delete-only",
+        } satisfies ExecutorOwnerPolicyContext,
+      }),
+    );
     const blobs = config.blobs ?? makeFumaBlobStore(fuma);
     const transaction = <A, E>(effect: Effect.Effect<A, E>) => fuma.transaction(effect);
 
@@ -3071,6 +3087,13 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
         where: (b: AnyCb) => b("slug", "=", String(slug)),
       });
 
+    /** Every slug in the tenant's catalog — the set an owned row's
+     *  `integration` must belong to for the row to be servable. */
+    const listCatalogSlugs = (): Effect.Effect<ReadonlySet<string>, StorageFailure> =>
+      core
+        .findMany("integration", { select: ["slug"] })
+        .pipe(Effect.map((rows) => new Set(rows.map((row) => String(row.slug)))));
+
     // Project a row's stored config into declared auth methods via the owning
     // plugin's `describeAuthMethods` hook. The hook is plugin-authored, so a
     // throw (malformed config it didn't guard) degrades to `[]` rather than
@@ -3337,14 +3360,21 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
                 ),
               );
           }
-          // Drop owned connections / tools / definitions for this integration.
-          const where = (b: AnyCb) => b("integration", "=", String(slug));
-          yield* core.deleteMany("tool", { where });
-          yield* core.deleteMany("definition", { where });
-          yield* core.deleteMany("connection", { where });
+          // The catalog row goes first through the bound handle: a read-only
+          // (platform-view) context is refused here, before the widened
+          // cascade below could touch anything.
           yield* core.deleteMany("integration", {
             where: (b: AnyCb) => b("slug", "=", String(slug)),
           });
+          // Drop connections / tools / definitions for this integration across
+          // EVERY subject in the tenant, not just the remover's own rows. A
+          // removed integration has no reason to keep anyone's rows, and rows
+          // left behind become orphans: invisible in the catalog, yet still
+          // listed to agents and still targetable by reconnect.
+          const where = (b: AnyCb) => b("integration", "=", String(slug));
+          yield* cascadeCore.deleteMany("tool", { where });
+          yield* cascadeCore.deleteMany("definition", { where });
+          yield* cascadeCore.deleteMany("connection", { where });
           return existing.plugin_id;
         }),
       ).pipe(
@@ -4650,7 +4680,13 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
               filter?.owner === undefined ? true : b("owner", "=", filter.owner),
             ),
         });
-        const connections = rows.map(rowToConnection);
+        // Same catalog gate as `toolsList`: a connection whose integration was
+        // removed is an orphan, and offering it (in the accounts list, or to
+        // an agent as a reconnect target) leads into flows that cannot mint.
+        const catalogSlugs = yield* listCatalogSlugs();
+        const connections = rows
+          .filter((row) => catalogSlugs.has(String(row.integration)))
+          .map(rowToConnection);
         if (!activeToolPolicyProvider) return connections;
 
         const visibleTools = yield* toolsList({ includeAnnotations: false });
@@ -5587,8 +5623,14 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
         });
         const includeBlocked = filter?.includeBlocked ?? false;
         const policyRules = yield* listActivePolicyRuleSet();
+        // Only tools whose integration is still in the catalog. A tool row
+        // whose integration was removed is an orphan (a removal that could
+        // not reach this subject's rows): listing it invites an invoke that
+        // cannot resolve its config and a reconnect that cannot mint.
+        const catalogSlugs = yield* listCatalogSlugs();
         const tools: Tool[] = [];
         for (const row of rows) {
+          if (!catalogSlugs.has(String(row.integration))) continue;
           const tool = rowToTool(row);
           if (!matchesToolFilter(tool, filter)) continue;
           if (!includeBlocked) {
@@ -6489,6 +6531,13 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
             Effect.onError(() => Fiber.interrupt(integrationRowFiber)),
           );
           const integrationRow = yield* Fiber.join(integrationRowFiber);
+          // A tool row that outlived its integration (an orphan the catalog no
+          // longer lists) is not invokable: its plugin config is gone, and
+          // the auth-recovery hints would steer the caller into an OAuth
+          // flow that cannot mint. Report it as the missing integration it is.
+          if (!integrationRow) {
+            return yield* new IntegrationNotFoundError({ slug: parsed.integration });
+          }
           const grantedScopes = grantedScopesFromRow(connectionRow);
           const invokeTool = runtime.plugin.invokeTool;
           const invokeWith = (
@@ -6501,7 +6550,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
               template: AuthTemplateSlug.make(connectionRow.template),
               value: resolved[PRIMARY_INPUT_VARIABLE] ?? null,
               values: resolved,
-              config: integrationRow ? decodeJsonColumn(integrationRow.config) : undefined,
+              config: decodeJsonColumn(integrationRow.config),
               ...(grantedScopes ? { grantedScopes } : {}),
             };
             return wrapInvocationError(
@@ -6609,6 +6658,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
       guardOrgWrite: (owner: Owner) => guardOrgWrite(owner),
       defaultWritableProvider,
       mintOAuthConnection: (input: MintOAuthConnectionInput) => mintOAuthConnection(input),
+      integrationExists: (slug) => findIntegrationRow(slug).pipe(Effect.map((row) => row !== null)),
       connectionNameTaken: (ref) => findConnectionRow(ref).pipe(Effect.map((row) => row !== null)),
       // One integration-row read + one projector run. Resolve the method this
       // template selects exactly as the runtime's `selectAuthMethod` does —

packages/core/sdk/src/fuma-runtime.ts

@@ -1,5 +1,5 @@
 import { Cause, Context, Data, Effect, Exit, Layer, Predicate } from "effect";
-import type { AbstractQuery } from "@executor-js/fumadb/query";
+import { withQueryContext, type AbstractQuery } from "@executor-js/fumadb/query";
 import type { AnySchema, AnyTable, Schema as FumaSchema } from "@executor-js/fumadb/schema";
 
 export class StorageError extends Data.TaggedError("StorageError")<{
@@ -306,6 +306,12 @@ export type IFumaClient<TSchema extends AnySchema = AnySchema> = Readonly<{
 
 export interface MakeFumaClientOptions {
   readonly tables?: ReadonlySet<string>;
+  /** Owner-policy context to rebind EVERY query to, including queries issued
+   *  inside an enclosing transaction (whose handle otherwise carries the
+   *  context of whoever opened it). Lets a narrowly-scoped client (the
+   *  integration-removal cascade) join a bound transaction without inheriting
+   *  the bound reach. */
+  readonly context?: unknown;
 }
 
 const isAllowedTable = (tables: ReadonlySet<string> | undefined, table: PropertyKey): boolean =>
@@ -347,9 +353,11 @@ const makeSafeFumaQuery = <TSchema extends AnySchema>(
 };
 
 export const makeFumaClient = (db: FumaDb, options: MakeFumaClientOptions = {}): IFumaClient => {
+  const rebind = (handle: FumaDb): FumaDb =>
+    options.context === undefined ? handle : withQueryContext(handle, options.context);
   const use: IFumaClient["use"] = (label, fn) =>
     Effect.flatMap(Effect.service(activeFumaDbRef), (active) =>
-      fumaEffect(label, () => fn(makeSafeFumaQuery(active ?? db, options))),
+      fumaEffect(label, () => fn(makeSafeFumaQuery(rebind(active ?? db), options))),
     ).pipe(Effect.withSpan(`fumadb.${label}`));
 
   const transaction = <A, E>(effect: Effect.Effect<A, E>): Effect.Effect<A, E | StorageFailure> =>