From 34c772cef0574a59e81e54a80bbbed17cafc4058 Mon Sep 17 00:00:00 2001 From: fforres Date: Thu, 8 Oct 2026 23:00:17 -0700 Subject: [PATCH 1/2] feat(host-cloudflare): refresh declared internal integration catalogs once per deployed version After reconcile, refresh the org main connection of each declared internal integration that already existed and whose stored endpoint equals the declared one. The last refreshed Worker version id is kept in the BLOBS bucket and only written after the refreshes ran. A missing binding or an R2 error refreshes anyway and never fails boot. --- apps/host-cloudflare/src/app.ts | 21 ++- apps/host-cloudflare/src/config.ts | 5 +- .../internal-integrations/reconcile.test.ts | 135 +++++++++++++++++- .../src/internal-integrations/reconcile.ts | 71 ++++++++- apps/host-cloudflare/wrangler.jsonc | 2 + 5 files changed, 221 insertions(+), 13 deletions(-) diff --git a/apps/host-cloudflare/src/app.ts b/apps/host-cloudflare/src/app.ts index a523a408d9..0d3e97d1fb 100644 --- a/apps/host-cloudflare/src/app.ts +++ b/apps/host-cloudflare/src/app.ts @@ -10,7 +10,11 @@ import { } from "@executor-js/api/server"; import { internalPrincipal } from "./auth/cloudflare-access"; -import { reconcileInternalIntegrations } from "./internal-integrations/reconcile"; +import { + needsCatalogRefresh, + reconcileInternalIntegrations, + recordRefreshedVersion, +} from "./internal-integrations/reconcile"; import { loadConfig, type CloudflareConfig, type CloudflareEnv } from "./config"; import { makeCloudflarePlugins, type CloudflarePlugins } from "./plugins"; import { createD1ExecutorDb } from "./db/d1"; @@ -43,10 +47,12 @@ import { preloadQuickJs } from "./quickjs"; let internalIntegrationsReconciled: Promise | null = null; -// Once per isolate, by a system executor that may write org rows. Never fails the boot. +// Once per isolate, by a system executor that may write org rows. Also refreshes +// existing declared catalogs once per deployed version. Never fails the boot. const reconcileOncePerIsolate = ( config: CloudflareConfig, dbHandle: ExecutorDbHandle, + env: CloudflareEnv, ): Promise => { if (config.internalIntegrations.length === 0) return Promise.resolve(); internalIntegrationsReconciled ??= Effect.runPromise( @@ -59,9 +65,12 @@ const reconcileOncePerIsolate = ( config.organizationName, { orgWrites: "allowed" }, ); - yield* reconcileInternalIntegrations(executor, config.internalIntegrations).pipe( - Effect.ensuring(executor.close().pipe(Effect.ignore)), - ); + const versionId = env.CF_VERSION_METADATA?.id; + const refreshExisting = yield* needsCatalogRefresh(versionId, env.BLOBS); + yield* reconcileInternalIntegrations(executor, config.internalIntegrations, { + refreshExisting, + }).pipe(Effect.ensuring(executor.close().pipe(Effect.ignore))); + if (refreshExisting) yield* recordRefreshedVersion(versionId, env.BLOBS); }).pipe( Effect.provide(dbProviderLayer(Effect.succeed(dbHandle))), Effect.provide(makeCloudflarePluginsProvider(config)), @@ -89,7 +98,7 @@ export const makeCloudflareApp = async ( // Open and idempotently bring up the D1 schema once. This is the long-lived // handle the per-request scoped executor reads through the DbProvider seam. const dbHandle = await createD1ExecutorDb(env.DB, env.BLOBS); - await reconcileOncePerIsolate(config, dbHandle); + await reconcileOncePerIsolate(config, dbHandle, env); const identityLayer = cloudflareAccessIdentityLayer(config); const mcpAgentHandler = makeCloudflareMcpAgentHandler(config); const approvalHandler = makeCloudflareApprovalHandler(config, env); diff --git a/apps/host-cloudflare/src/config.ts b/apps/host-cloudflare/src/config.ts index 6cfc752769..381e67fd51 100644 --- a/apps/host-cloudflare/src/config.ts +++ b/apps/host-cloudflare/src/config.ts @@ -34,6 +34,9 @@ export interface CloudflareEnv { readonly DB: D1Database; /** R2 bucket binding — holds values too large for a D1 row (~1-2MB cap). */ readonly BLOBS?: R2Bucket; + /** Version metadata binding (wrangler.jsonc `version_metadata`): the id of + * the deployed Worker version. */ + readonly CF_VERSION_METADATA?: { readonly id: string }; /** Static assets binding (wrangler.jsonc `assets.binding`). The MCP session * DO fetches the built MCP-Apps shell document through it — a deployed * Worker has no filesystem to read the shell from. */ @@ -145,7 +148,7 @@ export interface CloudflareConfig { type CloudflareConfigEnv = Omit< CloudflareEnv, - "DB" | "BLOBS" | "ASSETS" | "MCP_SESSION" | "MCP_EXECUTION_OWNER" + "DB" | "BLOBS" | "CF_VERSION_METADATA" | "ASSETS" | "MCP_SESSION" | "MCP_EXECUTION_OWNER" >; type CloudflareAccessEnv = Pick< diff --git a/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts b/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts index d1f5856d19..1c14209b7f 100644 --- a/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts +++ b/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts @@ -9,7 +9,13 @@ import { makeAnnotationsMcpServer } from "@executor-js/plugin-mcp/testing"; import type { InternalIntegration } from "../config"; import { makeCloudflarePlugins } from "../plugins"; -import { reconcileInternalIntegrations } from "./reconcile"; +import { + needsCatalogRefresh, + reconcileInternalIntegrations, + recordRefreshedVersion, + REFRESHED_VERSION_KEY, + type RefreshedVersionStore, +} from "./reconcile"; const decodeJson = Schema.decodeUnknownOption(Schema.fromJsonString(Schema.Unknown)); @@ -65,6 +71,28 @@ const makeExecutor = (binding: ReturnType) => const toolIds = (tools: ReadonlyArray<{ readonly address: unknown }>) => tools.map((tool) => String(tool.address)).sort(); +const listCalls = (binding: ReturnType) => + binding.methods.filter((call) => call === "/mcp/fixture tools/list").length; + +const makeStore = (fail = false) => { + const data = new Map(); + const store: RefreshedVersionStore = { + get: (key) => { + // oxlint-disable-next-line executor/no-promise-reject -- boundary: stand-in for a failing R2 binding + if (fail) return Promise.reject("r2 down"); + const value = data.get(key); + return Promise.resolve(value === undefined ? null : { text: () => Promise.resolve(value) }); + }, + put: (key, value) => { + // oxlint-disable-next-line executor/no-promise-reject -- boundary: stand-in for a failing R2 binding + if (fail) return Promise.reject("r2 down"); + data.set(key, value); + return Promise.resolve(); + }, + }; + return { data, store }; +}; + describe("reconcileInternalIntegrations", () => { it.live("creates the integration and its org connection main with the tools loaded", () => Effect.scoped( @@ -168,4 +196,109 @@ describe("reconcileInternalIntegrations", () => { }), ), ); + + it.live("refreshes an existing integration with the declared endpoint when asked", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { executor } = yield* makeExecutor(binding); + yield* reconcileInternalIntegrations(executor, [FIXTURE]); + const before = listCalls(binding); + + const outcome = yield* reconcileInternalIntegrations(executor, [FIXTURE], { + refreshExisting: true, + }); + + expect(outcome).toEqual([["fixture", "refreshed"]]); + expect(listCalls(binding)).toBe(before + 1); + }), + ), + ); + + it.live("does not refresh an existing integration when not asked", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { executor } = yield* makeExecutor(binding); + yield* reconcileInternalIntegrations(executor, [FIXTURE]); + const before = listCalls(binding); + + yield* reconcileInternalIntegrations(executor, [FIXTURE]); + + expect(listCalls(binding)).toBe(before); + }), + ), + ); + + it.live("never refreshes an existing integration with a different endpoint", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { executor } = yield* makeExecutor(binding); + yield* executor.mcp.addServer({ + transport: "remote", + name: "Admin made", + slug: "fixture", + endpoint: "https://tools.internal/mcp/other", + }); + + const outcome = yield* reconcileInternalIntegrations(executor, [FIXTURE], { + refreshExisting: true, + }); + + expect(outcome).toEqual([["fixture", "skipped"]]); + expect(binding.methods).toEqual([]); + }), + ), + ); + + it.live("a newly created integration is refreshed once, not twice", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { executor } = yield* makeExecutor(binding); + + const outcome = yield* reconcileInternalIntegrations(executor, [FIXTURE], { + refreshExisting: true, + }); + + expect(outcome).toEqual([["fixture", "created"]]); + const plain = makeBinding(); + const { executor: other } = yield* makeExecutor(plain); + yield* reconcileInternalIntegrations(other, [FIXTURE]); + expect(listCalls(binding)).toBe(listCalls(plain)); + }), + ), + ); +}); + +describe("refreshed version guard", () => { + it.live("refreshes on a new version id and not on the same one", () => + Effect.gen(function* () { + const { data, store } = makeStore(); + expect(yield* needsCatalogRefresh("v1", store)).toBe(true); + yield* recordRefreshedVersion("v1", store); + expect(data.get(REFRESHED_VERSION_KEY)).toBe("v1"); + expect(yield* needsCatalogRefresh("v1", store)).toBe(false); + expect(yield* needsCatalogRefresh("v2", store)).toBe(true); + }), + ); + + it.live("refreshes and does not fail when R2 errors or the bindings are missing", () => + Effect.gen(function* () { + const { store } = makeStore(true); + expect(yield* needsCatalogRefresh("v1", store)).toBe(true); + yield* recordRefreshedVersion("v1", store); + expect(yield* needsCatalogRefresh(undefined, makeStore().store)).toBe(true); + expect(yield* needsCatalogRefresh("v1", undefined)).toBe(true); + }), + ); + + it.live("does not record a version when the id is missing", () => + Effect.gen(function* () { + const { data, store } = makeStore(); + yield* recordRefreshedVersion(undefined, store); + expect(data.size).toBe(0); + }), + ); }); diff --git a/apps/host-cloudflare/src/internal-integrations/reconcile.ts b/apps/host-cloudflare/src/internal-integrations/reconcile.ts index 64c5178936..448b0aecca 100644 --- a/apps/host-cloudflare/src/internal-integrations/reconcile.ts +++ b/apps/host-cloudflare/src/internal-integrations/reconcile.ts @@ -11,7 +11,7 @@ import type { InternalIntegration } from "../config"; export const INTERNAL_CONNECTION_NAME = "main"; -export type ReconcileOutcome = "created" | "skipped" | "failed"; +export type ReconcileOutcome = "created" | "refreshed" | "skipped" | "failed"; export interface InternalIntegrationExecutor { readonly integrations: { @@ -40,11 +40,30 @@ export interface InternalIntegrationExecutor { }; } -const reconcileOne = (executor: InternalIntegrationExecutor, entry: InternalIntegration) => +const storedEndpoint = (integration: unknown): string | undefined => { + if (typeof integration !== "object" || integration === null) return undefined; + const url: unknown = Reflect.get(integration, "displayUrl"); + return typeof url === "string" ? url : undefined; +}; + +const reconcileOne = ( + executor: InternalIntegrationExecutor, + entry: InternalIntegration, + refreshExisting: boolean, +) => Effect.gen(function* () { const slug = IntegrationSlug.make(entry.slug); const existing = yield* executor.integrations.get(slug); if (existing !== null && existing !== undefined) { + if (refreshExisting && storedEndpoint(existing) === entry.endpoint) { + yield* executor.connections.refresh({ + owner: "org", + integration: slug, + name: ConnectionName.make(INTERNAL_CONNECTION_NAME), + }); + yield* Effect.logInfo(`internal integration ${entry.slug} catalog refreshed`); + return "refreshed" as const; + } yield* Effect.logInfo(`internal integration ${entry.slug} already exists, left untouched`); return "skipped" as const; } @@ -78,13 +97,55 @@ const reconcileOne = (executor: InternalIntegrationExecutor, entry: InternalInte /** * Creates each declared integration that does not exist yet, with its org - * connection `main`, and loads its tools. Never updates or deletes anything, - * and never fails the caller. + * connection `main`, and loads its tools. With `refreshExisting`, also refreshes + * the `main` connection of an existing integration whose stored endpoint equals + * the declared one. Never updates or deletes anything, and never fails the caller. */ export const reconcileInternalIntegrations = ( executor: InternalIntegrationExecutor, entries: readonly InternalIntegration[], + options: { readonly refreshExisting?: boolean } = {}, ): Effect.Effect> => Effect.forEach(entries, (entry) => - reconcileOne(executor, entry).pipe(Effect.map((outcome) => [entry.slug, outcome] as const)), + reconcileOne(executor, entry, options.refreshExisting === true).pipe( + Effect.map((outcome) => [entry.slug, outcome] as const), + ), + ); + +export const REFRESHED_VERSION_KEY = "internal-integrations/refreshed-version"; + +export interface RefreshedVersionStore { + readonly get: (key: string) => Promise<{ readonly text: () => Promise } | null>; + readonly put: (key: string, value: string) => Promise; +} + +const warnStore = (message: string) => (cause: unknown) => + Effect.logWarning(message, cause).pipe(Effect.annotateLogs({ step: "internal-catalog-refresh" })); + +/** True unless the stored version id equals `versionId`; any doubt means refresh. */ +export const needsCatalogRefresh = ( + versionId: string | undefined, + store: RefreshedVersionStore | undefined, +): Effect.Effect => { + if (versionId === undefined || store === undefined) return Effect.succeed(true); + return Effect.tryPromise(async () => { + const stored = await store.get(REFRESHED_VERSION_KEY); + return stored === null ? undefined : await stored.text(); + }).pipe( + Effect.map((stored) => stored !== versionId), + Effect.catch((cause) => + warnStore("refreshed version could not be read, refreshing")(cause).pipe(Effect.as(true)), + ), + ); +}; + +export const recordRefreshedVersion = ( + versionId: string | undefined, + store: RefreshedVersionStore | undefined, +): Effect.Effect => { + if (versionId === undefined || store === undefined) return Effect.void; + return Effect.tryPromise(() => store.put(REFRESHED_VERSION_KEY, versionId)).pipe( + Effect.asVoid, + Effect.catch(warnStore("refreshed version could not be recorded")), ); +}; diff --git a/apps/host-cloudflare/wrangler.jsonc b/apps/host-cloudflare/wrangler.jsonc index cb3838fa86..54ba61a4ff 100644 --- a/apps/host-cloudflare/wrangler.jsonc +++ b/apps/host-cloudflare/wrangler.jsonc @@ -44,6 +44,8 @@ "bucket_name": "executor-blobs", }, ], + // Deployed version id, used to refresh internal integration catalogs once per version. + "version_metadata": { "binding": "CF_VERSION_METADATA" }, // Dynamic Worker Loader binding: when present, sandboxed code runs in real // workerd isolates with `globalOutbound: null` (platform-enforced zero egress) // via @executor-js/runtime-dynamic-worker; without it the host falls back to From ba34cf3d628f37295272311c68d37b7afa7ed235 Mon Sep 17 00:00:00 2001 From: fforres Date: Thu, 8 Oct 2026 23:02:26 -0700 Subject: [PATCH 2/2] fix(host-cloudflare): record the refreshed version only when no internal integration failed A refresh that cannot reach its server succeeds but records a tool sync failure on the connection, so that now counts as failed. The version id is written only when every entry succeeded, so a binding that was down at boot is retried on the next boot. --- apps/host-cloudflare/src/app.ts | 18 +++--- .../internal-integrations/reconcile.test.ts | 31 ++++++++++- .../src/internal-integrations/reconcile.ts | 55 +++++++++++++++++-- 3 files changed, 86 insertions(+), 18 deletions(-) diff --git a/apps/host-cloudflare/src/app.ts b/apps/host-cloudflare/src/app.ts index 0d3e97d1fb..46f36313e6 100644 --- a/apps/host-cloudflare/src/app.ts +++ b/apps/host-cloudflare/src/app.ts @@ -10,11 +10,7 @@ import { } from "@executor-js/api/server"; import { internalPrincipal } from "./auth/cloudflare-access"; -import { - needsCatalogRefresh, - reconcileInternalIntegrations, - recordRefreshedVersion, -} from "./internal-integrations/reconcile"; +import { reconcileForVersion } from "./internal-integrations/reconcile"; import { loadConfig, type CloudflareConfig, type CloudflareEnv } from "./config"; import { makeCloudflarePlugins, type CloudflarePlugins } from "./plugins"; import { createD1ExecutorDb } from "./db/d1"; @@ -65,12 +61,12 @@ const reconcileOncePerIsolate = ( config.organizationName, { orgWrites: "allowed" }, ); - const versionId = env.CF_VERSION_METADATA?.id; - const refreshExisting = yield* needsCatalogRefresh(versionId, env.BLOBS); - yield* reconcileInternalIntegrations(executor, config.internalIntegrations, { - refreshExisting, - }).pipe(Effect.ensuring(executor.close().pipe(Effect.ignore))); - if (refreshExisting) yield* recordRefreshedVersion(versionId, env.BLOBS); + yield* reconcileForVersion( + executor, + config.internalIntegrations, + env.CF_VERSION_METADATA?.id, + env.BLOBS, + ).pipe(Effect.ensuring(executor.close().pipe(Effect.ignore))); }).pipe( Effect.provide(dbProviderLayer(Effect.succeed(dbHandle))), Effect.provide(makeCloudflarePluginsProvider(config)), diff --git a/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts b/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts index 1c14209b7f..4e4d474fd0 100644 --- a/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts +++ b/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts @@ -12,6 +12,7 @@ import { makeCloudflarePlugins } from "../plugins"; import { needsCatalogRefresh, reconcileInternalIntegrations, + reconcileForVersion, recordRefreshedVersion, REFRESHED_VERSION_KEY, type RefreshedVersionStore, @@ -45,7 +46,7 @@ const makeBinding = () => { }; }; -const makeExecutor = (binding: ReturnType) => +const makeExecutor = (binding: { readonly fetch: (request: Request) => Promise }) => Effect.acquireRelease( Effect.gen(function* () { const internalHosts = { "tools.internal": binding }; @@ -294,6 +295,34 @@ describe("refreshed version guard", () => { }), ); + it.live("does not record the version when a refresh failed, and retries on the next run", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + let healthy = true; + const toggled = { + fetch: (request: Request) => + healthy + ? binding.fetch(request) + : Promise.resolve(new Response("down", { status: 503 })), + }; + const { executor } = yield* makeExecutor(toggled); + yield* reconcileInternalIntegrations(executor, [FIXTURE]); + const { data, store } = makeStore(); + healthy = false; + + const first = yield* reconcileForVersion(executor, [FIXTURE], "v1", store); + + expect(first).toEqual([["fixture", "failed"]]); + expect(data.has(REFRESHED_VERSION_KEY)).toBe(false); + healthy = true; + const second = yield* reconcileForVersion(executor, [FIXTURE], "v1", store); + expect(second).toEqual([["fixture", "refreshed"]]); + expect(data.get(REFRESHED_VERSION_KEY)).toBe("v1"); + }), + ), + ); + it.live("does not record a version when the id is missing", () => Effect.gen(function* () { const { data, store } = makeStore(); diff --git a/apps/host-cloudflare/src/internal-integrations/reconcile.ts b/apps/host-cloudflare/src/internal-integrations/reconcile.ts index 448b0aecca..4365f51180 100644 --- a/apps/host-cloudflare/src/internal-integrations/reconcile.ts +++ b/apps/host-cloudflare/src/internal-integrations/reconcile.ts @@ -37,6 +37,9 @@ export interface InternalIntegrationExecutor { readonly values: Record; }) => Effect.Effect; readonly refresh: (ref: ConnectionRef) => Effect.Effect; + readonly list: (filter: { + readonly integration: IntegrationSlug; + }) => Effect.Effect, unknown>; }; } @@ -46,6 +49,30 @@ const storedEndpoint = (integration: unknown): string | undefined => { return typeof url === "string" ? url : undefined; }; +class ToolSyncFailed { + readonly _tag = "ToolSyncFailed"; +} + +// A refresh that cannot reach the server succeeds and records the failure on the connection. +const refreshMain = (executor: InternalIntegrationExecutor, slug: IntegrationSlug) => + Effect.gen(function* () { + const name = ConnectionName.make(INTERNAL_CONNECTION_NAME); + yield* executor.connections.refresh({ owner: "org", integration: slug, name }); + const connections = yield* executor.connections.list({ integration: slug }); + const main = connections.find( + (connection) => + typeof connection === "object" && + connection !== null && + Reflect.get(connection, "owner") === "org" && + String(Reflect.get(connection, "name")) === INTERNAL_CONNECTION_NAME, + ); + const health: unknown = + main === undefined ? undefined : Reflect.get(main as object, "lastHealth"); + const reason: unknown = + typeof health === "object" && health !== null ? Reflect.get(health, "reason") : undefined; + if (reason === "tool_sync_failed") return yield* Effect.fail(new ToolSyncFailed()); + }); + const reconcileOne = ( executor: InternalIntegrationExecutor, entry: InternalIntegration, @@ -56,11 +83,7 @@ const reconcileOne = ( const existing = yield* executor.integrations.get(slug); if (existing !== null && existing !== undefined) { if (refreshExisting && storedEndpoint(existing) === entry.endpoint) { - yield* executor.connections.refresh({ - owner: "org", - integration: slug, - name: ConnectionName.make(INTERNAL_CONNECTION_NAME), - }); + yield* refreshMain(executor, slug); yield* Effect.logInfo(`internal integration ${entry.slug} catalog refreshed`); return "refreshed" as const; } @@ -84,7 +107,7 @@ const reconcileOne = ( template: AuthTemplateSlug.make("none"), values: {}, }); - yield* executor.connections.refresh({ owner: "org", integration: slug, name }); + yield* refreshMain(executor, slug); return "created" as const; }).pipe( Effect.catch((cause) => @@ -149,3 +172,23 @@ export const recordRefreshedVersion = ( Effect.catch(warnStore("refreshed version could not be recorded")), ); }; + +/** + * Reconciles, refreshing existing catalogs when this version has not refreshed + * yet. The version is recorded only when no entry failed, so a binding that was + * down at boot is retried on the next boot. + */ +export const reconcileForVersion = ( + executor: InternalIntegrationExecutor, + entries: readonly InternalIntegration[], + versionId: string | undefined, + store: RefreshedVersionStore | undefined, +): Effect.Effect> => + Effect.gen(function* () { + const refreshExisting = yield* needsCatalogRefresh(versionId, store); + const outcomes = yield* reconcileInternalIntegrations(executor, entries, { refreshExisting }); + if (refreshExisting && outcomes.every(([, outcome]) => outcome !== "failed")) { + yield* recordRefreshedVersion(versionId, store); + } + return outcomes; + });