Selfhost patches

apps/host-cloudflare/src/worker.ts

@@ -1,3 +1,5 @@
+import { decodeOAuthCallbackState } from "@executor-js/sdk/shared";
+
 import { makeCloudflareApp } from "./app";
 import type { CloudflareEnv } from "./config";
 
@@ -27,12 +29,24 @@ const resolveHandler = (env: CloudflareEnv) => {
   return handlerPromise;
 };
 
+const OAUTH_CALLBACK_PATH = "/api/oauth/callback";
+
+const normalizeOAuthCallbackState = (request: Request): Request => {
+  if (request.method !== "GET" && request.method !== "HEAD") return request;
+  const url = new URL(request.url);
+  if (url.pathname !== OAUTH_CALLBACK_PATH) return request;
+  const callbackState = decodeOAuthCallbackState(url.searchParams.get("state"));
+  if (callbackState === null) return request;
+  url.searchParams.set("state", callbackState.state);
+  return new Request(url, request);
+};
+
 export default {
   fetch: async (request: Request, env: CloudflareEnv, ctx: ExecutionContext): Promise<Response> => {
     const serve = await resolveHandler(env);
     if (new URL(request.url).pathname === "/mcp") {
       return serve.mcp(request, env, ctx);
     }
-    return serve.app(request);
+    return serve.app(normalizeOAuthCallbackState(request));
   },
 };

apps/host-cloudflare/wrangler.jsonc

@@ -22,7 +22,7 @@
     {
       "binding": "DB",
       "database_name": "executor",
-      "database_id": "ae748ca1-032c-4427-a1a0-fe39db77d1a9",
+      "database_id": "029d8b51-2a1b-43ec-961b-991a3fea0a0c",
     },
   ],
   // Plugin blob seam backend: multi-MB values (resolved OpenAPI specs,
@@ -55,13 +55,15 @@
   // (the at-rest secret-encryption key) is a SECRET — set it with
   // `wrangler secret put EXECUTOR_SECRET_KEY`, never in vars.
   "vars": {
-    "ACCESS_TEAM_DOMAIN": "your-team.cloudflareaccess.com",
-    "ACCESS_AUD": "",
+    "ACCESS_TEAM_DOMAIN": "mateusz-7ac.cloudflareaccess.com",
+    "ACCESS_AUD": "44e8b9acc8e46bae7e4671d7bb6af41d56a042864cf14dea70e33b85b6f99922",
     "ACCESS_NAME_CLAIM": "name",
     "ACCESS_GROUPS_CLAIM": "groups",
-    "ADMIN_EMAILS": "",
+    "ADMIN_EMAILS": "mateusz@synthetic.ai",
     "SELF_HOSTED_ORG_ID": "default",
-    "SELF_HOSTED_ORG_NAME": "Default",
+    "SELF_HOSTED_ORG_NAME": "Synthetic",
+    "VITE_PUBLIC_SITE_URL": "https://executor-cloudflare.mateusz-7ac.workers.dev",
+    "SELF_HOSTED_ORG_SLUG": "synthetic",
     // VITE_PUBLIC_SITE_URL is intentionally unset: with no static URL the worker
     // derives the web base URL from each request's origin (RequestWebOrigin), so
     // secret/OAuth handoff links match whatever host the user actually reached.

packages/core/sdk/src/core-schema.ts

@@ -33,7 +33,7 @@ export const dateColumn = (name: string) => column(name, "timestamp");
 // The policy callback hands us a `ConditionBuilder` typed to the specific table's
 // columns; it isn't assignable to the generic `Record<string, AnyColumn>` builder
 // (column-name positions are contravariant), so accept it loosely and re-narrow.
-const ownerVisibility = (builder: unknown, context: ExecutorOwnerPolicyContext) =>
+const ownerVisibility = (builder: unknown, context: ExecutorOwnerPolicyContext | undefined) =>
   ownerVisibilityCondition(builder as AnyConditionBuilder, context) as Condition | boolean;
 
 /** A truly global table (the blob store). Isolation is carried in the row's

packages/core/sdk/src/executor.ts

@@ -106,7 +106,11 @@ import {
   type ToolPolicy,
   type UpdateToolPolicyInput,
 } from "./policies";
-import type { CredentialProvider, ProviderEntry } from "./provider";
+import type {
+  CredentialProvider,
+  CredentialProviderScope,
+  ProviderEntry,
+} from "./provider";
 import type {
   AnyPlugin,
   Elicit,
@@ -1371,6 +1375,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
     const ownerContext: ExecutorOwnerPolicyContext = { tenant, subject };
     const rootDb = withQueryContext(rootDbUntyped, ownerContext);
     const fuma = makeFumaClient(rootDb);
+    const sessionFuma = makeFumaClient(rootDbUntyped);
     const core = makeCoreDb(fuma);
     const blobs = config.blobs ?? makeFumaBlobStore(fuma);
     const transaction = <A, E>(effect: Effect.Effect<A, E>) => fuma.transaction(effect);
@@ -1492,6 +1497,11 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
     const connectionKey = (row: ConnectionRow): string =>
       `${row.owner}:${row.subject}:${row.integration}:${row.name}`;
 
+    const connectionCredentialScope = (row: ConnectionRow): CredentialProviderScope => ({
+      owner: row.owner as Owner,
+      subject: String(row.subject),
+    });
+
     const loadOAuthClientRow = (
       owner: Owner,
       slug: string,
@@ -1526,7 +1536,10 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
 
         // The secret is stored in the provider (a vault item id), not inline.
         const clientSecret = clientRow.client_secret_item_id
-          ? ((yield* provider.get(ProviderItemId.make(String(clientRow.client_secret_item_id)))) ??
+          ? ((yield* provider.get(ProviderItemId.make(String(clientRow.client_secret_item_id)), {
+              owner: clientRow.owner as Owner,
+              subject: String(clientRow.subject),
+            })) ??
             "")
           : "";
         // Re-request the scopes this connection was GRANTED (RFC 6749 §6: a
@@ -1576,7 +1589,10 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
                 if (!row.refresh_item_id) {
                   return yield* reauth("No refresh token is stored for this connection.");
                 }
-                const refreshToken = yield* provider.get(ProviderItemId.make(row.refresh_item_id));
+                const refreshToken = yield* provider.get(
+                  ProviderItemId.make(row.refresh_item_id),
+                  connectionCredentialScope(row),
+                );
                 if (!refreshToken) {
                   return yield* reauth("Stored refresh token could not be resolved.");
                 }
@@ -1629,9 +1645,18 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
           const tokenItemId =
             connectionItemIds(row)[PRIMARY_INPUT_VARIABLE] ??
             `connection:${row.owner}:${row.integration}:${row.name}:${PRIMARY_INPUT_VARIABLE}`;
-          yield* provider.set(ProviderItemId.make(tokenItemId), token.access_token);
+          const credentialScope = connectionCredentialScope(row);
+          yield* provider.set(
+            ProviderItemId.make(tokenItemId),
+            token.access_token,
+            credentialScope,
+          );
           if (token.refresh_token && row.refresh_item_id) {
-            yield* provider.set(ProviderItemId.make(row.refresh_item_id), token.refresh_token);
+            yield* provider.set(
+              ProviderItemId.make(row.refresh_item_id),
+              token.refresh_token,
+              credentialScope,
+            );
           }
         }
 
@@ -1658,6 +1683,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
     const refreshConnectionToken = (
       row: ConnectionRow,
       provider: CredentialProvider,
+      options?: { readonly force?: boolean },
     ): Effect.Effect<string | null, StorageFailure | CredentialResolutionError> =>
       // Share a single refresh per connection so concurrent resolves of the same
       // connection all await one refresh-token grant (the AS rotates the refresh
@@ -1666,14 +1692,24 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
       // expiry can refresh again.
       Effect.gen(function* () {
         const key = connectionKey(row);
+        if (options?.force) refreshInFlight.delete(key);
         const existing = refreshInFlight.get(key);
         if (existing) return yield* existing;
-        // `Effect.cached` memoizes the grant onto a deferred: it runs once and
-        // replays to every awaiter sharing this entry.
-        const memoized = yield* Effect.cached(performTokenRefresh(row, provider));
-        const gated = memoized.pipe(
-          Effect.ensuring(Effect.sync(() => refreshInFlight.delete(key))),
+        // Cache the grant with cleanup attached to the underlying execution,
+        // rather than to an awaiting caller fiber. This keeps the entry only for
+        // the grant's de-duplication window, regardless of waiter interruption.
+        let gated!: Effect.Effect<
+          string | null,
+          StorageFailure | CredentialResolutionError
+        >;
+        const grant = performTokenRefresh(row, provider).pipe(
+          Effect.ensuring(
+            Effect.sync(() => {
+              if (refreshInFlight.get(key) === gated) refreshInFlight.delete(key);
+            }),
+          ),
         );
+        gated = yield* Effect.cached(grant);
         // Re-check after building (a peer fiber may have registered first while
         // we built ours) so everyone converges on the same shared grant.
         const winner = refreshInFlight.get(key) ?? gated;
@@ -1687,8 +1723,19 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
     // first (always single-input → `{ token: <access> }`).
     const resolveConnectionValues = (
       row: ConnectionRow,
+      options?: { readonly force?: boolean },
     ): Effect.Effect<Record<string, string | null>, StorageFailure | CredentialResolutionError> =>
       Effect.gen(function* () {
+        // The caller may hold a row loaded by an earlier operation. Re-read it
+        // before consulting expiry or provider values so rotations performed by
+        // another executor instance are observed.
+        const freshRow = yield* findConnectionRow({
+          owner: row.owner as Owner,
+          integration: IntegrationSlug.make(row.integration),
+          name: ConnectionName.make(row.name),
+        });
+        if (!freshRow) return {};
+        row = freshRow;
         const provider = credentialProviders.get(row.provider);
         if (!provider) {
           return yield* new CredentialProviderNotRegisteredError({
@@ -1698,13 +1745,19 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
         // OAuth connections refresh their access token before resolving when
         // it has expired (or is within the skew window).
         const expiresAt = row.expires_at == null ? null : Number(row.expires_at);
-        if (row.oauth_client != null && shouldRefreshToken({ expiresAt })) {
-          const access = yield* refreshConnectionToken(row, provider);
+        if (
+          row.oauth_client != null &&
+          (options?.force === true || shouldRefreshToken({ expiresAt }))
+        ) {
+          const access = yield* refreshConnectionToken(row, provider, options);
           return { [PRIMARY_INPUT_VARIABLE]: access };
         }
         const out: Record<string, string | null> = {};
         for (const [variable, itemId] of Object.entries(connectionItemIds(row))) {
-          out[variable] = yield* provider.get(ProviderItemId.make(itemId));
+          out[variable] = yield* provider.get(
+            ProviderItemId.make(itemId),
+            connectionCredentialScope(row),
+          );
         }
         return out;
       }).pipe(
@@ -1724,8 +1777,9 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
      *  callers that only ever need one value. */
     const resolveConnectionValue = (
       row: ConnectionRow,
+      options?: { readonly force?: boolean },
     ): Effect.Effect<string | null, StorageFailure | CredentialResolutionError> =>
-      resolveConnectionValues(row).pipe(
+      resolveConnectionValues(row, options).pipe(
         Effect.map((values) => values[PRIMARY_INPUT_VARIABLE] ?? null),
       );
 
@@ -1749,23 +1803,25 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
 
     const resolveConnectionValueByRef = (
       ref: ConnectionRef,
+      options?: { readonly force?: boolean },
     ): Effect.Effect<string | null, StorageFailure> =>
       foldResolutionFailure(
         Effect.gen(function* () {
           const row = yield* findConnectionRow(ref);
           if (!row) return null;
-          return yield* resolveConnectionValue(row);
+          return yield* resolveConnectionValue(row, options);
         }),
       );
 
     const resolveConnectionValuesByRef = (
       ref: ConnectionRef,
+      options?: { readonly force?: boolean },
     ): Effect.Effect<Record<string, string | null>, StorageFailure> =>
       foldResolutionFailure(
         Effect.gen(function* () {
           const row = yield* findConnectionRow(ref);
           if (!row) return {};
-          return yield* resolveConnectionValues(row);
+          return yield* resolveConnectionValues(row, options);
         }),
       );
 
@@ -2167,8 +2223,8 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
             connection: ref,
             template: existingRow ? AuthTemplateSlug.make(existingRow.template) : null,
             storage: runtime.storage,
-            getValue: () => resolveConnectionValueByRef(ref),
-            getValues: () => resolveConnectionValuesByRef(ref),
+            getValue: (options) => resolveConnectionValueByRef(ref, options),
+            getValues: (options) => resolveConnectionValuesByRef(ref, options),
           })
           .pipe(
             Effect.mapError((cause) =>
@@ -2363,10 +2419,14 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
             });
           }
           providerKey = String(provider.key);
+          const credentialScope = yield* Effect.try({
+            try: () => ownedKeys(input.owner),
+            catch: (cause) => storageFailureFromUnknown("invalid owner", cause),
+          });
           for (const i of pasted) {
             const itemId = `connection:${input.owner}:${input.integration}:${name}:${i.variable}`;
             if ("value" in i.origin && provider.set) {
-              yield* provider.set(ProviderItemId.make(itemId), i.origin.value);
+              yield* provider.set(ProviderItemId.make(itemId), i.origin.value, credentialScope);
             }
             itemIds[i.variable] = itemId;
           }
@@ -3842,6 +3902,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
 
     const oauth = makeOAuthService({
       fuma,
+      sessionFuma,
       owner: ownerBinding,
       tenant,
       subject,
@@ -3966,7 +4027,7 @@ export const createExecutor = <const TPlugins extends readonly AnyPlugin[] = rea
           remove: (ref) => connectionsRemove(ref),
           refresh: (ref) => connectionsRefresh(ref),
           markToolsStale: (ref) => connectionsMarkToolsStale(ref),
-          resolveValue: (ref) => resolveConnectionValueByRef(ref),
+          resolveValue: (ref, options) => resolveConnectionValueByRef(ref, options),
         },
         providers: {
           list: () => providersList(),

packages/core/sdk/src/index.ts

@@ -99,7 +99,7 @@ export type {
 export type { Tool, ToolDef, ToolListFilter, ToolAnnotations } from "./tool";
 
 // Credential providers.
-export type { CredentialProvider, ProviderEntry } from "./provider";
+export type { CredentialProvider, CredentialProviderScope, ProviderEntry } from "./provider";
 
 // Public projections / detection.
 export { ToolSchemaView, IntegrationDetectionResult } from "./types";

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

@@ -114,6 +114,8 @@ export type OAuthScopePolicy =
  *  connection row + produces tools), the owner binding, and the redirect base. */
 export interface OAuthServiceDeps {
   readonly fuma: IFumaClient;
+  /** Policy-free handle used only for OAuth callback state lifecycle. */
+  readonly sessionFuma: IFumaClient;
   readonly owner: OwnerBinding;
   readonly tenant: string;
   readonly subject: string | null;
@@ -1173,9 +1175,14 @@ export const makeOAuthService = (deps: OAuthServiceDeps): OAuthService => {
     input: OAuthCompleteInput,
   ): Effect.Effect<Connection, OAuthCompleteError | OAuthSessionNotFoundError | StorageFailure> =>
     Effect.gen(function* () {
-      const sessionRow = yield* deps.fuma.use("oauth_session.findFirst", (db) =>
+      // OAuth callbacks may arrive under a different authenticated identity than
+      // the browser that started the flow. State is an unguessable 256-bit value
+      // with a 15-minute TTL, so this single lookup deliberately bypasses owner
+      // visibility while retaining tenant/state matching.
+      const sessionRow = yield* deps.sessionFuma.use("oauth_session.findFirst", (db) =>
         looseDb(db).findFirst("oauth_session", {
-          where: (b: any) => b("state", "=", String(input.state)),
+          where: (b: any) =>
+            b.and(b("tenant", "=", deps.tenant), b("state", "=", String(input.state))),
         }),
       );
       if (!sessionRow) {
@@ -1327,12 +1334,17 @@ export const makeOAuthService = (deps: OAuthServiceDeps): OAuthService => {
         });
       }
       const itemId = accessItemId(target.owner, target.integration, target.name);
-      yield* provider.set(ProviderItemId.make(itemId), token.access_token);
+      const credentialScope = deps.ownedKeys(target.owner);
+      yield* provider.set(ProviderItemId.make(itemId), token.access_token, credentialScope);
 
       let refreshItemId: string | null = null;
       if (token.refresh_token) {
         refreshItemId = refreshItemIdFor(itemId);
-        yield* provider.set(ProviderItemId.make(refreshItemId), token.refresh_token);
+        yield* provider.set(
+          ProviderItemId.make(refreshItemId),
+          token.refresh_token,
+          credentialScope,
+        );
       }
 
       const oauthScope = recordedOAuthScope(token, requestedScopes);
@@ -1362,10 +1374,11 @@ export const makeOAuthService = (deps: OAuthServiceDeps): OAuthService => {
     });
 
   const deleteSession = (state: OAuthState): Effect.Effect<void, StorageFailure> =>
-    deps.fuma
+    deps.sessionFuma
       .use("oauth_session.delete", (db) =>
         looseDb(db).deleteMany("oauth_session", {
-          where: (b: any) => b("state", "=", String(state)),
+          where: (b: any) =>
+            b.and(b("tenant", "=", deps.tenant), b("state", "=", String(state))),
         }),
       )
       .pipe(Effect.asVoid);