feat(plugins): server plugin host

apps/server/package.json

@@ -77,6 +77,7 @@
     "@effect/vitest": "catalog:",
     "@opencode-ai/sdk": "^1.3.15",
     "@t3tools/contracts": "workspace:*",
+    "@t3tools/plugin-sdk": "workspace:*",
     "@t3tools/provider-testing": "workspace:*",
     "@t3tools/secret-provider-core": "workspace:*",
     "@t3tools/secret-provider-definitions": "workspace:*",

apps/server/src/auth/RpcAuthorization.test.ts

@@ -68,6 +68,16 @@ describe("RPC authorization scopes", () => {
     }
   });
 
+  it("keeps enabling, reloading and configuring plugins, which runs their code, behind settings write", () => {
+    for (const method of [
+      WS_METHODS.pluginsSetEnabled,
+      WS_METHODS.pluginsReload,
+      WS_METHODS.pluginsSettingsUpdate,
+    ]) {
+      expect(requiredScopeForRpcMethod(method)).toBe(AuthSettingsWriteScope);
+    }
+  });
+
   it("rejects unknown RPC method names", () => {
     for (const method of ["server.notRegistered", "toString", "constructor"]) {
       expect(() => requiredScopeForRpcMethod(method)).toThrow(

apps/server/src/auth/RpcAuthorization.ts

@@ -205,6 +205,13 @@ export const RPC_REQUIRED_SCOPES = {
   [WS_METHODS.deviceDetail]: AuthOrchestrationReadScope,
   [WS_METHODS.deviceAction]: AuthOrchestrationOperateScope,
   [WS_METHODS.subscribeDeviceState]: AuthOrchestrationReadScope,
+  // `plugins.invoke` also needs the operate scope for methods a plugin registered as writes;
+  // PluginHost checks that against the plugin's own declaration. Channels are reads.
+  [WS_METHODS.pluginsList]: AuthOrchestrationReadScope,
+  [WS_METHODS.pluginsSubscribeStates]: AuthOrchestrationReadScope,
+  [WS_METHODS.pluginsSettingsGet]: AuthOrchestrationReadScope,
+  [WS_METHODS.pluginsInvoke]: AuthOrchestrationReadScope,
+  [WS_METHODS.pluginsSubscribe]: AuthOrchestrationReadScope,
   [WS_METHODS.subscribeServerConfig]: AuthOrchestrationReadScope,
   [WS_METHODS.subscribeServerLifecycle]: AuthOrchestrationReadScope,
   [WS_METHODS.subscribeAuthAccess]: AuthAccessReadScope,

apps/server/src/environment/ServerEnvironment.ts

@@ -243,6 +243,7 @@ export const make = Effect.gen(function* () {
       threadSnooze: true,
       threadForceStop: true,
       environmentThemes: true,
+      plugins: true,
       usageLimitSources: true,
       providerWorkspaceSnapshotsOnDemand: true,
       usagePriceOverrides: true,

apps/server/src/observability/RpcInstrumentation.ts

@@ -87,6 +87,14 @@ const RPC_AGGREGATES = {
   [WS_METHODS.serverRefreshUsageRates]: "server",
   [WS_METHODS.serverSignalProcess]: "server",
   [WS_METHODS.serverReportClientActivity]: "server",
+  [WS_METHODS.pluginsList]: "plugins",
+  [WS_METHODS.pluginsSubscribeStates]: "plugins",
+  [WS_METHODS.pluginsSetEnabled]: "plugins",
+  [WS_METHODS.pluginsReload]: "plugins",
+  [WS_METHODS.pluginsSettingsGet]: "plugins",
+  [WS_METHODS.pluginsSettingsUpdate]: "plugins",
+  [WS_METHODS.pluginsInvoke]: "plugins",
+  [WS_METHODS.pluginsSubscribe]: "plugins",
   [WS_METHODS.scheduledTasksSubscribe]: "scheduledTasks",
   [WS_METHODS.scheduledTasksUpsert]: "scheduledTasks",
   [WS_METHODS.scheduledTasksSetEnabled]: "scheduledTasks",

apps/server/src/plugins/PluginApi.ts

@@ -0,0 +1,149 @@
+import { PluginRpcName, type PluginId } from "@t3tools/contracts";
+import type { PluginApi, PluginRpcAccess, PluginRpcHandler } from "@t3tools/plugin-sdk/server";
+import * as Cause from "effect/Cause";
+import type * as Context from "effect/Context";
+import * as Effect from "effect/Effect";
+import * as Exit from "effect/Exit";
+import * as Option from "effect/Option";
+import * as PubSub from "effect/PubSub";
+import * as Schema from "effect/Schema";
+import * as Stream from "effect/Stream";
+
+import type * as PluginSettingsStore from "./PluginSettingsStore.ts";
+import type * as PluginStorage from "./PluginStorage.ts";
+
+const isJson = Schema.is(Schema.Json);
+const isPluginRpcName = Schema.is(PluginRpcName);
+const isPluginRpcAccess = Schema.is(Schema.Literals(["read", "write"]));
+
+export interface PluginPublication {
+  readonly channel: string;
+  readonly value: Schema.Json;
+}
+
+/** Everything one activation registered. The host routes calls through it and tears it down. */
+export interface PluginActivation {
+  readonly methods: Map<
+    string,
+    { readonly handler: PluginRpcHandler; readonly access: PluginRpcAccess }
+  >;
+  readonly cleanups: Array<() => void | Promise<void>>;
+  /** Aborted on deactivate; handler signals and settings iterators follow it. */
+  readonly controller: AbortController;
+  readonly publications: PubSub.PubSub<PluginPublication>;
+}
+
+export const makePluginActivation = (
+  publications: PubSub.PubSub<PluginPublication>,
+): PluginActivation => ({
+  methods: new Map(),
+  cleanups: [],
+  controller: new AbortController(),
+  publications,
+});
+
+const assertActive = (activation: PluginActivation) => {
+  if (activation.controller.signal.aborted) {
+    throw new Error("This plugin activation has ended.");
+  }
+};
+
+/**
+ * Builds the object a plugin's `activate` receives. Effect stays on this side: every call runs
+ * with the host's services and settles as a plain Promise that rejects with an `Error`.
+ */
+export const makePluginApi = (input: {
+  readonly id: PluginId;
+  readonly rootDir: string;
+  readonly settingsSchema: Schema.JsonObject | undefined;
+  readonly activation: PluginActivation;
+  readonly context: Context.Context<never>;
+  readonly settings: PluginSettingsStore.PluginSettingsStore["Service"];
+  readonly storage: PluginStorage.PluginStorage["Service"];
+}): PluginApi => {
+  const { id, activation, context } = input;
+  const runExit = Effect.runPromiseExitWith(context);
+  const call = <A, E>(effect: Effect.Effect<A, E>): Promise<A> =>
+    runExit(effect).then((exit) => {
+      if (Exit.isSuccess(exit)) return exit.value;
+      const error = Cause.squash(exit.cause);
+      throw new Error(error instanceof Error ? error.message : "Plugin API call failed.");
+    });
+  const log =
+    (level: "Debug" | "Info" | "Warn" | "Error") =>
+    (message: string, fields?: Schema.JsonObject) => {
+      Effect.runForkWith(context)(
+        Effect.logWithLevel(level)(message, fields ?? {}).pipe(
+          Effect.annotateLogs({ pluginId: id }),
+        ),
+      );
+    };
+
+  return {
+    version: 1,
+    id,
+    rootDir: input.rootDir,
+    log: { debug: log("Debug"), info: log("Info"), warn: log("Warn"), error: log("Error") },
+    settings: {
+      get: () => call(input.settings.get(id, input.settingsSchema)),
+      changes: () =>
+        Stream.toAsyncIterableWith(
+          input.settings.changes(id).pipe(
+            Stream.interruptWhen(
+              Effect.callback<void>((resume) => {
+                const onAbort = () => resume(Effect.void);
+                if (activation.controller.signal.aborted) return onAbort();
+                activation.controller.signal.addEventListener("abort", onAbort, { once: true });
+                return Effect.sync(() =>
+                  activation.controller.signal.removeEventListener("abort", onAbort),
+                );
+              }),
+            ),
+          ),
+          context,
+        ),
+    },
+    storage: {
+      get: (key) => call(input.storage.get(id, key)).then(Option.getOrUndefined),
+      set: (key, value) => {
+        if (!isJson(value)) return Promise.reject(new TypeError("Storage values must be JSON."));
+        return call(input.storage.set(id, key, value));
+      },
+      delete: (key) => call(input.storage.remove(id, key)),
+      list: (prefix) => call(input.storage.list(id, prefix)),
+    },
+    rpc: {
+      register: (method, handler, options) => {
+        assertActive(activation);
+        if (!isPluginRpcName(method))
+          throw new TypeError(`"${method}" is not a valid method name.`);
+        // Plugins are untyped JavaScript, and a misspelled access level must not open a write.
+        if (!isPluginRpcAccess(options.access)) {
+          throw new TypeError(`Method "${method}" needs access "read" or "write".`);
+        }
+        if (activation.methods.has(method)) {
+          throw new Error(`Method "${method}" is already registered.`);
+        }
+        activation.methods.set(method, { handler, access: options.access });
+      },
+      publish: (channel, value) => {
+        assertActive(activation);
+        if (!isPluginRpcName(channel)) {
+          throw new TypeError(`"${channel}" is not a valid channel name.`);
+        }
+        if (!isJson(value)) throw new TypeError("Published values must be JSON.");
+        PubSub.publishUnsafe(activation.publications, { channel, value });
+      },
+    },
+    // Something created after deactivation has nothing left to wait for, so it cleans up now.
+    onDeactivate: (cleanup) => {
+      if (activation.controller.signal.aborted) {
+        void Promise.resolve()
+          .then(cleanup)
+          .catch(() => undefined);
+        return;
+      }
+      activation.cleanups.push(cleanup);
+    },
+  };
+};