diff --git a/apps/host-cloudflare/src/app.ts b/apps/host-cloudflare/src/app.ts index 46f36313e6..d15860fa33 100644 --- a/apps/host-cloudflare/src/app.ts +++ b/apps/host-cloudflare/src/app.ts @@ -10,7 +10,8 @@ import { } from "@executor-js/api/server"; import { internalPrincipal } from "./auth/cloudflare-access"; -import { reconcileForVersion } from "./internal-integrations/reconcile"; +import { reconcileInternalIntegrations } from "./internal-integrations/reconcile"; +import { makeInternalListTools } from "./internal-integrations/list-tools"; import { loadConfig, type CloudflareConfig, type CloudflareEnv } from "./config"; import { makeCloudflarePlugins, type CloudflarePlugins } from "./plugins"; import { createD1ExecutorDb } from "./db/d1"; @@ -44,7 +45,7 @@ import { preloadQuickJs } from "./quickjs"; let internalIntegrationsReconciled: Promise | null = null; // 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. +// existing declared catalogs whose tool list changed. Never fails the boot. const reconcileOncePerIsolate = ( config: CloudflareConfig, dbHandle: ExecutorDbHandle, @@ -61,12 +62,10 @@ const reconcileOncePerIsolate = ( config.organizationName, { orgWrites: "allowed" }, ); - yield* reconcileForVersion( - executor, - config.internalIntegrations, - env.CF_VERSION_METADATA?.id, - env.BLOBS, - ).pipe(Effect.ensuring(executor.close().pipe(Effect.ignore))); + yield* reconcileInternalIntegrations(executor, config.internalIntegrations, { + listTools: makeInternalListTools(config.internalHosts), + store: 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/config.ts b/apps/host-cloudflare/src/config.ts index 381e67fd51..6cfc752769 100644 --- a/apps/host-cloudflare/src/config.ts +++ b/apps/host-cloudflare/src/config.ts @@ -34,9 +34,6 @@ 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. */ @@ -148,7 +145,7 @@ export interface CloudflareConfig { type CloudflareConfigEnv = Omit< CloudflareEnv, - "DB" | "BLOBS" | "CF_VERSION_METADATA" | "ASSETS" | "MCP_SESSION" | "MCP_EXECUTION_OWNER" + "DB" | "BLOBS" | "ASSETS" | "MCP_SESSION" | "MCP_EXECUTION_OWNER" >; type CloudflareAccessEnv = Pick< diff --git a/apps/host-cloudflare/src/internal-integrations/list-tools.ts b/apps/host-cloudflare/src/internal-integrations/list-tools.ts new file mode 100644 index 0000000000..ad88cfad3e --- /dev/null +++ b/apps/host-cloudflare/src/internal-integrations/list-tools.ts @@ -0,0 +1,21 @@ +import { Effect } from "effect"; + +import { discoverToolsFromInput } from "@executor-js/plugin-mcp"; +import { + makeHostedHttpClientLayer, + type HostedInternalHosts, +} from "@executor-js/sdk/host-internal"; + +import type { ListTools } from "./reconcile"; + +/** Lists tools with the MCP plugin's discovery, routed like the plugin's own calls. */ +export const makeInternalListTools = (internalHosts: HostedInternalHosts): ListTools => { + const httpClientLayer = makeHostedHttpClientLayer({ internalHosts }); + return (endpoint) => + discoverToolsFromInput({ + transport: "remote", + endpoint, + remoteTransport: "streamable-http", + httpClientLayer, + }).pipe(Effect.map(({ manifest }) => manifest.tools)); +}; diff --git a/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts b/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts index 4e4d474fd0..7af283b651 100644 --- a/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts +++ b/apps/host-cloudflare/src/internal-integrations/reconcile.test.ts @@ -9,13 +9,13 @@ import { makeAnnotationsMcpServer } from "@executor-js/plugin-mcp/testing"; import type { InternalIntegration } from "../config"; import { makeCloudflarePlugins } from "../plugins"; +import { makeInternalListTools } from "./list-tools"; import { - needsCatalogRefresh, + hashToolList, reconcileInternalIntegrations, - reconcileForVersion, - recordRefreshedVersion, - REFRESHED_VERSION_KEY, - type RefreshedVersionStore, + toolsHashKey, + type HashStore, + type ListedTool, } from "./reconcile"; const decodeJson = Schema.decodeUnknownOption(Schema.fromJsonString(Schema.Unknown)); @@ -27,21 +27,44 @@ const FIXTURE: InternalIntegration = { endpoint: "https://tools.internal/mcp/fixture", }; +interface ToolRecord { + readonly name: string; + readonly [key: string]: unknown; +} + const makeBinding = () => { const methods: string[] = []; + const state: { transformTools: (tools: ToolRecord[]) => ToolRecord[]; down: boolean } = { + transformTools: (tools) => tools, + down: false, + }; return { methods, - fetch: async (request: Request) => { + state, + fetch: async (request: Request): Promise => { + if (state.down) return new Response("down", { status: 503 }); const body = decodeJson(await request.clone().text()); - if (Option.isSome(body) && typeof body.value === "object" && body.value !== null) { - if ("method" in body.value) { - methods.push(`${new URL(request.url).pathname} ${String(body.value.method)}`); - } + const method = + Option.isSome(body) && typeof body.value === "object" && body.value !== null + ? Reflect.get(body.value, "method") + : undefined; + if (method !== undefined) { + methods.push(`${new URL(request.url).pathname} ${String(method)}`); } const server = makeAnnotationsMcpServer(); const transport = new WebStandardStreamableHTTPServerTransport({ enableJsonResponse: true }); await server.connect(transport); - return transport.handleRequest(request); + const response = await transport.handleRequest(request); + if (method !== "tools/list") return response; + const payload = decodeJson(await response.text()); + if (Option.isNone(payload)) return response; + const result: unknown = Reflect.get(payload.value as object, "result"); + const tools: unknown = Reflect.get(result as object, "tools"); + Reflect.set(result as object, "tools", state.transformTools(tools as ToolRecord[])); + return new Response(JSON.stringify(payload.value), { + status: response.status, + headers: response.headers, + }); }, }; }; @@ -77,14 +100,14 @@ const listCalls = (binding: ReturnType) => const makeStore = (fail = false) => { const data = new Map(); - const store: RefreshedVersionStore = { - get: (key) => { + const store: HashStore = { + get: (key: string) => { // 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) => { + put: (key: string, value: string) => { // 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); @@ -94,6 +117,13 @@ const makeStore = (fail = false) => { return { data, store }; }; +const depsFor = (binding: ReturnType, store?: HashStore) => ({ + listTools: makeInternalListTools({ "tools.internal": binding }), + store, +}); + +const HASH_KEY = toolsHashKey("fixture"); + describe("reconcileInternalIntegrations", () => { it.live("creates the integration and its org connection main with the tools loaded", () => Effect.scoped( @@ -198,25 +228,7 @@ 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", () => + it.live("an existing integration is left alone when no tool lister is given", () => Effect.scoped( Effect.gen(function* () { const binding = makeBinding(); @@ -231,11 +243,12 @@ describe("reconcileInternalIntegrations", () => { ), ); - it.live("never refreshes an existing integration with a different endpoint", () => + it.live("never touches an existing integration with a different endpoint", () => Effect.scoped( Effect.gen(function* () { const binding = makeBinding(); const { executor } = yield* makeExecutor(binding); + const { data, store } = makeStore(); yield* executor.mcp.addServer({ transport: "remote", name: "Admin made", @@ -243,91 +256,248 @@ describe("reconcileInternalIntegrations", () => { endpoint: "https://tools.internal/mcp/other", }); - const outcome = yield* reconcileInternalIntegrations(executor, [FIXTURE], { - refreshExisting: true, - }); + const outcome = yield* reconcileInternalIntegrations( + executor, + [FIXTURE], + depsFor(binding, store), + ); expect(outcome).toEqual([["fixture", "skipped"]]); expect(binding.methods).toEqual([]); + expect(data.size).toBe(0); }), ), ); +}); + +describe("content-hash catalog refresh", () => { + const created = (binding: ReturnType, store: HashStore) => + Effect.gen(function* () { + const { executor } = yield* makeExecutor(binding); + yield* reconcileInternalIntegrations(executor, [FIXTURE]); + return { executor, store }; + }); - it.live("a newly created integration is refreshed once, not twice", () => + it.live("refreshes and stores the hash when none is stored", () => Effect.scoped( Effect.gen(function* () { const binding = makeBinding(); - const { executor } = yield* makeExecutor(binding); + const { data, store } = makeStore(); + const { executor } = yield* created(binding, store); + const before = listCalls(binding); - const outcome = yield* reconcileInternalIntegrations(executor, [FIXTURE], { - refreshExisting: true, - }); + const outcome = yield* reconcileInternalIntegrations( + executor, + [FIXTURE], + depsFor(binding, store), + ); - expect(outcome).toEqual([["fixture", "created"]]); - const plain = makeBinding(); - const { executor: other } = yield* makeExecutor(plain); - yield* reconcileInternalIntegrations(other, [FIXTURE]); - expect(listCalls(binding)).toBe(listCalls(plain)); + expect(outcome).toEqual([["fixture", "refreshed"]]); + expect(listCalls(binding)).toBe(before + 2); + expect(data.get(HASH_KEY)).toMatch(/^[0-9a-f]{64}$/); }), ), ); -}); -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("does not refresh when the tools are unchanged, only lists them", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { data, store } = makeStore(); + const { executor } = yield* created(binding, store); + const deps = depsFor(binding, store); + yield* reconcileInternalIntegrations(executor, [FIXTURE], deps); + const hash = data.get(HASH_KEY); + const before = listCalls(binding); + + const outcome = yield* reconcileInternalIntegrations(executor, [FIXTURE], deps); + + expect(outcome).toEqual([["fixture", "skipped"]]); + expect(listCalls(binding)).toBe(before + 1); + expect(data.get(HASH_KEY)).toBe(hash); + }), + ), ); - 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("refreshes and stores a new hash when a tool is added", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { data, store } = makeStore(); + const { executor } = yield* created(binding, store); + const deps = depsFor(binding, store); + yield* reconcileInternalIntegrations(executor, [FIXTURE], deps); + const oldHash = data.get(HASH_KEY); + binding.state.transformTools = (tools) => [ + ...tools, + { name: "brand_new", description: "new", inputSchema: { type: "object" } }, + ]; + + const outcome = yield* reconcileInternalIntegrations(executor, [FIXTURE], deps); + + expect(outcome).toEqual([["fixture", "refreshed"]]); + expect(data.get(HASH_KEY)).not.toBe(oldHash); + const tools = yield* executor.tools.list({ integration: IntegrationSlug.make("fixture") }); + expect(toolIds(tools)).toContain("tools.fixture.org.main.brand_new"); + }), + ), ); - it.live("does not record the version when a refresh failed, and retries on the next run", () => + it.live("does not store the hash when the refresh fails, 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 { executor } = yield* created(binding, store); + const failing = { + listTools: makeInternalListTools({ "tools.internal": binding }), + store, + }; + let listed = 0; + const gated = { + ...failing, + listTools: (endpoint: string) => + failing.listTools(endpoint).pipe( + Effect.tap(() => + Effect.sync(() => { + listed += 1; + binding.state.down = true; + }), + ), + ), + }; - const first = yield* reconcileForVersion(executor, [FIXTURE], "v1", store); + const first = yield* reconcileInternalIntegrations(executor, [FIXTURE], gated); expect(first).toEqual([["fixture", "failed"]]); - expect(data.has(REFRESHED_VERSION_KEY)).toBe(false); - healthy = true; - const second = yield* reconcileForVersion(executor, [FIXTURE], "v1", store); + expect(listed).toBe(1); + expect(data.has(HASH_KEY)).toBe(false); + binding.state.down = false; + const second = yield* reconcileInternalIntegrations(executor, [FIXTURE], failing); expect(second).toEqual([["fixture", "refreshed"]]); - expect(data.get(REFRESHED_VERSION_KEY)).toBe("v1"); + expect(data.has(HASH_KEY)).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); + it.live("a tools/list failure skips the integration and records nothing", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { data, store } = makeStore(); + const { executor } = yield* created(binding, store); + binding.state.down = true; + + const outcome = yield* reconcileInternalIntegrations( + executor, + [FIXTURE], + depsFor(binding, store), + ); + + expect(outcome).toEqual([["fixture", "failed"]]); + expect(data.size).toBe(0); + }), + ), + ); + + it.live("stores the hash of a newly created integration", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { data, store } = makeStore(); + const { executor } = yield* makeExecutor(binding); + + const outcome = yield* reconcileInternalIntegrations( + executor, + [FIXTURE], + depsFor(binding, store), + ); + + expect(outcome).toEqual([["fixture", "created"]]); + const second = yield* reconcileInternalIntegrations( + executor, + [FIXTURE], + depsFor(binding, store), + ); + expect(second).toEqual([["fixture", "skipped"]]); + expect(data.has(HASH_KEY)).toBe(true); + }), + ), + ); + + it.live("refreshes and does not fail when R2 errors", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { store } = makeStore(true); + const { executor } = yield* created(binding, store); + + const outcome = yield* reconcileInternalIntegrations( + executor, + [FIXTURE], + depsFor(binding, store), + ); + + expect(outcome).toEqual([["fixture", "refreshed"]]); + }), + ), + ); + + it.live("refreshes every time when no store is bound", () => + Effect.scoped( + Effect.gen(function* () { + const binding = makeBinding(); + const { executor } = yield* created(binding, makeStore().store); + + const outcome = yield* reconcileInternalIntegrations(executor, [FIXTURE], depsFor(binding)); + + expect(outcome).toEqual([["fixture", "refreshed"]]); + }), + ), + ); +}); + +describe("hashToolList", () => { + const tool = (toolName: string, extra: Partial = {}): ListedTool => ({ + toolName, + description: `${toolName} tool`, + inputSchema: { type: "object", properties: { a: { type: "string" }, b: { type: "number" } } }, + annotations: { readOnlyHint: true }, + ...extra, + }); + + it.live("is stable across tool order and object key order", () => + Effect.promise(async () => { + const reordered = tool("beta", { + inputSchema: { + properties: { b: { type: "number" }, a: { type: "string" } }, + type: "object", + }, + }); + expect(await hashToolList([tool("alpha"), tool("beta")])).toBe( + await hashToolList([reordered, tool("alpha")]), + ); + }), + ); + + it.live("changes with a description, a schema or an annotation", () => + Effect.promise(async () => { + const base = await hashToolList([tool("alpha")]); + expect(await hashToolList([tool("alpha", { description: "other" })])).not.toBe(base); + expect(await hashToolList([tool("alpha", { inputSchema: { type: "string" } })])).not.toBe( + base, + ); + expect( + await hashToolList([tool("alpha", { annotations: { readOnlyHint: false } })]), + ).not.toBe(base); + }), + ); + + it.live("has a known SHA-256 for an empty list", () => + Effect.promise(async () => { + expect(await hashToolList([])).toBe( + "4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945", + ); }), ); }); diff --git a/apps/host-cloudflare/src/internal-integrations/reconcile.ts b/apps/host-cloudflare/src/internal-integrations/reconcile.ts index 4365f51180..ef013535a3 100644 --- a/apps/host-cloudflare/src/internal-integrations/reconcile.ts +++ b/apps/host-cloudflare/src/internal-integrations/reconcile.ts @@ -13,6 +13,100 @@ export const INTERNAL_CONNECTION_NAME = "main"; export type ReconcileOutcome = "created" | "refreshed" | "skipped" | "failed"; +export const TOOLS_HASH_KEY_PREFIX = "internal-integrations/tools-hash/"; + +export const toolsHashKey = (slug: string): string => `${TOOLS_HASH_KEY_PREFIX}${slug}`; + +/** The parts of a listed tool that a catalog refresh would change. */ +export interface ListedTool { + readonly toolName: string; + readonly description?: string | null; + readonly inputSchema?: unknown; + readonly outputSchema?: unknown; + readonly annotations?: unknown; +} + +/** Lists the current tools of an MCP endpoint over the host's internal routing. */ +export type ListTools = (endpoint: string) => Effect.Effect, unknown>; + +export interface HashStore { + readonly get: (key: string) => Promise<{ readonly text: () => Promise } | null>; + readonly put: (key: string, value: string) => Promise; +} + +export interface ReconcileDeps { + /** Without it, existing integrations are never refreshed. */ + readonly listTools?: ListTools; + /** Without it, every existing integration with the declared endpoint is refreshed. */ + readonly store?: HashStore; +} + +const canonicalJson = (value: unknown): string => { + if (Array.isArray(value)) + return `[${value.map((item) => canonicalJson(item ?? null)).join(",")}]`; + if (typeof value === "object" && value !== null) { + const entries = Object.entries(value) + .filter(([, item]) => item !== undefined) + .sort(([a], [b]) => (a < b ? -1 : a > b ? 1 : 0)) + .map(([key, item]) => `${JSON.stringify(key)}:${canonicalJson(item)}`); + return `{${entries.join(",")}}`; + } + return JSON.stringify(value) ?? "null"; +}; + +/** SHA-256 hex of the tool list, independent of tool order and object key order. */ +export const hashToolList = async (tools: ReadonlyArray): Promise => { + const normalized = tools + .map((tool) => ({ + name: tool.toolName, + description: tool.description ?? null, + inputSchema: tool.inputSchema ?? null, + outputSchema: tool.outputSchema ?? null, + annotations: tool.annotations ?? null, + })) + .sort((a, b) => (a.name < b.name ? -1 : a.name > b.name ? 1 : 0)); + const digest = await crypto.subtle.digest( + "SHA-256", + new TextEncoder().encode(canonicalJson(normalized)), + ); + return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join(""); +}; + +const warnStore = (message: string) => (cause: unknown) => + Effect.logWarning(message, cause).pipe(Effect.annotateLogs({ step: "internal-catalog-refresh" })); + +const readStoredHash = ( + store: HashStore | undefined, + slug: string, +): Effect.Effect => { + if (store === undefined) return Effect.succeed(undefined as string | undefined); + return Effect.tryPromise(async () => { + const stored = await store.get(toolsHashKey(slug)); + return stored === null ? undefined : await stored.text(); + }).pipe( + Effect.catch((cause) => + warnStore(`tools hash for ${slug} could not be read, refreshing`)(cause).pipe( + Effect.as(undefined), + ), + ), + ); +}; + +const writeStoredHash = ( + store: HashStore | undefined, + slug: string, + hash: string, +): Effect.Effect => { + if (store === undefined) return Effect.void; + return Effect.tryPromise(() => store.put(toolsHashKey(slug), hash)).pipe( + Effect.asVoid, + Effect.catch(warnStore(`tools hash for ${slug} could not be recorded`)), + ); +}; + +const listAndHash = (listTools: ListTools, endpoint: string) => + listTools(endpoint).pipe(Effect.flatMap((tools) => Effect.promise(() => hashToolList(tools)))); + export interface InternalIntegrationExecutor { readonly integrations: { readonly get: (slug: IntegrationSlug) => Effect.Effect; @@ -76,20 +170,38 @@ const refreshMain = (executor: InternalIntegrationExecutor, slug: IntegrationSlu const reconcileOne = ( executor: InternalIntegrationExecutor, entry: InternalIntegration, - refreshExisting: boolean, + deps: ReconcileDeps, ) => 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; + if (deps.listTools === undefined || storedEndpoint(existing) !== entry.endpoint) { + yield* Effect.logInfo(`internal integration ${entry.slug} already exists, left untouched`); + return "skipped" as const; + } + const current = yield* listAndHash(deps.listTools, entry.endpoint); + const stored = yield* readStoredHash(deps.store, entry.slug); + if (stored === current) { + yield* Effect.logInfo(`internal integration ${entry.slug} catalog is current`); + return "skipped" as const; } - yield* Effect.logInfo(`internal integration ${entry.slug} already exists, left untouched`); - return "skipped" as const; + yield* refreshMain(executor, slug); + yield* writeStoredHash(deps.store, entry.slug, current); + yield* Effect.logInfo(`internal integration ${entry.slug} catalog refreshed`); + return "refreshed" as const; } + const listTools = deps.listTools; + const current = + listTools === undefined + ? undefined + : yield* listAndHash(listTools, entry.endpoint).pipe( + Effect.catch((cause) => + warnStore(`tools of ${entry.slug} could not be listed before creation`)(cause).pipe( + Effect.as(undefined), + ), + ), + ); yield* executor.mcp.addServer({ transport: "remote", name: entry.name, @@ -108,6 +220,7 @@ const reconcileOne = ( values: {}, }); yield* refreshMain(executor, slug); + if (current !== undefined) yield* writeStoredHash(deps.store, entry.slug, current); return "created" as const; }).pipe( Effect.catch((cause) => @@ -120,75 +233,19 @@ const reconcileOne = ( /** * Creates each declared integration that does not exist yet, with its org - * 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. + * connection `main`, and loads its tools. For an existing integration whose + * stored endpoint equals the declared one, lists the current tools and + * refreshes the `main` connection only when their content hash differs from the + * stored one; the hash is stored after a successful refresh. Never updates or + * deletes anything, and never fails the caller. */ export const reconcileInternalIntegrations = ( executor: InternalIntegrationExecutor, entries: readonly InternalIntegration[], - options: { readonly refreshExisting?: boolean } = {}, + deps: ReconcileDeps = {}, ): Effect.Effect> => Effect.forEach(entries, (entry) => - reconcileOne(executor, entry, options.refreshExisting === true).pipe( + reconcileOne(executor, entry, deps).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 54ba61a4ff..cb3838fa86 100644 --- a/apps/host-cloudflare/wrangler.jsonc +++ b/apps/host-cloudflare/wrangler.jsonc @@ -44,8 +44,6 @@ "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 diff --git a/packages/plugins/mcp/src/sdk/index.ts b/packages/plugins/mcp/src/sdk/index.ts index 99eae6baee..0e01e50ff2 100644 --- a/packages/plugins/mcp/src/sdk/index.ts +++ b/packages/plugins/mcp/src/sdk/index.ts @@ -37,3 +37,6 @@ export { variable, type ApiKeyAuthTemplate } from "@executor-js/sdk/http-auth"; export { McpConnectionError, McpToolDiscoveryError, McpOAuthError } from "./errors"; export { deriveMcpNamespace, joinToolPath, extractManifestFromListToolsResult } from "./manifest"; + +export { discoverToolsFromInput } from "./discover"; +export type { McpToolManifest, McpToolManifestEntry } from "./manifest";