From 479a8cfd1780f3fcfde071cb2bf868820d4739c0 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Tue, 25 Aug 2026 12:47:18 -0700 Subject: [PATCH 1/4] Gate enterprise-managed authorization behind a host-owned rollout seam --- packages/core/sdk/src/executor.ts | 35 +- packages/core/sdk/src/index.ts | 12 + .../core/sdk/src/oauth-ema-rollout.test.ts | 467 ++++++++++++++++++ packages/core/sdk/src/oauth-ema.ts | 114 ++++- packages/core/sdk/src/oauth-service.ts | 248 +++++++--- packages/core/sdk/src/test-config.ts | 2 + 6 files changed, 803 insertions(+), 75 deletions(-) create mode 100644 packages/core/sdk/src/oauth-ema-rollout.test.ts diff --git a/packages/core/sdk/src/executor.ts b/packages/core/sdk/src/executor.ts index d59328daed..f1b9443477 100644 --- a/packages/core/sdk/src/executor.ts +++ b/packages/core/sdk/src/executor.ts @@ -173,6 +173,7 @@ import { enterpriseManagedStateFrom, mintEnterpriseManagedAccessToken, type EnterpriseManagedMintError, + type EnterpriseManagedRollout, } from "./oauth-ema"; import { connectionIdentifier } from "./connection-name-identifier"; import { annotateToolResultOutcome } from "./tool-result"; @@ -633,6 +634,25 @@ export interface ExecutorConfig enterprise-managed authorization is attempted, which is exactly + * what every host did before this seam existed. + */ + readonly enterpriseManagedRollout?: EnterpriseManagedRollout; /** * Host-operated OAuth apps (the deployment's own registered GitHub/Google/… * apps), addressed as `first-party:`. Users connect through them with @@ -1885,7 +1905,16 @@ export const createExecutor = readonly EnterpriseManagedRolloutContext[]; + readonly events: () => readonly EnterpriseManagedRolloutEvent[]; +} + +const scriptedRollout = (options: { + /** Verdict for the nth consultation; the last entry repeats forever. */ + readonly answers: readonly EnterpriseManagedRolloutDecision[]; + /** Make `record` blow up, to prove observation cannot reach the caller. */ + readonly recordDies?: boolean; +}): ScriptedRollout => { + const consultations: EnterpriseManagedRolloutContext[] = []; + const events: EnterpriseManagedRolloutEvent[] = []; + return { + consultations: () => consultations, + events: () => events, + rollout: { + decide: (context) => + Effect.sync(() => { + consultations.push(context); + const index = Math.min(consultations.length - 1, options.answers.length - 1); + return options.answers[index] ?? ENABLED; + }), + record: (event) => + options.recordDies + ? Effect.die("the flag service exploded") + : Effect.sync(() => { + events.push(event); + }), + }, + }; +}; + +const eventKinds = (spy: ScriptedRollout): readonly string[] => + spy.events().map((event) => event.kind); + +// --------------------------------------------------------------------------- +// Fixture: an integration whose only auth method is the enterprise-managed +// OAuth client, plus the two client registrations the profile needs. +// --------------------------------------------------------------------------- + +const oauthPlugin = definePlugin(() => ({ + id: "acme" as const, + storage: () => ({}), + resolveTools: () => + Effect.succeed({ + tools: [{ name: ToolName.make("whoami"), description: "whoami" }], + }), + describeAuthMethods: () => [ + { + id: "oauth", + label: "OAuth2", + kind: "oauth" as const, + template: String(TEMPLATE), + oauth: { scopes: ["mcp.read"] }, + }, + ], + invokeTool: ({ credential }) => Effect.succeed({ token: credential.value }), + extension: (ctx) => ({ + seed: () => + ctx.core.integrations.register({ + slug: INTEG, + description: "Acme", + config: {}, + }), + }), +}))(); + +const plugins = [memoryCredentialsPlugin(), oauthPlugin] as const; + +interface EnterpriseServers { + readonly idp: OAuthTestServerShape; + readonly resource: OAuthTestServerShape; + readonly subjectToken: string; +} + +const enterpriseServers = (options: { + readonly denyExchangeWith?: { + readonly error: string; + readonly errorDescription: string; + }; + readonly resourceTokenExpiresInSeconds?: number; +}) => + Effect.gen(function* () { + const idp = yield* serveOAuthTestServer({ + clients: { [CLIENT_AT_IDP]: null }, + scopes: ["mcp.read"], + enterpriseIdp: { + resourceClientIds: { [CLIENT_AT_IDP]: CLIENT_AT_RESOURCE }, + ...(options.denyExchangeWith ? { denyExchangeWith: options.denyExchangeWith } : {}), + }, + }); + const resource = yield* serveOAuthTestServer({ + clients: { [CLIENT_AT_RESOURCE]: null }, + scopes: ["mcp.read"], + ...(options.resourceTokenExpiresInSeconds === undefined + ? {} + : { tokenExpiresInSeconds: options.resourceTokenExpiresInSeconds }), + enterpriseResourceServer: { trustedIdpIssuer: idp.issuerUrl }, + }); + const session = yield* idp.completeAuthorizationCodeTokenFlow({ + clientId: CLIENT_AT_IDP, + clientSecret: "", + scopes: ["mcp.read"], + }); + return { + idp, + resource, + subjectToken: session.accessToken, + } satisfies EnterpriseServers; + }); + +const registerClients = (createClient: OAuthService["createClient"], servers: EnterpriseServers) => + Effect.gen(function* () { + yield* createClient({ + owner: "org", + slug: IDP_CLIENT, + authorizationUrl: servers.idp.authorizationEndpoint, + tokenUrl: servers.idp.tokenEndpoint, + grant: "authorization_code", + clientId: CLIENT_AT_IDP, + clientSecret: "", + }); + yield* createClient({ + owner: "org", + slug: RESOURCE_CLIENT, + authorizationUrl: servers.resource.authorizationEndpoint, + tokenUrl: servers.resource.tokenEndpoint, + grant: "id_jag", + clientId: CLIENT_AT_RESOURCE, + clientSecret: "", + resource: servers.resource.mcpResourceUrl, + }); + }); + +const startEnterpriseConnect = (servers: EnterpriseServers) => + ({ + owner: "org", + client: RESOURCE_CLIENT, + clientOwner: "org", + name: CONNECTION, + integration: INTEG, + template: TEMPLATE, + enterprise: { + idpClient: IDP_CLIENT, + idpClientOwner: "org", + subjectToken: servers.subjectToken, + subjectTokenType: ACCESS_TOKEN_TYPE, + }, + }) as const; + +/** How many RFC 8693 token exchanges the IdP has served. The identity assertion + * is spent here, so this is the measure of "did the client actually try". */ +const tokenExchangeCount = (servers: EnterpriseServers) => + servers.idp.requests.pipe( + Effect.map( + (entries) => + entries.filter((entry) => entry.path === "/token" && entry.body.includes("token-exchange")) + .length, + ), + ); + +const harness = (rollout: EnterpriseManagedRollout | undefined) => + makeTestWorkspaceHarness({ + plugins, + tenant: TENANT, + subject: SUBJECT, + enterpriseManagedRollout: rollout, + }); + +describe("enterprise-managed rollout gate", () => { + it.effect("attempts the enterprise-managed path when the gate allows it", () => + Effect.scoped( + Effect.gen(function* () { + const servers = yield* enterpriseServers({}); + const spy = scriptedRollout({ answers: [ENABLED] }); + const { executor } = yield* harness(spy.rollout); + yield* executor.acme.seed(); + yield* registerClients(executor.oauth.createClient, servers); + + const started = yield* executor.oauth.start(startEnterpriseConnect(servers)); + + assert(started.status === "connected"); + expect( + spy.consultations().length, + "one connect asks the gate once — never per request, never per step", + ).toBe(1); + expect(spy.consultations()[0], "the gate is handed the identity it rolls out by").toEqual({ + userId: SUBJECT, + organizationId: TENANT, + integration: INTEG, + }); + expect(eventKinds(spy)).toEqual(["attempted", "connected"]); + }), + ), + ); + + it.effect("falls back to the interactive flow when the gate withholds", () => + Effect.scoped( + Effect.gen(function* () { + const servers = yield* enterpriseServers({}); + const spy = scriptedRollout({ answers: [DISABLED] }); + const { executor } = yield* harness(spy.rollout); + yield* executor.acme.seed(); + yield* registerClients(executor.oauth.createClient, servers); + + const started = yield* executor.oauth.start(startEnterpriseConnect(servers)); + + expect( + started.status, + "a user outside the rollout gets exactly what a user connecting to a server without the profile gets", + ).toBe("redirect"); + expect( + yield* tokenExchangeCount(servers), + "the gate runs before discovery, so a withheld connect spends no identity assertion and makes no request", + ).toBe(0); + expect(spy.events()[0]?.decision).toEqual(DISABLED); + }), + ), + ); + + it.effect("fails closed to the interactive flow when the gate cannot reach a verdict", () => + Effect.scoped( + Effect.gen(function* () { + const servers = yield* enterpriseServers({}); + const spy = scriptedRollout({ answers: [UNAVAILABLE] }); + const { executor } = yield* harness(spy.rollout); + yield* executor.acme.seed(); + yield* registerClients(executor.oauth.createClient, servers); + + const started = yield* executor.oauth.start(startEnterpriseConnect(servers)); + + expect( + started.status, + "an unreachable flag service degrades the rollout; it must never fail the connect", + ).toBe("redirect"); + expect(yield* tokenExchangeCount(servers)).toBe(0); + expect( + spy.events()[0]?.decision, + "the reason travels, so an outage is distinguishable from a deliberate 'not yet'", + ).toEqual(UNAVAILABLE); + }), + ), + ); + + it.effect("attempts the enterprise-managed path when no host injected a gate", () => + Effect.scoped( + Effect.gen(function* () { + const servers = yield* enterpriseServers({}); + const { executor } = yield* harness(undefined); + yield* executor.acme.seed(); + yield* registerClients(executor.oauth.createClient, servers); + + const started = yield* executor.oauth.start(startEnterpriseConnect(servers)); + + expect( + started.status, + "desktop, CLI, local and self-host have no flag service, and their behavior must not change because cloud grew one", + ).toBe("connected"); + }), + ), + ); + + it.effect("never consults the gate again after the identity provider has denied", () => + Effect.scoped( + Effect.gen(function* () { + const servers = yield* enterpriseServers({ + denyExchangeWith: { + error: "unauthorized_client", + errorDescription: "This client is not approved for the requested MCP server.", + }, + }); + // The gate allows the attempt, then flips to every withheld verdict + // there is. If ANY code path re-read the flag after the denial, one of + // those answers would route this connect into the interactive flow — + // which is precisely the escape hatch around enterprise policy the + // profile exists to prevent. + const spy = scriptedRollout({ + answers: [ENABLED, DISABLED, UNAVAILABLE], + }); + const { executor } = yield* harness(spy.rollout); + yield* executor.acme.seed(); + yield* registerClients(executor.oauth.createClient, servers); + + const failure = yield* executor.oauth + .start(startEnterpriseConnect(servers)) + .pipe(Effect.flip); + + assert(Predicate.isTagged(failure, "OAuthStartError")); + expect( + failure.blockedByAdmin, + "the administrator's decision stands, and the flag is not a way around it", + ).toBe(true); + expect(failure.oauthErrorCode).toBe("unauthorized_client"); + expect( + spy.consultations().length, + "the gate is asked once, before discovery, and never again", + ).toBe(1); + expect( + (yield* executor.connections.list()).length, + "a denied connect leaves no connection behind for the flag to rescue", + ).toBe(0); + expect(eventKinds(spy)).toEqual(["attempted", "blocked-by-admin"]); + const blocked = spy.events()[1]; + assert(blocked?.kind === "blocked-by-admin"); + expect(blocked.oauthErrorCode, "the IdP's own code is what makes the event useful").toBe( + "unauthorized_client", + ); + }), + ), + ); + + it.effect("never re-evaluates the gate when renewing an existing connection", () => + Effect.scoped( + Effect.gen(function* () { + const servers = yield* enterpriseServers({ + resourceTokenExpiresInSeconds: 1, + }); + // Allowed for the connect, withheld from then on: the rollout is dialled + // back after this connection already exists. + const spy = scriptedRollout({ answers: [ENABLED, DISABLED] }); + const { executor } = yield* harness(spy.rollout); + yield* executor.acme.seed(); + yield* registerClients(executor.oauth.createClient, servers); + yield* executor.oauth.start(startEnterpriseConnect(servers)); + const consultationsAtConnect = spy.consultations().length; + + // Short-lived access tokens put the second execute inside the refresh + // skew, so this drives the real credential-renewal path. + const first = (yield* executor.execute(TOOL, {})) as { + readonly token: string; + }; + const second = (yield* executor.execute(TOOL, {})) as { + readonly token: string; + }; + + expect(second.token, "the expiring token was replaced").not.toBe(first.token); + expect( + yield* servers.resource.acceptsAccessToken(second.token), + "turning the flag off must not strand or silently downgrade a live managed connection", + ).toBe(true); + expect( + spy.consultations().length, + "credential resolution never asks the flag service anything — it follows the state persisted on the connection", + ).toBe(consultationsAtConnect); + }), + ), + ); + + it.effect("keeps every credential out of the rollout events", () => + Effect.scoped( + Effect.gen(function* () { + const servers = yield* enterpriseServers({}); + const spy = scriptedRollout({ answers: [ENABLED] }); + const { executor } = yield* harness(spy.rollout); + yield* executor.acme.seed(); + yield* registerClients(executor.oauth.createClient, servers); + + const started = yield* executor.oauth.start(startEnterpriseConnect(servers)); + assert(started.status === "connected"); + + const minted = (yield* executor.execute(TOOL, {})) as { + readonly token: string; + }; + const recorded = JSON.stringify(spy.events()); + + expect( + recorded, + "the identity assertion is the credential this whole profile turns on; it must never reach an analytics sink", + ).not.toContain(servers.subjectToken); + expect(recorded, "nor may the access token it was exchanged for").not.toContain( + minted.token, + ); + }), + ), + ); + + it.effect("cannot be failed by a rollout observer that throws", () => + Effect.scoped( + Effect.gen(function* () { + const servers = yield* enterpriseServers({}); + const spy = scriptedRollout({ answers: [ENABLED], recordDies: true }); + const { executor } = yield* harness(spy.rollout); + yield* executor.acme.seed(); + yield* registerClients(executor.oauth.createClient, servers); + + const started = yield* executor.oauth.start(startEnterpriseConnect(servers)); + + expect( + started.status, + "analytics is an observer; a broken one can never cost a user their connection", + ).toBe("connected"); + }), + ), + ); +}); diff --git a/packages/core/sdk/src/oauth-ema.ts b/packages/core/sdk/src/oauth-ema.ts index a485db49a8..8cc11bb01f 100644 --- a/packages/core/sdk/src/oauth-ema.ts +++ b/packages/core/sdk/src/oauth-ema.ts @@ -24,7 +24,7 @@ import { Data, Effect, Option, Schema } from "effect"; -import { OAuthClientSlug, Owner } from "./ids"; +import { IntegrationSlug, OAuthClientSlug, Owner } from "./ids"; import { DEFAULT_SUBJECT_TOKEN_TYPE, SubjectTokenTypeSchema, @@ -362,3 +362,115 @@ export const runEnterpriseManagedAuthorization = ( }, }); }); + +// --------------------------------------------------------------------------- +// Rollout gate +// +// Enterprise-managed authorization is shipped behind a host-owned rollout gate. +// The SDK declares the PORT and stays vendor-free: no feature-flag service, no +// analytics client, no network dependency of its own. A host that has one +// injects an implementation through `OAuthServiceDeps`; a host that has none +// (desktop, CLI, local, self-hosted) injects nothing and the profile behaves +// exactly as it did before the gate existed. +// +// Two properties are load-bearing and are pinned by tests +// (`oauth-ema-rollout.test.ts`): +// +// 1. The gate is consulted EXACTLY ONCE per connect, BEFORE discovery, and +// NEVER after the IdP has spoken. A policy denial that could be re-routed +// through a flag check would turn the flag into an escape hatch around the +// very enterprise control this profile exists to enforce. +// +// 2. The gate is a CONNECT-time decision only. It is recorded on the +// connection (`ENTERPRISE_MANAGED_PROVIDER_STATE_KEY`) and the +// credential-refresh path follows the stored state, never the gate. So +// turning the flag off stops NEW enterprise-managed connects and leaves +// every existing managed connection renewing untouched — and no +// third-party network dependency ever enters credential resolution. +// --------------------------------------------------------------------------- + +/** Who is connecting, as far as a rollout gate needs to know. Carries identity + * and catalog identifiers only — never the identity assertion, the client + * secret, or any other credential material. */ +export interface EnterpriseManagedRolloutContext { + /** The acting user, as the host identifies them. Null when the executor is + * bound to an org-level caller with no individual subject. */ + readonly userId: string | null; + /** The organization the executor is bound to, when the host names one. */ + readonly organizationId: string | null; + /** The integration being connected. A spec-derived catalog slug, safe to + * report; the `oauth_client` slug is deliberately NOT here, because a + * user-registered client's slug is user-entered text. */ + readonly integration: IntegrationSlug; +} + +/** Why the gate withheld the enterprise-managed path. Kept structural so a host + * can tell "the operator has not enabled this yet" apart from "we could not + * find out", which are the same user-visible behavior but different + * operational events. */ +export type EnterpriseManagedRolloutWithheldReason = + /** The gate answered, and the answer was no. */ + | "disabled" + /** The gate could not reach a decision (timeout, transport failure, + * unusable answer, missing configuration) and failed closed. */ + | "evaluation-unavailable"; + +/** The gate's verdict on one connect. */ +export type EnterpriseManagedRolloutDecision = + | { readonly kind: "enabled" } + | { readonly kind: "withheld"; readonly reason: EnterpriseManagedRolloutWithheldReason }; + +/** What happened on a connect against a client whose grant is `id_jag`, for a + * host that records rollout analytics. + * + * Every variant carries the context and the gate's decision and NOTHING else + * that could be sensitive: there is no field here that can hold a token, an + * identity assertion, a client secret, or a scope value. */ +export type EnterpriseManagedRolloutEvent = + /** A connect reached the enterprise-managed branch. Fires for BOTH arms of + * the rollout — read `decision` for which — so the funnel below it has a + * denominator. */ + | { + readonly kind: "attempted"; + readonly context: EnterpriseManagedRolloutContext; + readonly decision: EnterpriseManagedRolloutDecision; + } + /** The ID-JAG chain completed and a connection was minted. */ + | { + readonly kind: "connected"; + readonly context: EnterpriseManagedRolloutContext; + readonly decision: EnterpriseManagedRolloutDecision; + } + /** The enterprise IdP declined to authorize this user for this server. */ + | { + readonly kind: "blocked-by-admin"; + readonly context: EnterpriseManagedRolloutContext; + readonly decision: EnterpriseManagedRolloutDecision; + /** The IdP's RFC 6749 §5.2 error code, when it returned one. */ + readonly oauthErrorCode: string | undefined; + }; + +/** The host-owned rollout seam for enterprise-managed authorization. + * + * `decide` is the gate: it MUST NOT fail, because a rollout mechanism that can + * fail a connect is worse than no rollout mechanism. An implementation that + * cannot reach a verdict returns + * `{ kind: "withheld", reason: "evaluation-unavailable" }` — failing closed — + * rather than failing the effect. + * + * `record` is a best-effort observer. The OAuth service runs it with its + * failures and defects discarded, so it can neither fail a connect nor change + * its outcome no matter what the implementation does. */ +export interface EnterpriseManagedRollout { + readonly decide: ( + context: EnterpriseManagedRolloutContext, + ) => Effect.Effect; + readonly record: (event: EnterpriseManagedRolloutEvent) => Effect.Effect; +} + +/** The decision that applies when no host injected a gate: enterprise-managed + * authorization is attempted, which is the behavior every host had before the + * rollout seam existed. */ +export const ENTERPRISE_MANAGED_ROLLOUT_ENABLED: EnterpriseManagedRolloutDecision = { + kind: "enabled", +}; diff --git a/packages/core/sdk/src/oauth-service.ts b/packages/core/sdk/src/oauth-service.ts index 6134b35d4c..7515799794 100644 --- a/packages/core/sdk/src/oauth-service.ts +++ b/packages/core/sdk/src/oauth-service.ts @@ -65,10 +65,15 @@ import { type OAuthAuthorizationServerMetadata, } from "./oauth-discovery"; import { + ENTERPRISE_MANAGED_ROLLOUT_ENABLED, runEnterpriseManagedAuthorization, type EnterpriseManagedConnectionState, type EnterpriseManagedGrant, type EnterpriseManagedMintError, + type EnterpriseManagedRollout, + type EnterpriseManagedRolloutContext, + type EnterpriseManagedRolloutDecision, + type EnterpriseManagedRolloutEvent, } from "./oauth-ema"; import { assertSupportedOAuthEndpointUrl, @@ -212,6 +217,18 @@ export interface OAuthServiceDeps { readonly httpClientLayer?: Layer.Layer; readonly fetch?: typeof globalThis.fetch; readonly endpointUrlPolicy?: OAuthEndpointUrlPolicy; + /** + * Host-owned rollout gate for enterprise-managed authorization (see + * {@link EnterpriseManagedRollout}). Consulted ONCE per `id_jag` connect, + * before discovery, and never again — not after the IdP has ruled, and not on + * the credential-refresh path, which follows the state persisted on the + * connection instead. + * + * OMITTED means enterprise-managed authorization is attempted, which is what + * every host did before this seam existed. Only a host that actually operates + * a flag service supplies one; core takes no dependency on any. + */ + readonly enterpriseManagedRollout?: EnterpriseManagedRollout; /** * The OAuth callback URL (`${webBaseUrl}${mountPrefix}/oauth/callback`) the host * serves and sends to providers on every authorization request + DCR registration. @@ -593,6 +610,41 @@ export const makeOAuthService = (deps: OAuthServiceDeps): OAuthService => { const redirectUri = deps.redirectUri; const discoveryOptions = { endpointUrlPolicy: deps.endpointUrlPolicy }; + // ------------------------------------------------------------------------- + // Enterprise-managed rollout seam. + // + // ROLLOUT SEMANTIC, stated once here because it is the whole reason the gate + // sits where it does: the gate answers "may this connect attempt the + // enterprise-managed path", and nothing else. It runs once, before discovery, + // so a withheld verdict costs no round trip and spends no identity assertion. + // The verdict it produces is then FROZEN onto the connection + // (`ENTERPRISE_MANAGED_PROVIDER_STATE_KEY`), and the credential-refresh path + // reads that state instead of re-asking. Turning the flag off therefore stops + // new enterprise-managed connects and leaves every existing one renewing — no + // stranded connections, no silent downgrade, and no third-party network + // dependency anywhere in credential resolution. + // ------------------------------------------------------------------------- + const rollout = deps.enterpriseManagedRollout; + + /** The gate's verdict, or "enabled" when no host injected a gate. */ + const decideEnterpriseManagedRollout = ( + context: EnterpriseManagedRolloutContext, + ): Effect.Effect => + rollout === undefined + ? Effect.succeed(ENTERPRISE_MANAGED_ROLLOUT_ENABLED) + : rollout.decide(context); + + /** Best-effort rollout observation. Failures AND defects are discarded here, + * so no implementation of `record` can fail a connect or change its outcome; + * keeping it off the critical path is the host's side of the contract. + * Mirrors how `afterCommit` treats `onIntegrationChange`. */ + const recordEnterpriseManagedRollout = ( + event: EnterpriseManagedRolloutEvent, + ): Effect.Effect => + rollout === undefined + ? Effect.void + : rollout.record(event).pipe(Effect.ignoreCause({ log: false })); + const filterAuthorizationCodeScopes = ( client: LoadedOAuthClient, requestedScopes: readonly string[], @@ -1405,87 +1457,137 @@ export const makeOAuthService = (deps: OAuthServiceDeps): OAuthService => { // policy decision and stops here, because offering the interactive flow // instead would let the user route straight around it. if (client.grant === "id_jag") { - const enterprise = input.enterprise; - if (enterprise === undefined) { - return yield* new OAuthStartError({ - message: - "This OAuth app uses enterprise-managed authorization, which requires an enterprise identity provider and an identity assertion on the connect request.", - }); - } - const idpClient = yield* loadClient(enterprise.idpClientOwner, enterprise.idpClient); - if (!idpClient) { - return yield* new OAuthStartError({ - message: `Enterprise identity provider OAuth client not found: ${enterprise.idpClient}`, - }); - } - const metadata = yield* discoverResourceAuthorizationServer(client.resource).pipe( - Effect.mapError( - (cause) => - new OAuthStartError({ - // oxlint-disable-next-line executor/no-unknown-error-message -- boundary: OAuthDiscoveryError carries a typed `message` field - message: `Failed to discover the MCP server's authorization server: ${cause.message}`, - }), - ), - ); - // Resolve the caller's optional assertion type ONCE: the chain sends it - // and the connection persists it, and those two must not be able to - // disagree about what was presented. - const resolvedEnterprise = { - ...enterprise, - subjectTokenType: enterprise.subjectTokenType ?? DEFAULT_SUBJECT_TOKEN_TYPE, + // The rollout gate, consulted ONCE and BEFORE anything else in this + // branch: before the IdP registration is loaded, before discovery, + // before a single request leaves the process. A withheld verdict must + // therefore cost no round trip and spend no identity assertion. + // + // Its answer is read exactly here and never again. Once the IdP has + // ruled, that verdict is final: re-consulting a flag after a denial + // would turn the flag into an escape hatch around the enterprise + // control this whole profile exists to enforce. + const rolloutContext: EnterpriseManagedRolloutContext = { + userId: deps.subject, + organizationId: deps.tenant, + integration: input.integration, }; - const enterpriseGrant = yield* runEnterpriseManagedAuthorization({ - authorizationServerMetadata: metadata, - idp: { - tokenUrl: idpClient.tokenUrl, - clientId: idpClient.clientId, - clientSecret: idpClient.clientSecret, - }, - resourceAuthorizationServer: { - clientId: client.clientId, - clientSecret: client.clientSecret, - }, - subjectToken: resolvedEnterprise.subjectToken, - subjectTokenType: resolvedEnterprise.subjectTokenType, - resource: client.resource, - scopes: requestedScopes, - endpointUrlPolicy: deps.endpointUrlPolicy, - // No `httpClientLayer` here, deliberately: like every other token - // request in this service, the ID-JAG chain runs through oauth4webapi - // on the configured `fetch`, not Effect's HttpClient. Only discovery - // speaks HttpClient. Providing the layer here would claim otherwise. - fetch, - }).pipe( - Effect.map((grant) => ({ supported: true as const, grant })), - // Only the unsupported-profile failure is recoverable; every other - // tag reaches the caller as a start error carrying its own verdict. - Effect.catchTag("EmaGrantProfileUnsupported", () => - Effect.succeed({ supported: false as const }), - ), - Effect.mapError(startErrorFromEnterpriseManaged), - ); - if (enterpriseGrant.supported) { - const connection = yield* mintEnterpriseManagedConnection( - { ...input, name }, - client, - input.clientOwner, - enterpriseGrant.grant, - resolvedEnterprise, - metadata.issuer, - ).pipe( + const rolloutDecision = yield* decideEnterpriseManagedRollout(rolloutContext); + // Recorded for BOTH arms, so the funnel below it has a denominator. + yield* recordEnterpriseManagedRollout({ + kind: "attempted", + context: rolloutContext, + decision: rolloutDecision, + }); + + if (rolloutDecision.kind === "withheld") { + // Withheld takes the SAME exit an authorization server that never + // implemented the profile takes: fall through to the ordinary + // interactive flow below. There is deliberately no second fallback + // path to keep in step with the first. + yield* Effect.annotateCurrentSpan({ + "executor.oauth.enterprise_managed_fallback": true, + "executor.oauth.enterprise_managed_withheld": rolloutDecision.reason, + }); + } else { + const enterprise = input.enterprise; + if (enterprise === undefined) { + return yield* new OAuthStartError({ + message: + "This OAuth app uses enterprise-managed authorization, which requires an enterprise identity provider and an identity assertion on the connect request.", + }); + } + const idpClient = yield* loadClient(enterprise.idpClientOwner, enterprise.idpClient); + if (!idpClient) { + return yield* new OAuthStartError({ + message: `Enterprise identity provider OAuth client not found: ${enterprise.idpClient}`, + }); + } + const metadata = yield* discoverResourceAuthorizationServer(client.resource).pipe( Effect.mapError( (cause) => new OAuthStartError({ - // oxlint-disable-next-line executor/no-unknown-error-message -- boundary: StorageFailure carries a typed `message` field - message: `Failed to mint OAuth connection: ${cause.message}`, + // oxlint-disable-next-line executor/no-unknown-error-message -- boundary: OAuthDiscoveryError carries a typed `message` field + message: `Failed to discover the MCP server's authorization server: ${cause.message}`, }), ), ); - return { status: "connected", connection } as const; + // Resolve the caller's optional assertion type ONCE: the chain sends + // it and the connection persists it, and those two must not be able + // to disagree about what was presented. + const resolvedEnterprise = { + ...enterprise, + subjectTokenType: enterprise.subjectTokenType ?? DEFAULT_SUBJECT_TOKEN_TYPE, + }; + const enterpriseGrant = yield* runEnterpriseManagedAuthorization({ + authorizationServerMetadata: metadata, + idp: { + tokenUrl: idpClient.tokenUrl, + clientId: idpClient.clientId, + clientSecret: idpClient.clientSecret, + }, + resourceAuthorizationServer: { + clientId: client.clientId, + clientSecret: client.clientSecret, + }, + subjectToken: resolvedEnterprise.subjectToken, + subjectTokenType: resolvedEnterprise.subjectTokenType, + resource: client.resource, + scopes: requestedScopes, + endpointUrlPolicy: deps.endpointUrlPolicy, + // No `httpClientLayer` here, deliberately: like every other token + // request in this service, the ID-JAG chain runs through oauth4webapi + // on the configured `fetch`, not Effect's HttpClient. Only discovery + // speaks HttpClient. Providing the layer here would claim otherwise. + fetch, + }).pipe( + Effect.map((grant) => ({ supported: true as const, grant })), + // Only the unsupported-profile failure is recoverable; every other + // tag reaches the caller as a start error carrying its own verdict. + Effect.catchTag("EmaGrantProfileUnsupported", () => + Effect.succeed({ supported: false as const }), + ), + Effect.mapError(startErrorFromEnterpriseManaged), + // OBSERVATION ONLY. This taps the denial on its way out; it does not + // recover it, and no branch below reads the rollout decision again. + Effect.tapError((failure) => + failure.blockedByAdmin === true + ? recordEnterpriseManagedRollout({ + kind: "blocked-by-admin", + context: rolloutContext, + decision: rolloutDecision, + oauthErrorCode: failure.oauthErrorCode, + }) + : Effect.void, + ), + ); + if (enterpriseGrant.supported) { + const connection = yield* mintEnterpriseManagedConnection( + { ...input, name }, + client, + input.clientOwner, + enterpriseGrant.grant, + resolvedEnterprise, + metadata.issuer, + ).pipe( + Effect.mapError( + (cause) => + new OAuthStartError({ + // oxlint-disable-next-line executor/no-unknown-error-message -- boundary: StorageFailure carries a typed `message` field + message: `Failed to mint OAuth connection: ${cause.message}`, + }), + ), + ); + yield* recordEnterpriseManagedRollout({ + kind: "connected", + context: rolloutContext, + decision: rolloutDecision, + }); + return { status: "connected", connection } as const; + } + yield* Effect.annotateCurrentSpan({ + "executor.oauth.enterprise_managed_fallback": true, + }); } - yield* Effect.annotateCurrentSpan({ - "executor.oauth.enterprise_managed_fallback": true, - }); } // authorization_code requires our callback to receive the code — fail diff --git a/packages/core/sdk/src/test-config.ts b/packages/core/sdk/src/test-config.ts index 4cac77d711..f452b94db4 100644 --- a/packages/core/sdk/src/test-config.ts +++ b/packages/core/sdk/src/test-config.ts @@ -124,6 +124,7 @@ export type TestConfigOptions["onIntegrationChange"]; readonly firstPartyOAuthClients?: ExecutorConfig["firstPartyOAuthClients"]; + readonly enterpriseManagedRollout?: ExecutorConfig["enterpriseManagedRollout"]; }; export const makeTestConfig = ( @@ -165,6 +166,7 @@ export const makeTestConfig = Date: Tue, 25 Aug 2026 12:47:19 -0700 Subject: [PATCH 2/4] Carry the rollout gate through to the cloud executor --- apps/cloud/src/engine/execution-stack.ts | 6 ++++++ packages/core/api/src/server/scoped-executor.ts | 10 ++++++++++ 2 files changed, 16 insertions(+) diff --git a/apps/cloud/src/engine/execution-stack.ts b/apps/cloud/src/engine/execution-stack.ts index 013774bc81..cc9afb38d1 100644 --- a/apps/cloud/src/engine/execution-stack.ts +++ b/apps/cloud/src/engine/execution-stack.ts @@ -52,6 +52,7 @@ import { } from "@executor-js/sdk"; import executorConfig from "../../executor.config"; +import { cloudEnterpriseManagedRollout } from "../analytics/ema-rollout"; import { DbService } from "../db/db"; import { cloudDbProviderLayer } from "../db/fuma"; @@ -192,6 +193,11 @@ export const CloudHostConfig: Layer.Layer = Layer.sync(HostConfig, ( // user-selectable provider surface. exposeCredentialProviders: false, firstPartyOAuthClients: cloudFirstPartyOAuthClients(), + // Enterprise-managed authorization ships behind a PostHog flag. Cloud is the + // one host with a flag service, so cloud is the one host that installs a + // gate; everywhere else the seam stays empty and the profile is attempted as + // before. Gating happens at connect only — see the SDK contract. + enterpriseManagedRollout: cloudEnterpriseManagedRollout(), })); export const CloudCodeExecutorProvider: Layer.Layer = Layer.sync( diff --git a/packages/core/api/src/server/scoped-executor.ts b/packages/core/api/src/server/scoped-executor.ts index 6c2b342311..9e6013d230 100644 --- a/packages/core/api/src/server/scoped-executor.ts +++ b/packages/core/api/src/server/scoped-executor.ts @@ -106,6 +106,15 @@ export interface HostConfigShape { * ship none simply omit it. */ readonly firstPartyOAuthClients?: readonly FirstPartyOAuthClientConfig[]; + /** + * Forwarded to `ExecutorConfig.enterpriseManagedRollout`: the host's rollout + * gate for enterprise-managed authorization (the MCP EMA profile). Declared + * here — not per-request — because it is a deployment-wide capability; the + * per-connect identity it needs is supplied by the SDK at the call site. + * Hosts that operate no feature-flag service omit it, and the profile is + * attempted as it was before the gate existed. + */ + readonly enterpriseManagedRollout?: ExecutorConfig["enterpriseManagedRollout"]; } export class HostConfig extends Context.Service()( @@ -291,6 +300,7 @@ export const makeScopedExecutor = < redirectUri, oauthCallbackStateOrgSlug: orgSlug, firstPartyOAuthClients: config.firstPartyOAuthClients, + enterpriseManagedRollout: config.enterpriseManagedRollout, coreTools: { webBaseUrl, orgSlug, From 33e267290d617857d6288d59f556941b3a8b2a77 Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Tue, 25 Aug 2026 12:47:19 -0700 Subject: [PATCH 3/4] Evaluate the enterprise-managed rollout flag in PostHog --- apps/cloud/src/analytics/ema-rollout.test.ts | 394 +++++++++++++++++++ apps/cloud/src/analytics/ema-rollout.ts | 255 ++++++++++++ apps/cloud/src/edge/passthrough.ts | 5 +- 3 files changed, 653 insertions(+), 1 deletion(-) create mode 100644 apps/cloud/src/analytics/ema-rollout.test.ts create mode 100644 apps/cloud/src/analytics/ema-rollout.ts diff --git a/apps/cloud/src/analytics/ema-rollout.test.ts b/apps/cloud/src/analytics/ema-rollout.test.ts new file mode 100644 index 0000000000..c62321b050 --- /dev/null +++ b/apps/cloud/src/analytics/ema-rollout.test.ts @@ -0,0 +1,394 @@ +// --------------------------------------------------------------------------- +// Cloud's PostHog-backed enterprise-managed rollout gate. +// +// The gate is exercised through its production seam: the real +// `makePostHogEnterpriseManagedRollout` with an injected `fetch` and an +// injected `waitUntil`, exactly as `auth/jwks-cache.node.test.ts` drives the +// JWKS client. Nothing is module-mocked, so the request these tests read is the +// request PostHog would receive. +// --------------------------------------------------------------------------- + +import { describe, expect, it } from "@effect/vitest"; +import { Effect } from "effect"; + +import type { EnterpriseManagedRolloutContext } from "@executor-js/sdk"; + +import { + ENTERPRISE_MANAGED_AUTH_FLAG_KEY, + cloudEnterpriseManagedRollout, + makePostHogEnterpriseManagedRollout, +} from "./ema-rollout"; + +const HOST = "https://us.i.posthog.com"; +const PROJECT_KEY = "phc_test_project_key"; + +const CONTEXT: EnterpriseManagedRolloutContext = { + userId: "user_01ABC", + organizationId: "org_01XYZ", + integration: "slack" as EnterpriseManagedRolloutContext["integration"], +}; + +interface CapturedRequest { + readonly url: string; + readonly body: Record; + /** The bound the gate put on this call, when it set one. */ + readonly signal: AbortSignal | null; +} + +interface FetchHarness { + readonly fetch: typeof globalThis.fetch; + readonly requests: () => readonly CapturedRequest[]; + /** Resolve every request the gate detached, so an assertion cannot race it. */ + readonly settle: () => Promise; + readonly waitUntil: (work: Promise) => void; +} + +/** + * Stands in for PostHog. `respond` decides what each call returns; throwing + * from it models a transport failure, and an aborted signal models the timeout. + */ +const makeFetchHarness = ( + respond: (request: CapturedRequest) => Response | Promise, +): FetchHarness => { + const requests: CapturedRequest[] = []; + const detached: Promise[] = []; + return { + requests: () => requests, + settle: async () => { + await Promise.all(detached); + }, + waitUntil: (work) => { + detached.push(work); + }, + fetch: async (input, init) => { + const raw = typeof init?.body === "string" ? init.body : "{}"; + const captured: CapturedRequest = { + url: String(input), + // oxlint-disable-next-line executor/no-json-parse -- boundary: test fixture reads back the exact JSON body the gate serialized for PostHog + body: JSON.parse(raw) as Record, + signal: init?.signal instanceof AbortSignal ? init.signal : null, + }; + requests.push(captured); + return respond(captured); + }, + }; +}; + +const jsonResponse = (body: unknown, status = 200): Response => + new Response(JSON.stringify(body), { + status, + headers: { "content-type": "application/json" }, + }); + +const flagsBody = (enabled: boolean) => ({ + flags: { + [ENTERPRISE_MANAGED_AUTH_FLAG_KEY]: { + key: ENTERPRISE_MANAGED_AUTH_FLAG_KEY, + enabled, + reason: { code: "condition_match" }, + }, + }, + errorsWhileComputingFlags: false, +}); + +const gate = (harness: FetchHarness, timeoutMs?: number) => + makePostHogEnterpriseManagedRollout({ + projectKey: PROJECT_KEY, + host: HOST, + fetch: harness.fetch, + waitUntil: harness.waitUntil, + ...(timeoutMs === undefined ? {} : { timeoutMs }), + }); + +describe("flag evaluation", () => { + it.effect("keys the rollout on the user and carries the org as group context", () => + Effect.gen(function* () { + const harness = makeFetchHarness(() => jsonResponse(flagsBody(true))); + + const decision = yield* gate(harness).decide(CONTEXT); + + expect(decision).toEqual({ kind: "enabled" }); + const [request] = harness.requests(); + expect(request?.url).toBe(`${HOST}/flags?v=2`); + expect( + request?.body.distinct_id, + "the rollout unit is the user, matching posthog.identify(user.id) in the browser", + ).toBe(CONTEXT.userId); + expect( + request?.body.groups, + "the org rides along so group targeting stays available without changing the rollout unit", + ).toEqual({ organization: CONTEXT.organizationId }); + expect(request?.body.api_key).toBe(PROJECT_KEY); + expect(request?.signal, "the evaluation is always bounded").not.toBeNull(); + }), + ); + + it.effect("omits group context when the host named no organization", () => + Effect.gen(function* () { + const harness = makeFetchHarness(() => jsonResponse(flagsBody(true))); + + yield* gate(harness).decide({ ...CONTEXT, organizationId: null }); + + expect(harness.requests()[0]?.body.groups).toBeUndefined(); + }), + ); + + it.effect("withholds when the flag is off for this user", () => + Effect.gen(function* () { + const harness = makeFetchHarness(() => jsonResponse(flagsBody(false))); + + expect(yield* gate(harness).decide(CONTEXT)).toEqual({ + kind: "withheld", + reason: "disabled", + }); + }), + ); + + it.effect("withholds when PostHog returned no verdict for the flag at all", () => + Effect.gen(function* () { + const harness = makeFetchHarness(() => + // What a quota-limited project answers. + jsonResponse({ + flags: {}, + errorsWhileComputingFlags: false, + quotaLimited: ["feature_flags"], + }), + ); + + expect(yield* gate(harness).decide(CONTEXT)).toEqual({ + kind: "withheld", + reason: "disabled", + }); + }), + ); +}); + +describe("failing closed", () => { + it.effect("withholds on a non-2xx answer", () => + Effect.gen(function* () { + const harness = makeFetchHarness(() => new Response("nope", { status: 503 })); + + expect(yield* gate(harness).decide(CONTEXT)).toEqual({ + kind: "withheld", + reason: "evaluation-unavailable", + }); + }), + ); + + it.effect("withholds on a transport failure", () => + Effect.gen(function* () { + const harness = makeFetchHarness(() => { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: models `fetch` rejecting with a TypeError, which is exactly how a Worker sees a dead upstream + throw new TypeError("network error"); + }); + + expect(yield* gate(harness).decide(CONTEXT)).toEqual({ + kind: "withheld", + reason: "evaluation-unavailable", + }); + }), + ); + + it.effect("withholds when the evaluation times out", () => + Effect.gen(function* () { + // Never answers. Only the gate's own AbortSignal ends this call, so the + // test fails by hanging if the gate ever stops bounding the evaluation. + const harness = makeFetchHarness( + (request) => + new Promise((_resolve, reject) => { + request.signal?.addEventListener("abort", () => + // oxlint-disable-next-line executor/no-promise-reject, executor/no-error-constructor -- boundary: fetch-compatible fixture mirrors the platform's abort rejection semantics + reject(new DOMException("aborted", "AbortError")), + ); + }), + ); + + expect( + yield* gate(harness, 1).decide(CONTEXT), + "a slow flag service must degrade the rollout, never hold up a connect", + ).toEqual({ kind: "withheld", reason: "evaluation-unavailable" }); + }), + ); + + it.effect("withholds on a body that is not a flags response", () => + Effect.gen(function* () { + const harness = makeFetchHarness(() => jsonResponse({ flags: "not-an-object" })); + + expect(yield* gate(harness).decide(CONTEXT)).toEqual({ + kind: "withheld", + reason: "evaluation-unavailable", + }); + }), + ); + + it.effect("withholds when there is no acting user to roll out by", () => + Effect.gen(function* () { + const harness = makeFetchHarness(() => jsonResponse(flagsBody(true))); + + expect(yield* gate(harness).decide({ ...CONTEXT, userId: null })).toEqual({ + kind: "withheld", + reason: "evaluation-unavailable", + }); + expect(harness.requests().length, "and asks PostHog nothing").toBe(0); + }), + ); + + it.effect("withholds when the deployment has no PostHog configuration", () => + Effect.gen(function* () { + // The cloud test env carries no VITE_PUBLIC_POSTHOG_KEY, so this builds + // the misconfigured-deployment gate — which is deliberately NOT the same + // as a host that injects no gate at all. + expect(yield* cloudEnterpriseManagedRollout().decide(CONTEXT)).toEqual({ + kind: "withheld", + reason: "evaluation-unavailable", + }); + }), + ); +}); + +describe("rollout events", () => { + it.effect("captures the connect outcome with the flag decision attached", () => + Effect.gen(function* () { + const harness = makeFetchHarness(() => jsonResponse({ status: 1 })); + const rollout = gate(harness); + + yield* rollout.record({ + kind: "connected", + context: CONTEXT, + decision: { kind: "enabled" }, + }); + yield* Effect.promise(() => harness.settle()); + + const [request] = harness.requests(); + expect(request?.url).toBe(`${HOST}/i/v0/e/`); + expect(request?.body.event).toBe("ema_connect_connected"); + expect(request?.body.distinct_id).toBe(CONTEXT.userId); + const properties = request?.body.properties as Record; + expect(properties.ema_flag_enabled).toBe(true); + expect(properties[`$feature/${ENTERPRISE_MANAGED_AUTH_FLAG_KEY}`]).toBe(true); + expect(properties.$groups).toEqual({ + organization: CONTEXT.organizationId, + }); + expect(properties.integration_slug).toBe("slack"); + }), + ); + + it.effect("carries the withheld reason on the attempt that never ran", () => + Effect.gen(function* () { + const harness = makeFetchHarness(() => jsonResponse({ status: 1 })); + + yield* gate(harness).record({ + kind: "attempted", + context: CONTEXT, + decision: { kind: "withheld", reason: "evaluation-unavailable" }, + }); + yield* Effect.promise(() => harness.settle()); + + const properties = harness.requests()[0]?.body.properties as Record; + expect(harness.requests()[0]?.body.event).toBe("ema_connect_attempted"); + expect(properties.ema_flag_enabled).toBe(false); + expect(properties.ema_flag_withheld_reason).toBe("evaluation-unavailable"); + }), + ); + + it.effect("carries the identity provider's own error code on a blocked connect", () => + Effect.gen(function* () { + const harness = makeFetchHarness(() => jsonResponse({ status: 1 })); + + yield* gate(harness).record({ + kind: "blocked-by-admin", + context: CONTEXT, + decision: { kind: "enabled" }, + oauthErrorCode: "unauthorized_client", + }); + yield* Effect.promise(() => harness.settle()); + + const [request] = harness.requests(); + expect(request?.body.event).toBe("ema_connect_blocked_by_admin"); + const properties = request?.body.properties as Record | undefined; + expect(properties?.oauth_error_code).toBe("unauthorized_client"); + }), + ); + + it.effect("sends nothing beyond identity, the integration and the flag decision", () => + Effect.gen(function* () { + const harness = makeFetchHarness(() => jsonResponse({ status: 1 })); + + yield* gate(harness).record({ + kind: "blocked-by-admin", + context: CONTEXT, + decision: { kind: "enabled" }, + oauthErrorCode: "access_denied", + }); + yield* Effect.promise(() => harness.settle()); + + const properties = harness.requests()[0]?.body.properties as Record; + expect( + Object.keys(properties).sort(), + "a closed property set is what keeps a token or an assertion from ever being added by accident", + ).toEqual( + [ + "$feature/mcp-enterprise-managed-auth", + "$groups", + "ema_flag_enabled", + "integration_slug", + "oauth_error_code", + ].sort(), + ); + }), + ); + + it.effect("hands the capture to the platform instead of awaiting it", () => + Effect.gen(function* () { + let resolveCapture: (() => void) | undefined; + const harness = makeFetchHarness( + () => + new Promise((resolve) => { + resolveCapture = () => resolve(jsonResponse({ status: 1 })); + }), + ); + + // Returns while the capture is still in flight: an analytics call can + // never be on the critical path of a connect. + yield* gate(harness).record({ + kind: "attempted", + context: CONTEXT, + decision: { kind: "enabled" }, + }); + + expect(harness.requests().length).toBe(1); + resolveCapture?.(); + yield* Effect.promise(() => harness.settle()); + }), + ); + + it.effect("survives a capture that fails outright", () => + Effect.gen(function* () { + const harness = makeFetchHarness(() => { + // oxlint-disable-next-line executor/no-try-catch-or-throw, executor/no-error-constructor -- boundary: models the ingest endpoint being unreachable + throw new TypeError("network error"); + }); + + yield* gate(harness).record({ + kind: "connected", + context: CONTEXT, + decision: { kind: "enabled" }, + }); + yield* Effect.promise(() => harness.settle()); + }), + ); + + it.effect("records nothing when there is no person to attach the event to", () => + Effect.gen(function* () { + const harness = makeFetchHarness(() => jsonResponse({ status: 1 })); + + yield* gate(harness).record({ + kind: "attempted", + context: { ...CONTEXT, userId: null }, + decision: { kind: "withheld", reason: "evaluation-unavailable" }, + }); + + expect(harness.requests().length).toBe(0); + }), + ); +}); diff --git a/apps/cloud/src/analytics/ema-rollout.ts b/apps/cloud/src/analytics/ema-rollout.ts new file mode 100644 index 0000000000..88d10e8e94 --- /dev/null +++ b/apps/cloud/src/analytics/ema-rollout.ts @@ -0,0 +1,255 @@ +// --------------------------------------------------------------------------- +// Cloud's rollout gate for MCP Enterprise-Managed Authorization. +// +// Implements the vendor-free `EnterpriseManagedRollout` port from +// `@executor-js/sdk` against PostHog, which is the only flag/analytics service +// this deployment operates. Two deliberate shapes: +// +// 1. NO SDK. This is a hand-written `fetch` against two documented PostHog +// endpoints, not `posthog-node`. Cloud already ate an unexplained 3-5s +// page-load regression from adding one dependency to the worker bundle and +// the mechanism was never identified (see the notes on the MCP SDK v2 +// revert). Until that is understood, a new runtime dependency in this bundle +// is a cost we are not willing to pay for a boolean, and the whole client we +// would be importing is two POSTs wide. +// +// 2. FAIL CLOSED, ALWAYS. Every way of not getting an answer — timeout, +// transport failure, non-2xx, a body we cannot parse, no PostHog key +// configured, no acting user to key the rollout on — withholds the +// enterprise-managed path and falls back to ordinary interactive OAuth. A +// PostHog outage therefore degrades the rollout and never fails a connect, +// and (because the SDK freezes the verdict onto the connection) never +// touches an enterprise-managed connection that already exists. +// +// The rollout UNIT is the user: `distinct_id` is the acting user's id, the same +// id `posthog.identify(user.id, …)` uses in the browser, so a percentage +// rollout here means the same population it means everywhere else in the +// project. The organization rides along as group context so org-level targeting +// stays available in the PostHog UI without changing that unit. +// --------------------------------------------------------------------------- + +import { env } from "cloudflare:workers"; +import { Effect, Schema } from "effect"; + +import type { + EnterpriseManagedRollout, + EnterpriseManagedRolloutContext, + EnterpriseManagedRolloutDecision, + EnterpriseManagedRolloutEvent, +} from "@executor-js/sdk"; + +import { POSTHOG_INGEST_HOST } from "../edge/passthrough"; + +/** The flag that gates enterprise-managed authorization. */ +export const ENTERPRISE_MANAGED_AUTH_FLAG_KEY = "mcp-enterprise-managed-auth"; + +/** PostHog group type for organizations — the same one + * `posthog.group("organization", …)` establishes in the browser. */ +const ORGANIZATION_GROUP_TYPE = "organization"; + +/** + * How long a flag evaluation may take before the gate gives up and withholds. + * Short on purpose: this call sits in front of an interactive connect, and the + * safe answer is already known, so waiting is strictly worse than answering. + */ +export const DEFAULT_FLAG_EVALUATION_TIMEOUT_MS = 1_000; + +/** Wire names of the rollout events, in the product's `object_verb` style. */ +const EVENT_NAMES = { + attempted: "ema_connect_attempted", + connected: "ema_connect_connected", + "blocked-by-admin": "ema_connect_blocked_by_admin", +} as const satisfies Record; + +// --------------------------------------------------------------------------- +// Boundary parsing +// --------------------------------------------------------------------------- + +/** The slice of PostHog's `POST /flags?v=2` response this gate reads. Unknown + * keys are ignored, so PostHog adding fields cannot make the gate fail closed; + * a body that is not this shape at all can, which is the intent. */ +const FlagsResponse = Schema.Struct({ + flags: Schema.Record(Schema.String, Schema.Struct({ enabled: Schema.Boolean })), +}); + +const decodeFlagsResponse = Schema.decodeUnknownEffect(FlagsResponse); + +// --------------------------------------------------------------------------- +// Detached work +// --------------------------------------------------------------------------- + +/** + * Hand a request to the platform and stop caring about it. + * + * `waitUntil` is the correct owner when the caller has an `ExecutionContext`; + * the executor composition seam this gate is installed through does not receive + * one, so the default owner attaches a terminal rejection handler and lets the + * in-flight request ride out the rest of the response. Either way the promise + * is owned by something that cannot report back, which is what makes an event + * structurally unable to fail or delay a connect. + */ +const detach = ( + work: Promise, + waitUntil: ((work: Promise) => void) | undefined, +): void => { + const settled = work.then( + () => undefined, + () => undefined, + ); + if (waitUntil) waitUntil(settled); +}; + +// --------------------------------------------------------------------------- +// The gate +// --------------------------------------------------------------------------- + +export interface PostHogRolloutConfig { + /** Public project key (`phc_…`). Public by design — it is already shipped to + * every browser — so this is a var, not a secret. */ + readonly projectKey: string; + /** PostHog API origin (no trailing slash), e.g. `https://us.i.posthog.com`. */ + readonly host: string; + readonly fetch: typeof globalThis.fetch; + readonly timeoutMs?: number; + /** Platform post-response hook, when the caller has one. */ + readonly waitUntil?: (work: Promise) => void; +} + +const withheld = ( + reason: "disabled" | "evaluation-unavailable", +): EnterpriseManagedRolloutDecision => ({ kind: "withheld", reason }); + +const ENABLED: EnterpriseManagedRolloutDecision = { kind: "enabled" }; + +/** Group context for a connect whose organization the host named. */ +const groupsFor = (context: EnterpriseManagedRolloutContext): Record | undefined => + context.organizationId === null + ? undefined + : { [ORGANIZATION_GROUP_TYPE]: context.organizationId }; + +/** + * Build cloud's enterprise-managed rollout gate over an explicit PostHog + * configuration. Everything environmental is a parameter so the composition + * root parses it once and tests drive the real code path with an injected + * `fetch`. + */ +export const makePostHogEnterpriseManagedRollout = ( + config: PostHogRolloutConfig, +): EnterpriseManagedRollout => { + const timeoutMs = config.timeoutMs ?? DEFAULT_FLAG_EVALUATION_TIMEOUT_MS; + + const evaluate = Effect.fn("Cloud.EnterpriseManagedRollout.decide")(function* ( + context: EnterpriseManagedRolloutContext, + ) { + // The rollout unit is the user. With no acting user there is no + // `distinct_id` to key it on, so there is no answer to be had — withhold + // rather than invent a rollout bucket. + const distinctId = context.userId; + if (distinctId === null) return withheld("evaluation-unavailable"); + + const groups = groupsFor(context); + const response = yield* Effect.tryPromise(() => + config.fetch(`${config.host}/flags?v=2`, { + method: "POST", + headers: { + "content-type": "application/json", + // Declares this a server-side evaluation, which is what selects + // server-runtime flags in PostHog's runtime detection. + "user-agent": "executor-cloud", + }, + body: JSON.stringify({ + api_key: config.projectKey, + distinct_id: distinctId, + ...(groups === undefined ? {} : { groups }), + }), + signal: AbortSignal.timeout(timeoutMs), + }), + ); + if (!response.ok) return withheld("evaluation-unavailable"); + + const body = yield* Effect.tryPromise(() => response.json() as Promise); + const decoded = yield* decodeFlagsResponse(body); + const flag = decoded.flags[ENTERPRISE_MANAGED_AUTH_FLAG_KEY]; + // A flag PostHog did not return is a flag that is not on for this user + // (it may not exist yet, or the project may be quota-limited). That is a + // real "no", not a failure to answer. + if (flag === undefined || !flag.enabled) return withheld("disabled"); + return ENABLED; + }); + + return { + decide: (context) => + evaluate(context).pipe( + // Timeout, transport failure and an unparseable body all land here and + // all mean the same thing: no answer, so no enterprise-managed attempt. + // `catchCause` rather than `catchAll` because a defect in this adapter + // must not become a failed connect either. + Effect.catchCause(() => Effect.succeed(withheld("evaluation-unavailable"))), + ), + record: (event) => + Effect.sync(() => { + const distinctId = event.context.userId; + // Same reason `decide` withholds: an event with no person to attach to + // would only pollute the project with anonymous rows. + if (distinctId === null) return; + const groups = groupsFor(event.context); + const decision = event.decision; + detach( + config.fetch(`${config.host}/i/v0/e/`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ + api_key: config.projectKey, + event: EVENT_NAMES[event.kind], + distinct_id: distinctId, + properties: { + // Product metadata only. There is no field on + // `EnterpriseManagedRolloutEvent` that can carry a token, an + // identity assertion, a client secret or a scope, and none is + // synthesized here. + integration_slug: String(event.context.integration), + ema_flag_enabled: decision.kind === "enabled", + ...(decision.kind === "withheld" + ? { ema_flag_withheld_reason: decision.reason } + : {}), + // PostHog's own convention, so these events can be filtered by + // flag value in the UI alongside every other flag. + [`$feature/${ENTERPRISE_MANAGED_AUTH_FLAG_KEY}`]: decision.kind === "enabled", + ...(event.kind === "blocked-by-admin" && event.oauthErrorCode !== undefined + ? { oauth_error_code: event.oauthErrorCode } + : {}), + ...(groups === undefined ? {} : { $groups: groups }), + }, + }), + }), + config.waitUntil, + ); + }), + }; +}; + +/** A gate that answers "no" to everything, for a deployment that has no PostHog + * configuration to evaluate against. Cloud is SUPPOSED to have one, so a + * missing key is a misconfiguration, and failing closed is the same choice the + * outage paths make. This is NOT the same as injecting no gate at all, which + * is how the hosts with no flag service (local, desktop, CLI, self-host) keep + * their existing behavior. */ +export const unavailableEnterpriseManagedRollout: EnterpriseManagedRollout = { + decide: () => Effect.succeed(withheld("evaluation-unavailable")), + record: () => Effect.void, +}; + +/** + * The gate as the cloud worker composes it: the public project key already + * shipped in `wrangler.jsonc`, and the same PostHog ingest origin the + * adblock-dodging passthrough proxies to. + */ +export const cloudEnterpriseManagedRollout = (): EnterpriseManagedRollout => { + const projectKey = env.VITE_PUBLIC_POSTHOG_KEY; + if (!projectKey) return unavailableEnterpriseManagedRollout; + return makePostHogEnterpriseManagedRollout({ + projectKey, + host: env.VITE_PUBLIC_POSTHOG_HOST ?? `https://${POSTHOG_INGEST_HOST}`, + fetch: (input, init) => globalThis.fetch(input, init), + }); +}; diff --git a/apps/cloud/src/edge/passthrough.ts b/apps/cloud/src/edge/passthrough.ts index 12b6229d44..76049436c6 100644 --- a/apps/cloud/src/edge/passthrough.ts +++ b/apps/cloud/src/edge/passthrough.ts @@ -28,7 +28,10 @@ import { SpanKind, SpanStatusCode, trace } from "@opentelemetry/api"; const DOCS_UPSTREAM_HOST = "executor.mintlify.dev"; -const POSTHOG_INGEST_HOST = "us.i.posthog.com"; +/** PostHog's US ingest origin. Exported because the server-side feature-flag + * gate (`../analytics/ema-rollout`) calls the same origin directly, and the + * two must not be able to drift onto different PostHog regions. */ +export const POSTHOG_INGEST_HOST = "us.i.posthog.com"; const POSTHOG_ASSETS_HOST = "us-assets.i.posthog.com"; export const POSTHOG_PROXY_PATH = `/api/${( From 22ed12d4f18ccde3a3eb589e70b06d79bce478bf Mon Sep 17 00:00:00 2001 From: Rhys Sullivan <39114868+RhysSullivan@users.noreply.github.com> Date: Wed, 26 Aug 2026 15:08:15 -0700 Subject: [PATCH 4/4] Re-run CI after rebase onto main