feat(host-cloudflare): refresh internal integration catalogs once per deployed version

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<void> | 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<void> => {
   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);

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<

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<typeof makeBinding>) =>
+const makeExecutor = (binding: { readonly fetch: (request: Request) => Promise<Response> }) =>
   Effect.acquireRelease(
     Effect.gen(function* () {
       const internalHosts = { "tools.internal": binding };
@@ -65,6 +72,28 @@ const makeExecutor = (binding: ReturnType<typeof makeBinding>) =>
 const toolIds = (tools: ReadonlyArray<{ readonly address: unknown }>) =>
   tools.map((tool) => String(tool.address)).sort();
 
+const listCalls = (binding: ReturnType<typeof makeBinding>) =>
+  binding.methods.filter((call) => call === "/mcp/fixture tools/list").length;
+
+const makeStore = (fail = false) => {
+  const data = new Map<string, string>();
+  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);
+    }),
+  );
 });