diff --git a/apps/host-cloudflare/src/app.ts b/apps/host-cloudflare/src/app.ts index a523a408d9..46f36313e6 100644 --- a/apps/host-cloudflare/src/app.ts +++ b/apps/host-cloudflare/src/app.ts @@ -10,7 +10,7 @@ import { } from "@executor-js/api/server"; import { internalPrincipal } from "./auth/cloudflare-access"; -import { reconcileInternalIntegrations } 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"; @@ -43,10 +43,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 +61,12 @@ const reconcileOncePerIsolate = ( config.organizationName, { orgWrites: "allowed" }, ); - yield* reconcileInternalIntegrations(executor, config.internalIntegrations).pipe( - Effect.ensuring(executor.close().pipe(Effect.ignore)), - ); + 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)), @@ -89,7 +94,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..4e4d474fd0 100644 --- a/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts +++ b/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts @@ -9,7 +9,14 @@ import { makeAnnotationsMcpServer } from "@executor-js/plugin-mcp/testing"; import type { InternalIntegration } from "../config"; import { makeCloudflarePlugins } from "../plugins"; -import { reconcileInternalIntegrations } from "./reconcile"; +import { + needsCatalogRefresh, + reconcileInternalIntegrations, + reconcileForVersion, + recordRefreshedVersion, + REFRESHED_VERSION_KEY, + type RefreshedVersionStore, +} from "./reconcile"; const decodeJson = Schema.decodeUnknownOption(Schema.fromJsonString(Schema.Unknown)); @@ -39,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 }; @@ -65,6 +72,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 +197,137 @@ 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 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(); + 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..4365f51180 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: { @@ -37,14 +37,56 @@ export interface InternalIntegrationExecutor { readonly values: Record; }) => Effect.Effect; readonly refresh: (ref: ConnectionRef) => Effect.Effect; + readonly list: (filter: { + readonly integration: IntegrationSlug; + }) => Effect.Effect, unknown>; }; } -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; +}; + +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, + 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* refreshMain(executor, slug); + 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; } @@ -65,7 +107,7 @@ const reconcileOne = (executor: InternalIntegrationExecutor, entry: InternalInte 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) => @@ -78,13 +120,75 @@ 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")), ); +}; + +/** + * 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; + }); 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