From 46d670152b01444e9f9edf3d79a7617171a4bffd Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Mon, 14 Sep 2026 10:39:19 -0700 Subject: [PATCH 1/2] Cascade integration removal to every member and stop serving orphaned rows --- .changeset/remove-integration-cascade.md | 5 + packages/core/sdk/src/errors.ts | 4 +- packages/core/sdk/src/executor.ts | 64 +++++- packages/core/sdk/src/fuma-runtime.ts | 12 +- .../src/integration-removal-cascade.test.ts | 191 ++++++++++++++++++ packages/core/sdk/src/oauth-flow.test.ts | 39 ++++ packages/core/sdk/src/oauth-service.ts | 14 ++ packages/core/sdk/src/owner-policy.ts | 19 +- packages/core/sdk/src/platform-view.test.ts | 8 + 9 files changed, 345 insertions(+), 11 deletions(-) create mode 100644 .changeset/remove-integration-cascade.md create mode 100644 packages/core/sdk/src/integration-removal-cascade.test.ts diff --git a/.changeset/remove-integration-cascade.md b/.changeset/remove-integration-cascade.md new file mode 100644 index 0000000000..e970b2a0ae --- /dev/null +++ b/.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. diff --git a/packages/core/sdk/src/errors.ts b/packages/core/sdk/src/errors.ts index 2e22ff0ca7..d0f16e7d12 100644 --- a/packages/core/sdk/src/errors.ts +++ b/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; diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index efaf119212..6bfe859466 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -1986,6 +1986,22 @@ export const createExecutor = (effect: Effect.Effect) => fuma.transaction(effect); @@ -3071,6 +3087,13 @@ export const createExecutor = 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, 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 = 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 = = Readonly<{ export interface MakeFumaClientOptions { readonly tables?: ReadonlySet; + /** 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 | undefined, table: PropertyKey): boolean => @@ -347,9 +353,11 @@ const makeSafeFumaQuery = ( }; 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 = (effect: Effect.Effect): Effect.Effect => diff --git a/packages/core/sdk/src/integration-removal-cascade.test.ts b/packages/core/sdk/src/integration-removal-cascade.test.ts new file mode 100644 index 0000000000..bc507efbc9 --- /dev/null +++ b/packages/core/sdk/src/integration-removal-cascade.test.ts @@ -0,0 +1,191 @@ +import { describe, expect, it } from "@effect/vitest"; +import { Effect, Predicate } from "effect"; +import { withQueryContext } from "@executor-js/fumadb/query"; + +import { createExecutor, type Executor } from "./executor"; +import { type StorageError } from "./fuma-runtime"; +import { + AuthTemplateSlug, + ConnectionName, + IntegrationSlug, + ProviderKey, + Subject, + ToolAddress, + ToolName, +} from "./ids"; +import { definePlugin } from "./plugin"; +import type { CredentialProvider } from "./provider"; +import { makeTestConfig } from "./testing"; + +// --------------------------------------------------------------------------- +// Removing an integration must cascade to EVERY member's rows under it, not +// just the remover's own. Before this, a bound admin's delete only reached +// its own subject's connections and tools; everyone else's survived as +// orphans — absent from the catalog, yet still listed to agents and still a +// valid `oauth.start` target that failed only at the mint. +// +// The fixtures build TWO executors over ONE test database, bound to two +// different subjects. `alice` (the admin) seeds and removes; `bob` connects. +// --------------------------------------------------------------------------- + +const memoryProvider = (): CredentialProvider => { + const store = new Map(); + return { + key: ProviderKey.make("memory"), + writable: true, + get: (id) => Effect.sync(() => store.get(String(id)) ?? null), + set: (id, value) => Effect.sync(() => void store.set(String(id), value)), + delete: (id) => Effect.sync(() => void store.delete(String(id))), + }; +}; + +const INTEG = IntegrationSlug.make("datadog"); +const KEPT = IntegrationSlug.make("linear"); +const TEMPLATE = AuthTemplateSlug.make("apiKey"); +const ALICE = "user_alice"; +const BOB = "user_bob"; + +const demoPlugin = definePlugin(() => ({ + id: "demo" as const, + credentialProviders: [memoryProvider()], + storage: () => ({}), + resolveTools: () => + Effect.succeed({ + tools: [{ name: ToolName.make("query"), description: "query" }], + }), + invokeTool: ({ toolRow }) => Effect.succeed({ ran: toolRow.name }), + extension: (ctx) => ({ + seed: (slug: IntegrationSlug) => + ctx.core.integrations.register({ slug, description: String(slug), config: {} }), + }), +}))(); + +const setup = () => + Effect.gen(function* () { + const config = makeTestConfig({ plugins: [demoPlugin] as const, subject: ALICE }); + const alice = yield* createExecutor(config); + const bob = yield* createExecutor({ + ...config, + subject: Subject.make(BOB), + db: withQueryContext(config.testDb.db, { tenant: String(config.tenant), subject: BOB }), + }); + yield* Effect.addFinalizer(() => + alice.close().pipe(Effect.andThen(bob.close()), Effect.ignore), + ); + yield* alice.demo.seed(INTEG); + yield* alice.demo.seed(KEPT); + return { alice, bob, config }; + }); + +const connectPersonal = (executor: Executor, integration: IntegrationSlug, name: string) => + executor.connections.create({ + owner: "user", + name: ConnectionName.make(name), + integration, + template: TEMPLATE, + value: `token-${name}`, + }); + +describe("integrations.remove cascade", () => { + it.effect("drops every subject's connections and tools under the removed slug", () => + Effect.gen(function* () { + const { alice, bob, config } = yield* setup(); + yield* connectPersonal(alice, INTEG, "aliceDd"); + yield* connectPersonal(bob, INTEG, "bobDd"); + yield* connectPersonal(bob, KEPT, "bobLinear"); + + yield* alice.integrations.remove(INTEG); + + // Tenant-wide view of what is actually stored, across both subjects. + const wide = withQueryContext(config.testDb.db, { + tenant: String(config.tenant), + subject: null, + reach: "tenant", + }); + const connectionRows = yield* Effect.promise(() => wide.findMany("connection", {})); + const toolRows = yield* Effect.promise(() => wide.findMany("tool", {})); + expect(connectionRows.map((row) => `${row["subject"]}:${row["integration"]}`).sort()).toEqual( + [`${BOB}:${KEPT}`], + ); + expect(toolRows.every((row) => row["integration"] === String(KEPT))).toBe(true); + expect(toolRows.length).toBeGreaterThan(0); + + // Bob's other integration is untouched, and he sees it. + const bobConnections = yield* bob.connections.list(); + expect(bobConnections.map((c) => String(c.integration))).toEqual([String(KEPT)]); + }).pipe(Effect.scoped), + ); + + it.effect("a member (org writes denied) still cannot remove", () => + Effect.gen(function* () { + const { config } = yield* setup(); + const member = yield* createExecutor({ ...config, orgWrites: "denied" }); + yield* Effect.addFinalizer(() => member.close().pipe(Effect.ignore)); + const error = yield* Effect.flip(member.integrations.remove(INTEG)); + expect(Predicate.isTagged("OrgWriteDeniedError")(error)).toBe(true); + }).pipe(Effect.scoped), + ); + + it.effect("the platform view cannot remove (read-only holds ahead of the cascade)", () => + Effect.gen(function* () { + const { config } = yield* setup(); + const platform = yield* createExecutor({ ...config, platformView: true }); + yield* Effect.addFinalizer(() => platform.close().pipe(Effect.ignore)); + yield* platform.integrations.remove(INTEG).pipe( + Effect.flatMap(() => Effect.die("expected the platform view to refuse the removal")), + Effect.catchTag("StorageError", (error: StorageError) => { + expect(error.message).toContain("read-only"); + return Effect.void; + }), + Effect.orDie, + ); + }).pipe(Effect.scoped), + ); +}); + +describe("orphaned rows are not served", () => { + // Simulate the pre-fix state: rows whose integration is gone from the + // catalog. Delete the catalog row directly, bypassing the cascade. + const orphanBob = (config: ReturnType) => + Effect.promise(() => + withQueryContext(config.testDb.db, { + tenant: String(config.tenant), + subject: ALICE, + }).deleteMany("integration", { where: (b) => b("slug", "=", String(INTEG)) }), + ); + + it.effect( + "tools.list and connections.list hide them; invoke reports the missing integration", + () => + Effect.gen(function* () { + const { bob, config } = yield* setup(); + yield* connectPersonal(bob, INTEG, "bobDd"); + yield* connectPersonal(bob, KEPT, "bobLinear"); + const address = ToolAddress.make(`tools.${INTEG}.user.bobDd.query`); + expect(yield* bob.execute(address, {})).toEqual({ ran: "query" }); + + yield* orphanBob(config); + // The orphan is still stored — only its catalog row is gone. + const stored = yield* Effect.promise(() => + withQueryContext(config.testDb.db, { + tenant: String(config.tenant), + subject: BOB, + }).findMany("tool", {}), + ); + expect(stored.map((row) => String(row.integration)).sort()).toEqual([ + String(INTEG), + String(KEPT), + ]); + + const tools = yield* bob.tools.list(); + expect(tools.filter((t) => !t.static).map((t) => String(t.integration))).toEqual([ + String(KEPT), + ]); + const connections = yield* bob.connections.list(); + expect(connections.map((c) => String(c.integration))).toEqual([String(KEPT)]); + + const error = yield* Effect.flip(bob.execute(address, {})); + expect(Predicate.isTagged("IntegrationNotFoundError")(error)).toBe(true); + }).pipe(Effect.scoped), + ); +}); diff --git a/packages/core/sdk/src/oauth-flow.test.ts b/packages/core/sdk/src/oauth-flow.test.ts index b9860574bb..319bf22db9 100644 --- a/packages/core/sdk/src/oauth-flow.test.ts +++ b/packages/core/sdk/src/oauth-flow.test.ts @@ -1003,6 +1003,45 @@ describe("oauth.start / oauth.complete", () => { }), ), ); + + it.effect("start refuses an integration that is not in the catalog, before any session", () => + Effect.scoped( + Effect.gen(function* () { + const server = yield* serveOAuthTestServer({ scopes: ["read"] }); + const { executor, config } = yield* makeTestWorkspaceHarness({ plugins }); + // Deliberately NOT seeded: the slug names nothing in the catalog — the + // shape of a reconnect against a connection whose integration was + // removed, or an agent replaying a stale slug. + yield* executor.oauth.createClient({ + owner: "org", + slug: CLIENT, + authorizationUrl: server.authorizationEndpoint, + tokenUrl: server.tokenEndpoint, + grant: "authorization_code", + clientId: "test-client", + clientSecret: "test-secret", + }); + + const error = yield* Effect.flip( + executor.oauth.start({ + owner: "org", + client: CLIENT, + clientOwner: "org", + name: ConnectionName.make("main"), + integration: IntegrationSlug.make("removed_mcp"), + template: TEMPLATE, + }), + ); + expect(Predicate.isTagged("OAuthStartError")(error)).toBe(true); + if (!Predicate.isTagged("OAuthStartError")(error)) return; + const startError = error as OAuthStartError; + expect(startError.message).toBe("Integration not found: removed_mcp"); + // Refused up front: no session row was created for the doomed flow. + const sessions = yield* Effect.promise(() => config.db.findMany("oauth_session", {})); + expect(sessions).toHaveLength(0); + }), + ), + ); }); describe("oauth token refresh in resolveConnectionValue", () => { diff --git a/packages/core/sdk/src/oauth-service.ts b/packages/core/sdk/src/oauth-service.ts index dd443144bd..5fcf64c26f 100644 --- a/packages/core/sdk/src/oauth-service.ts +++ b/packages/core/sdk/src/oauth-service.ts @@ -227,6 +227,11 @@ export interface OAuthServiceDeps { readonly mintOAuthConnection: ( input: MintOAuthConnectionInput, ) => Effect.Effect; + /** Whether `slug` is in the tenant's integration catalog. `start` refuses a + * flow for a missing integration up front — otherwise the user authorizes + * at the provider and only the mint at `complete` discovers there is + * nothing to mint against (the reconnect-an-orphan failure). */ + readonly integrationExists: (slug: IntegrationSlug) => Effect.Effect; /** Whether a connection row exists under `(owner, integration, name)`: the * raw row, not the policy-filtered list, so `start` can resolve a free * name for `newConnection` flows against what is actually stored. */ @@ -1659,6 +1664,15 @@ export const makeOAuthService = (deps: OAuthServiceDeps): OAuthService => { cause, }), }); + // The integration must exist BEFORE any session or provider round trip. + // A stale reference (a connection whose integration was removed, or an + // agent replaying an old slug) would otherwise complete authorization at + // the provider and fail only at the mint. + if (!(yield* deps.integrationExists(input.integration))) { + return yield* new OAuthStartError({ + message: `Integration not found: ${String(input.integration)}`, + }); + } // Sharing is one-directional (org → members): a Workspace (org) connection // cannot be backed by a member's private (user) app. The connection owner // and the app owner are otherwise independent — a Personal connection diff --git a/packages/core/sdk/src/owner-policy.ts b/packages/core/sdk/src/owner-policy.ts index 9f8b487b59..4c2e16afa9 100644 --- a/packages/core/sdk/src/owner-policy.ts +++ b/packages/core/sdk/src/owner-policy.ts @@ -41,7 +41,12 @@ export type ExecutorWrites = /** The default: writes are allowed, bounded by `assertOwnerWritable`. */ | "allowed" /** Every create/update/delete is rejected outright, at every reach. */ - | "denied"; + | "denied" + /** Deletes only, at the context's reach; creates and updates are rejected. + * The one sanctioned tenant-wide mutation: a catalog removal cascading to + * every subject's rows under the removed integration. Built only inside + * that cascade and never handed to a plugin or a request surface. */ + | "delete-only"; export interface ExecutorOwnerPolicyContext { readonly tenant: string; @@ -131,6 +136,12 @@ export const ownerVisibilityCondition = ( * Every write path calls this first and fails loudly — a write arriving on a * read-only handle is a programmer error, not something to quietly demote to * bound behavior. + * + * The single exception is `writes: "delete-only"`: a tenant-reach context that + * may DELETE (and only delete) across every subject. It exists for 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 leftovers would otherwise survive as orphans that agents still see. */ export const assertReachReadOnly = ( tableName: string, @@ -138,6 +149,12 @@ export const assertReachReadOnly = ( context: ExecutorOwnerPolicyContext | undefined, ): void => { if (context === undefined) return; + if (context.writes === "delete-only") { + if (access === "delete") return; + policyViolation( + `Storage ${access} on table "${tableName}" is not allowed: this context may only delete.`, + ); + } if (context.reach !== "tenant" && context.writes !== "denied") return; policyViolation( `Storage ${access} on table "${tableName}" is not allowed: the platform view is read-only.`, diff --git a/packages/core/sdk/src/platform-view.test.ts b/packages/core/sdk/src/platform-view.test.ts index 5eabbfd3b5..12cee13c8b 100644 --- a/packages/core/sdk/src/platform-view.test.ts +++ b/packages/core/sdk/src/platform-view.test.ts @@ -501,6 +501,10 @@ describe("platform view — read-only across every surface", () => { withDb((db) => Effect.gen(function* () { yield* seed(db); + // The list surface serves only catalog-backed connections. + yield* insertIntegration(db, "github"); + yield* insertIntegration(db, "linear"); + yield* insertIntegration(db, "stripe"); const executor = yield* makePlatformExecutor(db, { subject: null }); const connections = yield* executor.connections.list().pipe(Effect.orDie); @@ -799,6 +803,10 @@ describe("platform view — default off", () => { withDb((db) => Effect.gen(function* () { yield* seed(db); + // The list surface serves only catalog-backed connections. + yield* insertIntegration(db, "github"); + yield* insertIntegration(db, "linear"); + yield* insertIntegration(db, "stripe"); // Enabling the platform view must not widen ANY existing surface. const executor = yield* makePlatformExecutor(db, { subject: SUBJECT_A }); From a946394286388acaff959882837ac12c191e8819 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Mon, 14 Sep 2026 10:58:45 -0700 Subject: [PATCH 2/2] Register the MCP server before starting OAuth in the local app test --- apps/local/src/mcp-oauth.test.ts | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/apps/local/src/mcp-oauth.test.ts b/apps/local/src/mcp-oauth.test.ts index 3c9d848233..405471486f 100644 --- a/apps/local/src/mcp-oauth.test.ts +++ b/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; readonly dispose: () => Promise; } @@ -132,6 +136,18 @@ const startHarness = async (tmpDir: string): Promise => { 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) =>