Load server plugins from the deployment bucket at runtime

cloudflare/plugins.ts

@@ -1,68 +1,27 @@
-import type { ArtifactPlugin } from "../src/plugins/config";
-import type { PluginRequest, PluginUser } from "../src/plugins/types";
+import type { PluginInfo, PluginRequest } from "../src/plugins/types";
 import installed from "../dist/cloudflare/plugin-server";
 import { validatePluginInput, validatePluginOutput } from "../dist/cloudflare/plugin-validators.js";
 import catalog from "../dist/cloudflare/plugin-catalog.json";
+import { assertPluginRequest, createPluginDispatcher, pluginJson, PluginError, type PluginInvocationContext } from "../src/plugins/dispatch";
+import { callRuntimePlugin, runtimePlugin, runtimePluginCatalog, type RuntimePluginEnvironment } from "./runtime-plugins";
 
-export const PLUGIN_JSON_LIMIT = 256 * 1024;
+export { createPluginDispatcher, PluginError, PLUGIN_JSON_LIMIT, type PluginInvocationContext } from "../src/plugins/dispatch";
 export const PLUGIN_GUIDE = `Deployment plugins provide installed browser modules and authenticated server functions. Call plugins_list to discover operation names, schemas, and read-only hints. Call artifact_plugin_call with {plugin, operation, input}, or use pluginCall from sidequery/artifacts in a hosted artifact. Calls run as the authenticated deployment user, independently of artifact, version, workspace, or library selectors. Plugins may restrict users and must check row-level access in their handlers. Public standalone artifacts cannot call plugins. Provider credentials stay on the server. Artifact backends and scripts do not receive this bridge.`;
 export const pluginCatalog = catalog;
-export type PluginInvocationContext = { user: PluginUser; env: object };
-export class PluginError extends Error {
-  constructor(message: string, readonly status: 400 | 401 | 403 | 404 | 413 | 500 | 504) { super(message); }
-}
 
-function jsonValue(value: unknown, output = false): unknown {
-  let serialized: string | undefined;
-  try {
-    serialized = JSON.stringify(value, (_key, part) => {
-      if (typeof part === "number" && !Number.isFinite(part) || ["undefined", "function", "symbol", "bigint"].includes(typeof part)) throw new Error();
-      return part;
-    });
-  } catch { throw new PluginError(output ? "Plugin returned invalid JSON" : "Input must be JSON", output ? 500 : 400); }
-  if (serialized === undefined) throw new PluginError(output ? "Plugin returned invalid JSON" : "Input must be JSON", output ? 500 : 400);
-  if (new TextEncoder().encode(serialized).byteLength > PLUGIN_JSON_LIMIT) throw new PluginError(output ? "Plugin result exceeds 256 KiB" : "Plugin input exceeds 256 KiB", output ? 500 : 413);
-  return JSON.parse(serialized);
+const installedNames = new Set(installed.map(plugin => plugin.name));
+const dispatchInstalled = createPluginDispatcher(installed, { input: validatePluginInput, output: validatePluginOutput });
+
+/** Build-time plugins take precedence; a published plugin cannot shadow one. */
+export async function dispatchPlugin(request: PluginRequest, context?: PluginInvocationContext): Promise<unknown> {
+  assertPluginRequest(request, context);
+  if (installedNames.has(request.plugin)) return dispatchInstalled(request, context);
+  const record = await runtimePlugin(context.env as RuntimePluginEnvironment, request.plugin);
+  if (!record?.operations.some(operation => operation.name === request.operation)) throw new PluginError("Plugin operation not found", 404);
+  return callRuntimePlugin(record, request.operation, pluginJson(request.input), context.user, context);
 }
 
-export function createPluginDispatcher(plugins: readonly ArtifactPlugin[], validators: {
-  input: typeof validatePluginInput; output: typeof validatePluginOutput;
-}, timeoutMs = 30_000) {
-  return async (request: PluginRequest, context?: PluginInvocationContext): Promise<unknown> => {
-    if (!context?.user.subject || !context.user.authority) throw new PluginError("Plugin sign-in required", 401);
-    if (!request || typeof request !== "object" || typeof request.plugin !== "string" || typeof request.operation !== "string" || !Object.hasOwn(request, "input") || Object.keys(request).some(key => !["plugin", "operation", "input"].includes(key))) throw new PluginError("Invalid plugin request", 400);
-    const plugin = plugins.find(plugin => plugin.name === request.plugin);
-    const operation = plugin?.operations && Object.hasOwn(plugin.operations, request.operation) ? plugin.operations[request.operation] : undefined;
-    if (!plugin || !operation) throw new PluginError("Plugin operation not found", 404);
-    const input = jsonValue(request.input);
-    if (!validators.input(request.plugin, request.operation, input)) throw new PluginError("Invalid plugin input", 400);
-    // Handlers are trusted deployment code. The signal cancels cooperative I/O;
-    // the deadline bounds asynchronous work even if it ignores cancellation.
-    const controller = new AbortController();
-    let timer: ReturnType<typeof setTimeout>;
-    const timeout = new Promise<never>((_resolve, reject) => { timer = setTimeout(() => { controller.abort(); reject(new PluginError("Plugin operation timed out", 504)); }, timeoutMs); });
-    try {
-      return await Promise.race([timeout, (async () => {
-        const user = Object.freeze({ subject: context.user.subject, authority: context.user.authority });
-        let authorized: boolean;
-        try { authorized = operation.authorize ? await operation.authorize(user) : true; }
-        catch { throw new PluginError("Plugin authorization failed", 500); }
-        if (!authorized) throw new PluginError("Plugin operation is not allowed", 403);
-        if (controller.signal.aborted) throw new PluginError("Plugin operation timed out", 504);
-        const secrets: Record<string, string> = Object.create(null);
-        for (const name of plugin.secrets ?? []) {
-          const value = Object.hasOwn(context.env, name) ? (context.env as Record<string, unknown>)[name] : undefined;
-          if (typeof value !== "string") throw new PluginError("Plugin configuration unavailable", 500);
-          secrets[name] = value;
-        }
-        let result: unknown;
-        try { result = await operation.handler(input, { user, secrets: Object.freeze(secrets), signal: controller.signal }); }
-        catch { throw new PluginError("Plugin operation failed", 500); }
-        const output = jsonValue(result, true);
-        if (!validators.output(request.plugin, request.operation, output)) throw new PluginError("Plugin returned invalid output", 500);
-        return output;
-      })()]);
-    } finally { clearTimeout(timer!); }
-  };
+export async function pluginCatalogFor(env: object): Promise<PluginInfo[]> {
+  const published = await runtimePluginCatalog(env as RuntimePluginEnvironment);
+  return [...catalog, ...published.filter(plugin => !installedNames.has(plugin.name))];
 }
-export const dispatchPlugin = createPluginDispatcher(installed, { input: validatePluginInput, output: validatePluginOutput });

cloudflare/runtime-plugins-celld.test.ts

@@ -0,0 +1,76 @@
+import { afterAll, beforeAll, expect, test } from "bun:test";
+import { existsSync } from "node:fs";
+import { cp, mkdtemp, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join, resolve } from "node:path";
+import { Client } from "@modelcontextprotocol/sdk/client/index.js";
+import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";
+import { bundleRuntimePlugin } from "../scripts/bundle-plugin";
+import { prepareCelldConfig } from "../scripts/prepare-celld";
+import { spawnCelld, stopCelld, waitForReady, type RunningCelld } from "../e2e/celld-auth-runtime";
+
+const root = resolve(import.meta.dir, "..");
+const enabled = process.env.CELLD_INTEGRATION === "1";
+const celldTest = enabled ? test : test.skip;
+const plugins = { binding: "PLUGINS", bucket_name: "artifacts-plugins" };
+let project = "";
+let origin = "";
+let runtime: RunningCelld | undefined;
+
+function reservePort() {
+  const listener = Bun.listen({ hostname: "127.0.0.1", port: 0, socket: { data() {} } });
+  const selected = listener.port;
+  listener.stop(true);
+  return selected;
+}
+
+beforeAll(async () => {
+  if (!enabled) return;
+  const binary = process.env.CELLD_BIN ?? Bun.which("celld");
+  const esbuild = process.env.CELLD_ESBUILD ?? resolve(root, "node_modules/.bin/esbuild");
+  if (!binary) throw new Error("celld was not found; install celld or set CELLD_BIN to its executable");
+  if (!existsSync(join(root, "dist/worker-app/worker.js"))) throw new Error("missing dist/worker-app/worker.js; run bun run build:cloudflare");
+  const record = await bundleRuntimePlugin(join(root, "e2e/runtime-plugin-fixture.ts"));
+  project = await mkdtemp(join(tmpdir(), "artifact-celld-plugins-"));
+  await cp(join(root, "dist/worker-app"), join(project, "dist/worker-app"), { recursive: true });
+  await cp(join(root, "dist/cloudflare/assets"), join(project, "dist/cloudflare/assets"), { recursive: true });
+  const configPath = join(project, "wrangler.jsonc");
+  const config = await prepareCelldConfig(join(root, "wrangler.jsonc"), configPath);
+  config.r2_buckets = [...(config.r2_buckets as unknown[] ?? []), plugins];
+  config.vars = { ...(config.vars as Record<string, unknown>), PLUGIN_WAREHOUSE_TOKEN: "warehouse-test-token", HOST_ONLY_SECRET: "host-only-value" };
+  const port = reservePort();
+  origin = `http://127.0.0.1:${port}`;
+  // Publish through the binding into the project's persistent local storage,
+  // as the auth runtime bootstraps D1, then serve Artifacts from that storage.
+  await writeFile(join(project, "seed.js"), 'export default { async fetch(request, env) { if (request.method === "PUT") await env.PLUGINS.put(new URL(request.url).searchParams.get("key"), await request.text()); return new Response("ready"); } };\n');
+  await writeFile(configPath, JSON.stringify({ name: config.name, main: "seed.js", compatibility_date: config.compatibility_date, r2_buckets: [plugins] }));
+  runtime = spawnCelld(binary, esbuild, project, port);
+  await waitForReady(runtime, origin);
+  const seeded = await fetch(`${origin}/?key=plugins/test-warehouse.json`, { method: "PUT", body: JSON.stringify(record) });
+  if (!seeded.ok) throw new Error(`Seeding failed: ${seeded.status}\n${runtime.logs()}`);
+  await stopCelld(runtime);
+  await writeFile(configPath, `${JSON.stringify(config, null, 2)}\n`);
+  runtime = spawnCelld(binary, esbuild, project, port);
+  await waitForReady(runtime, `${origin}/health`);
+}, 120_000);
+
+afterAll(async () => {
+  await stopCelld(runtime);
+  if (project) await rm(project, { recursive: true, force: true });
+});
+
+celldTest("a published plugin runs in its own isolate on celld", async () => {
+  const client = new Client({ name: "runtime-plugin-celld", version: "1" });
+  try {
+    await client.connect(new StreamableHTTPClientTransport(new URL(`${origin}/mcp?workspace=test`)));
+    const listed = await client.callTool({ name: "plugins_list", arguments: {} });
+    expect((listed.structuredContent as { plugins: { name: string }[] }).plugins.map(plugin => plugin.name)).toContain("test-warehouse");
+    const called = await client.callTool({ name: "artifact_plugin_call", arguments: { plugin: "test-warehouse", operation: "query", input: { sql: "select 1" } } });
+    expect(called.isError, JSON.stringify(called.content) + runtime?.logs()).not.toBe(true);
+    const result = (called.structuredContent as { result: Record<string, unknown> }).result;
+    expect(result).toMatchObject({ sql: "select 1", user: { subject: "local", authority: "local" }, secrets: { PLUGIN_WAREHOUSE_TOKEN: "warehouse-test-token" } });
+    expect(result.processEnv).not.toContain("HOST_ONLY_SECRET");
+    const invalid = await client.callTool({ name: "artifact_plugin_call", arguments: { plugin: "test-warehouse", operation: "query", input: {} } });
+    expect(invalid.structuredContent).toMatchObject({ error: "Invalid plugin input", status: 400 });
+  } finally { await client.close(); }
+}, 60_000);