improved plugin api

Overlord-Server/src/main-server.ts

@@ -67,6 +67,7 @@ import {
   sendPluginBundle,
   dispatchAutoLoadPlugins,
 } from "./server/plugin-state-bundle";
+import { createPluginRuntime } from "./server/plugin-runtime/runtime";
 import {
   DISCONNECT_TIMEOUT_MS,
   HEARTBEAT_INTERVAL_MS,
@@ -224,6 +225,15 @@ const savePluginState = () => savePluginStateToDisk(PLUGIN_ROOT, PLUGIN_STATE_PA
 const loadPluginState = async () => {
   pluginState = await loadPluginStateFromDisk(PLUGIN_STATE_PATH);
 };
+
+const pluginRuntime = createPluginRuntime({
+  pluginRoot: PLUGIN_ROOT,
+  workerHostUrl: new URL("./server/plugin-runtime/worker-host.ts", import.meta.url).href,
+  setLastError: (pluginId, error) => {
+    pluginState.lastError[pluginId] = error;
+    void savePluginState();
+  },
+});
 const ensurePluginExtracted = (pluginId: string) =>
   ensurePluginExtractedFromRoot(PLUGIN_ROOT, pluginId, sanitizePluginId);
 const listPluginManifests = () =>
@@ -323,6 +333,8 @@ const notificationPluginHandlers = createNotificationPluginHandlers({
   getUserRole: (userId: number) => getUserById(userId)?.role,
   storeNotificationScreenshot: storeNotificationScreenshotForPending,
   deliverNotificationWithScreenshot: deliverNotificationWithScreenshotForRecord,
+  forwardPluginEventToRuntime: (clientId, pluginId, event, payload) =>
+    pluginRuntime.dispatchClientEvent(clientId, pluginId, event, payload),
   getDeliveryTargetsForClientEvent: (event: string, clientId: string): UserDeliveryTarget[] => {
     const mapRow = (u: any): UserDeliveryTarget => ({
       userId: u.id,
@@ -369,6 +381,21 @@ const pendingCommandReplies = new Map<string, PendingCommandReply>();
 async function startServer() {
   await loadPluginState();
 
+  try {
+    const manifests = await listPluginManifests();
+    for (const manifest of manifests) {
+      if (pluginState.enabled[manifest.id] === false) continue;
+      if (!pluginRuntime.hasServerCode(manifest.id)) continue;
+      try {
+        await pluginRuntime.startPlugin(manifest.id);
+      } catch (err) {
+        logger.warn(`[plugin-runtime] boot failed for ${manifest.id}: ${(err as Error).message}`);
+      }
+    }
+  } catch (err) {
+    logger.warn(`[plugin-runtime] startup scan failed: ${(err as Error).message}`);
+  }
+
   setPostApproveHook((clientId: string) => {
     const client = clientManager.getClient(clientId);
     if (!client) return;
@@ -460,6 +487,7 @@ async function startServer() {
       drainPluginUIEvents: notificationPluginHandlers.drainPluginUIEvents,
       secureHeaders,
       mimeType,
+      pluginRuntime,
     },
     fileShare: {
       FILE_SHARE_ROOT,
@@ -775,12 +803,14 @@ startServer();
 
 process.on("SIGINT", () => {
   logger.info("\n[server] Shutting down gracefully...");
+  void pluginRuntime.shutdownAll();
   flushAuditLogsSync();
   process.exit(0);
 });
 
 process.on("SIGTERM", () => {
   logger.info("\n[server] Shutting down gracefully...");
+  void pluginRuntime.shutdownAll();
   flushAuditLogsSync();
   process.exit(0);
 });

Overlord-Server/src/protocol.ts

@@ -274,6 +274,7 @@ export type PluginManifest = {
     label?: string;
     icon?: string;
   };
+  hasServer?: boolean;
 };
 
 export type PluginSignatureInfo = {

Overlord-Server/src/server/plugin-runtime/runtime.ts

@@ -0,0 +1,307 @@
+import path from "path";
+import fs from "fs/promises";
+import { existsSync } from "fs";
+import { v4 as uuidv4 } from "uuid";
+import { logger } from "../../logger";
+import type { WorkerInbound, WorkerOutbound, PluginRpcCaller } from "./types";
+
+type Subscriber = {
+  id: string;
+  send: (sse: string) => void;
+  close: () => void;
+};
+
+type PendingRpc = {
+  resolve: (value: unknown) => void;
+  reject: (err: Error) => void;
+  timer: ReturnType<typeof setTimeout>;
+};
+
+type PluginInstance = {
+  pluginId: string;
+  worker: Worker;
+  ready: Promise<void>;
+  resolveReady: () => void;
+  rejectReady: (err: Error) => void;
+  isReady: boolean;
+  pending: Map<string, PendingRpc>;
+  subscribers: Set<Subscriber>;
+  startedAt: number;
+};
+
+const RPC_TIMEOUT_MS = 30_000;
+const SHUTDOWN_GRACE_MS = 750;
+
+export type PluginRuntime = {
+  startPlugin: (pluginId: string) => Promise<void>;
+  stopPlugin: (pluginId: string) => Promise<void>;
+  restartPlugin: (pluginId: string) => Promise<void>;
+  dispatchClientEvent: (
+    clientId: string,
+    pluginId: string,
+    event: string,
+    payload: unknown,
+  ) => void;
+  rpc: (
+    pluginId: string,
+    method: string,
+    params: unknown,
+    caller: PluginRpcCaller,
+  ) => Promise<unknown>;
+  subscribe: (
+    pluginId: string,
+    send: (sse: string) => void,
+    close: () => void,
+  ) => (() => void) | null;
+  isRunning: (pluginId: string) => boolean;
+  hasServerCode: (pluginId: string) => boolean;
+  shutdownAll: () => Promise<void>;
+};
+
+export type PluginRuntimeOptions = {
+  pluginRoot: string;
+  workerHostUrl: string; // file:// URL or absolute path that `new Worker()` accepts
+  setLastError: (pluginId: string, error: string) => void;
+};
+
+export function createPluginRuntime(opts: PluginRuntimeOptions): PluginRuntime {
+  const instances = new Map<string, PluginInstance>();
+
+  function pluginServerScript(pluginId: string): string {
+    return path.join(opts.pluginRoot, pluginId, "server.js");
+  }
+
+  function hasServerCode(pluginId: string): boolean {
+    return existsSync(pluginServerScript(pluginId));
+  }
+
+  function isRunning(pluginId: string): boolean {
+    return instances.has(pluginId);
+  }
+
+  async function startPlugin(pluginId: string): Promise<void> {
+    if (instances.has(pluginId)) return;
+    const script = pluginServerScript(pluginId);
+    if (!existsSync(script)) return;
+
+    const dataDir = path.join(opts.pluginRoot, pluginId, "data");
+    await fs.mkdir(dataDir, { recursive: true });
+    const dbPath = path.join(dataDir, "plugin.db");
+
+    const worker = new Worker(opts.workerHostUrl);
+
+    let resolveReady!: () => void;
+    let rejectReady!: (err: Error) => void;
+    const ready = new Promise<void>((res, rej) => {
+      resolveReady = res;
+      rejectReady = rej;
+    });
+
+    const inst: PluginInstance = {
+      pluginId,
+      worker,
+      ready,
+      resolveReady,
+      rejectReady,
+      isReady: false,
+      pending: new Map(),
+      subscribers: new Set(),
+      startedAt: Date.now(),
+    };
+
+    worker.onmessage = (e: MessageEvent<WorkerOutbound>) => {
+      handleWorkerMessage(inst, e.data);
+    };
+
+    (worker as any).onerror = (event: any) => {
+      const message = event?.message || event?.error?.message || String(event);
+      logger.error(`[plugin:${pluginId}] worker error: ${message}`);
+      opts.setLastError(pluginId, `worker crash: ${message}`);
+      rejectReady(new Error(message));
+      void stopPlugin(pluginId);
+    };
+
+    instances.set(pluginId, inst);
+
+    const bootMsg: WorkerInbound = {
+      type: "boot",
+      pluginId,
+      serverScript: script,
+      dbPath,
+      pluginRoot: opts.pluginRoot,
+    };
+    worker.postMessage(bootMsg);
+
+    try {
+      await ready;
+      inst.isReady = true;
+      logger.info(`[plugin:${pluginId}] server runtime ready (db=${path.basename(dbPath)})`);
+    } catch (err) {
+      const error = err instanceof Error ? err.message : String(err);
+      logger.error(`[plugin:${pluginId}] failed to start: ${error}`);
+      opts.setLastError(pluginId, error);
+      try { worker.terminate(); } catch {}
+      instances.delete(pluginId);
+      throw err;
+    }
+  }
+
+  async function stopPlugin(pluginId: string): Promise<void> {
+    const inst = instances.get(pluginId);
+    if (!inst) return;
+    instances.delete(pluginId);
+
+    for (const sub of inst.subscribers) {
+      try { sub.close(); } catch {}
+    }
+    inst.subscribers.clear();
+
+    for (const pending of inst.pending.values()) {
+      clearTimeout(pending.timer);
+      pending.reject(new Error("Plugin runtime stopped"));
+    }
+    inst.pending.clear();
+
+    try {
+      const shutdownMsg: WorkerInbound = { type: "shutdown" };
+      inst.worker.postMessage(shutdownMsg);
+    } catch {}
+
+    setTimeout(() => {
+      try { inst.worker.terminate(); } catch {}
+    }, SHUTDOWN_GRACE_MS);
+
+    logger.info(`[plugin:${pluginId}] server runtime stopped`);
+  }
+
+  async function restartPlugin(pluginId: string): Promise<void> {
+    if (instances.has(pluginId)) {
+      await stopPlugin(pluginId);
+    }
+    await startPlugin(pluginId);
+  }
+
+  function dispatchClientEvent(
+    clientId: string,
+    pluginId: string,
+    event: string,
+    payload: unknown,
+  ): void {
+    const inst = instances.get(pluginId);
+    if (!inst) return;
+    const send = () => {
+      try {
+        const msg: WorkerInbound = { type: "event", clientId, event, payload };
+        inst.worker.postMessage(msg);
+      } catch (err) {
+        logger.warn(
+          `[plugin:${pluginId}] event dispatch failed: ${(err as Error).message}`,
+        );
+      }
+    };
+    if (inst.isReady) {
+      send();
+    } else {
+      inst.ready.then(send).catch(() => {});
+    }
+  }
+
+  function rpc(
+    pluginId: string,
+    method: string,
+    params: unknown,
+    caller: PluginRpcCaller,
+  ): Promise<unknown> {
+    const inst = instances.get(pluginId);
+    if (!inst) {
+      return Promise.reject(new Error("Plugin runtime not running"));
+    }
+    const callOnReady = () =>
+      new Promise((resolve, reject) => {
+        const id = uuidv4();
+        const timer = setTimeout(() => {
+          if (inst.pending.delete(id)) {
+            reject(new Error(`RPC timeout: ${method}`));
+          }
+        }, RPC_TIMEOUT_MS);
+        inst.pending.set(id, { resolve, reject, timer });
+        const msg: WorkerInbound = { type: "rpc", id, method, params, caller };
+        try {
+          inst.worker.postMessage(msg);
+        } catch (err) {
+          clearTimeout(timer);
+          inst.pending.delete(id);
+          reject(err instanceof Error ? err : new Error(String(err)));
+        }
+      });
+    return inst.isReady ? callOnReady() : inst.ready.then(callOnReady);
+  }
+
+  function subscribe(
+    pluginId: string,
+    send: (sse: string) => void,
+    close: () => void,
+  ): (() => void) | null {
+    const inst = instances.get(pluginId);
+    if (!inst) return null;
+    const sub: Subscriber = { id: uuidv4(), send, close };
+    inst.subscribers.add(sub);
+    return () => {
+      inst.subscribers.delete(sub);
+    };
+  }
+
+  function handleWorkerMessage(inst: PluginInstance, msg: WorkerOutbound) {
+    if (msg.type === "ready") {
+      inst.resolveReady();
+      return;
+    }
+    if (msg.type === "boot_error") {
+      opts.setLastError(inst.pluginId, msg.error);
+      inst.rejectReady(new Error(msg.error));
+      return;
+    }
+    if (msg.type === "rpc_reply") {
+      const pending = inst.pending.get(msg.id);
+      if (!pending) return;
+      clearTimeout(pending.timer);
+      inst.pending.delete(msg.id);
+      if (msg.ok) pending.resolve(msg.result);
+      else pending.reject(new Error(msg.error));
+      return;
+    }
+    if (msg.type === "broadcast") {
+      const safeChannel = String(msg.channel || "message").replace(/[\r\n]/g, "");
+      const dataLine = JSON.stringify(msg.data ?? null);
+      const sse = `event: ${safeChannel}\ndata: ${dataLine}\n\n`;
+      for (const sub of inst.subscribers) {
+        try { sub.send(sse); } catch {}
+      }
+      return;
+    }
+    if (msg.type === "log") {
+      const tag = `[plugin:${inst.pluginId}]`;
+      if (msg.level === "info") logger.info(`${tag} ${msg.message}`);
+      else if (msg.level === "warn") logger.warn(`${tag} ${msg.message}`);
+      else if (msg.level === "error") logger.error(`${tag} ${msg.message}`);
+      else logger.debug(`${tag} ${msg.message}`);
+    }
+  }
+
+  async function shutdownAll(): Promise<void> {
+    const ids = Array.from(instances.keys());
+    await Promise.all(ids.map((id) => stopPlugin(id)));
+  }
+
+  return {
+    startPlugin,
+    stopPlugin,
+    restartPlugin,
+    dispatchClientEvent,
+    rpc,
+    subscribe,
+    isRunning,
+    hasServerCode,
+    shutdownAll,
+  };
+}

Overlord-Server/src/server/plugin-runtime/types.ts

@@ -0,0 +1,37 @@
+export type PluginRpcCaller = {
+  id: number;
+  role: string;
+};
+
+export type WorkerInbound =
+  | {
+      type: "boot";
+      pluginId: string;
+      serverScript: string;
+      dbPath: string;
+      pluginRoot: string;
+    }
+  | {
+      type: "event";
+      clientId: string;
+      event: string;
+      payload: unknown;
+    }
+  | {
+      type: "rpc";
+      id: string;
+      method: string;
+      params: unknown;
+      caller: PluginRpcCaller;
+    }
+  | {
+      type: "shutdown";
+    };
+
+export type WorkerOutbound =
+  | { type: "ready" }
+  | { type: "boot_error"; error: string }
+  | { type: "rpc_reply"; id: string; ok: true; result: unknown }
+  | { type: "rpc_reply"; id: string; ok: false; error: string }
+  | { type: "broadcast"; channel: string; data: unknown }
+  | { type: "log"; level: "debug" | "info" | "warn" | "error"; message: string };

Overlord-Server/src/server/plugin-runtime/worker-host.ts

@@ -0,0 +1,134 @@
+import { Database } from "bun:sqlite";
+import path from "path";
+import { pathToFileURL } from "url";
+import type { WorkerInbound, WorkerOutbound } from "./types";
+
+declare const self: {
+  postMessage: (msg: WorkerOutbound) => void;
+  onmessage: ((e: MessageEvent<WorkerInbound>) => void) | null;
+};
+
+type PluginExports = {
+  setup?: (ctx: PluginContext) => unknown | Promise<unknown>;
+  onEvent?: (ctx: PluginContext, clientId: string, event: string, payload: unknown) => unknown | Promise<unknown>;
+  rpc?: Record<string, (ctx: PluginContext, params: unknown, meta: { caller: { id: number; role: string } }) => unknown | Promise<unknown>>;
+  teardown?: (ctx: PluginContext) => unknown | Promise<unknown>;
+};
+
+type PluginContext = {
+  pluginId: string;
+  db: Database;
+  dataDir: string;
+  log: {
+    debug: (message: string) => void;
+    info: (message: string) => void;
+    warn: (message: string) => void;
+    error: (message: string) => void;
+  };
+  broadcast: (channel: string, data: unknown) => void;
+};
+
+let plugin: PluginExports | null = null;
+let db: Database | null = null;
+let ctx: PluginContext | null = null;
+let pluginId = "";
+
+function send(msg: WorkerOutbound) {
+  self.postMessage(msg);
+}
+
+function makeLogger(): PluginContext["log"] {
+  return {
+    debug: (m) => send({ type: "log", level: "debug", message: m }),
+    info: (m) => send({ type: "log", level: "info", message: m }),
+    warn: (m) => send({ type: "log", level: "warn", message: m }),
+    error: (m) => send({ type: "log", level: "error", message: m }),
+  };
+}
+
+self.onmessage = async (e: MessageEvent<WorkerInbound>) => {
+  const msg = e.data;
+
+  if (msg.type === "boot") {
+    try {
+      pluginId = msg.pluginId;
+      db = new Database(msg.dbPath, { create: true });
+      try {
+        db.exec("PRAGMA journal_mode = WAL");
+        db.exec("PRAGMA synchronous = NORMAL");
+        db.exec("PRAGMA foreign_keys = ON");
+      } catch {}
+
+      const dataDir = path.join(msg.pluginRoot, msg.pluginId, "data");
+      ctx = {
+        pluginId,
+        db,
+        dataDir,
+        log: makeLogger(),
+        broadcast: (channel, data) => send({ type: "broadcast", channel, data }),
+      };
+
+      const moduleUrl = `${pathToFileURL(msg.serverScript).href}?v=${Date.now()}`;
+      const mod = await import(/* @vite-ignore */ moduleUrl);
+      plugin = (mod.default || mod) as PluginExports;
+
+      if (typeof plugin.setup === "function") {
+        await plugin.setup(ctx);
+      }
+
+      send({ type: "ready" });
+    } catch (err) {
+      const error = err instanceof Error ? `${err.message}\n${err.stack || ""}` : String(err);
+      send({ type: "boot_error", error });
+    }
+    return;
+  }
+
+  if (msg.type === "event") {
+    if (!plugin || !ctx) return;
+    if (typeof plugin.onEvent !== "function") return;
+    try {
+      await plugin.onEvent(ctx, msg.clientId, msg.event, msg.payload);
+    } catch (err) {
+      const error = err instanceof Error ? err.message : String(err);
+      send({ type: "log", level: "error", message: `onEvent(${msg.event}) failed: ${error}` });
+    }
+    return;
+  }
+
+  if (msg.type === "rpc") {
+    if (!plugin || !ctx) {
+      send({ type: "rpc_reply", id: msg.id, ok: false, error: "Plugin not initialised" });
+      return;
+    }
+    const handler = plugin.rpc?.[msg.method];
+    if (typeof handler !== "function") {
+      send({ type: "rpc_reply", id: msg.id, ok: false, error: `Unknown RPC method: ${msg.method}` });
+      return;
+    }
+    try {
+      const result = await handler(ctx, msg.params, { caller: msg.caller });
+      send({ type: "rpc_reply", id: msg.id, ok: true, result: result ?? null });
+    } catch (err) {
+      const error = err instanceof Error ? err.message : String(err);
+      send({ type: "rpc_reply", id: msg.id, ok: false, error });
+    }
+    return;
+  }
+
+  if (msg.type === "shutdown") {
+    if (plugin && ctx && typeof plugin.teardown === "function") {
+      try {
+        await plugin.teardown(ctx);
+      } catch (err) {
+        const error = err instanceof Error ? err.message : String(err);
+        send({ type: "log", level: "warn", message: `teardown failed: ${error}` });
+      }
+    }
+    try { db?.close(); } catch {}
+    db = null;
+    plugin = null;
+    ctx = null;
+    return;
+  }
+};

Overlord-Server/src/server/plugin-state-bundle.ts

@@ -95,6 +95,7 @@ export async function ensurePluginExtracted(
   let htmlEntry: Buffer | null = null;
   let cssEntry: Buffer | null = null;
   let jsEntry: Buffer | null = null;
+  let serverEntry: Buffer | null = null;
   let configEntry: Buffer | null = null;
   const nativeBinaries: Map<string, Buffer> = new Map();
 
@@ -104,6 +105,8 @@ export async function ensurePluginExtracted(
     const lower = base.toLowerCase();
     if (lower.endsWith(".so") || lower.endsWith(".dll") || lower.endsWith(".dylib")) {
       nativeBinaries.set(base, entry.getData());
+    } else if (lower === "server.js") {
+      serverEntry = entry.getData();
     } else if (lower.endsWith(".html")) {
       htmlEntry = entry.getData();
     } else if (lower.endsWith(".css")) {
@@ -132,6 +135,15 @@ export async function ensurePluginExtracted(
   await fs.writeFile(path.join(assetsDir, `${safeId}.css`), cssEntry);
   await fs.writeFile(path.join(assetsDir, `${safeId}.js`), jsEntry);
 
+  const serverScriptPath = path.join(pluginDir, "server.js");
+  if (serverEntry) {
+    await fs.writeFile(serverScriptPath, serverEntry);
+  } else {
+    // A previous extraction may have left a stale server.js — drop it so the
+    // runtime accurately reflects the current bundle.
+    try { await fs.rm(serverScriptPath, { force: true }); } catch {}
+  }
+
   let extraConfig: any = {};
   if (configEntry) {
     try {
@@ -154,6 +166,7 @@ export async function ensurePluginExtracted(
       js: `${safeId}.js`,
     },
     ...(extraConfig.navbar && { navbar: extraConfig.navbar }),
+    hasServer: serverEntry !== null,
   };
   await fs.writeFile(manifestPath, JSON.stringify(manifest, null, 2));
 

Overlord-Server/src/server/routes/plugin-routes.ts

@@ -11,11 +11,13 @@ import { metrics } from "../../metrics";
 import { encodeMessage, type PluginSignatureInfo } from "../../protocol";
 import { getConfig, updatePluginsConfig } from "../../config";
 import { getOrVerifySignature, BUILTIN_TRUSTED_KEYS } from "../plugin-signature";
+import type { PluginRuntime } from "../plugin-runtime/runtime";
 
 type PluginManifest = {
   id: string;
   name: string;
   signature?: PluginSignatureInfo;
+  hasServer?: boolean;
 };
 
 type PluginBundle = {
@@ -50,6 +52,7 @@ type PluginRouteDeps = {
   drainPluginUIEvents: (clientId: string, pluginId: string) => Array<{ event: string; payload: any }>;
   secureHeaders: (contentType?: string) => Record<string, string>;
   mimeType: (path: string) => string;
+  pluginRuntime: PluginRuntime;
 };
 
 export async function handlePluginRoutes(
@@ -85,6 +88,8 @@ export async function handlePluginRoutes(
       autoLoad: deps.pluginState.autoLoad[p.id] === true,
       autoStartEvents: deps.pluginState.autoStartEvents[p.id] || [],
       signature: p.signature || { signed: false, trusted: false, valid: false },
+      hasServer: p.hasServer === true || deps.pluginRuntime.hasServerCode(p.id),
+      serverRunning: deps.pluginRuntime.isRunning(p.id),
     }));
     return Response.json({ plugins: enriched });
   }
@@ -244,6 +249,16 @@ export async function handlePluginRoutes(
       await deps.savePluginState();
     }
 
+    // Re-extracting the bundle may have replaced server.js, so cycle the
+    // runtime if the plugin is enabled.
+    if (deps.pluginState.enabled[pluginId] !== false && deps.pluginRuntime.hasServerCode(pluginId)) {
+      try {
+        await deps.pluginRuntime.restartPlugin(pluginId);
+      } catch {}
+    } else if (deps.pluginRuntime.isRunning(pluginId)) {
+      await deps.pluginRuntime.stopPlugin(pluginId);
+    }
+
     return Response.json({ ok: true, id: pluginId, enabled: deps.pluginState.enabled[pluginId], signature: sigInfo });
   }
 
@@ -282,6 +297,15 @@ export async function handlePluginRoutes(
 
     deps.pluginState.enabled[pluginId] = enabled;
     await deps.savePluginState();
+
+    if (enabled) {
+      if (deps.pluginRuntime.hasServerCode(pluginId) && !deps.pluginRuntime.isRunning(pluginId)) {
+        try { await deps.pluginRuntime.startPlugin(pluginId); } catch {}
+      }
+    } else if (deps.pluginRuntime.isRunning(pluginId)) {
+      await deps.pluginRuntime.stopPlugin(pluginId);
+    }
+
     return Response.json({ ok: true, id: pluginId, enabled });
   }
 
@@ -623,6 +647,10 @@ export async function handlePluginRoutes(
     delete deps.pluginState.autoStartEvents[pluginId];
     await deps.savePluginState();
 
+    if (deps.pluginRuntime.isRunning(pluginId)) {
+      await deps.pluginRuntime.stopPlugin(pluginId);
+    }
+
     return Response.json({ ok: true, id: pluginId });
   }
 
@@ -799,6 +827,116 @@ export async function handlePluginRoutes(
     return Response.json({ ok: true, id: pluginId });
   }
 
+  const pluginRpcMatch = url.pathname.match(/^\/api\/plugins\/([^/]+)\/rpc$/);
+  if (req.method === "POST" && pluginRpcMatch) {
+    const user = await authenticateRequest(req);
+    if (!user) return new Response("Unauthorized", { status: 401 });
+    try {
+      requirePermission(user, "clients:control");
+    } catch (error) {
+      if (error instanceof Response) return error;
+      return new Response("Forbidden", { status: 403 });
+    }
+
+    let pluginId = "";
+    try { pluginId = deps.sanitizePluginId(pluginRpcMatch[1]); } catch {
+      return new Response("Invalid plugin id", { status: 400 });
+    }
+
+    if (deps.pluginState.enabled[pluginId] === false) {
+      return Response.json({ ok: false, error: "Plugin disabled" }, { status: 400 });
+    }
+    if (!deps.pluginRuntime.isRunning(pluginId)) {
+      return Response.json(
+        { ok: false, error: "Plugin server runtime not running" },
+        { status: 503 },
+      );
+    }
+
+    let body: any = {};
+    try { body = await req.json(); } catch {}
+    const method = typeof body.method === "string" ? body.method.trim() : "";
+    if (!method) {
+      return Response.json({ ok: false, error: "Missing 'method'" }, { status: 400 });
+    }
+    if (method.length > 128) {
+      return Response.json({ ok: false, error: "Method name too long" }, { status: 400 });
+    }
+
+    try {
+      const result = await deps.pluginRuntime.rpc(pluginId, method, body.params, {
+        id: user.userId,
+        role: user.role,
+      });
+      metrics.recordCommand("plugin_rpc");
+      return Response.json({ ok: true, result });
+    } catch (err) {
+      return Response.json({ ok: false, error: (err as Error).message }, { status: 500 });
+    }
+  }
+
+  const pluginStreamMatch = url.pathname.match(/^\/api\/plugins\/([^/]+)\/stream$/);
+  if (req.method === "GET" && pluginStreamMatch) {
+    const user = await authenticateRequest(req);
+    if (!user) return new Response("Unauthorized", { status: 401 });
+    try {
+      requirePermission(user, "clients:control");
+    } catch (error) {
+      if (error instanceof Response) return error;
+      return new Response("Forbidden", { status: 403 });
+    }
+
+    let pluginId = "";
+    try { pluginId = deps.sanitizePluginId(pluginStreamMatch[1]); } catch {
+      return new Response("Invalid plugin id", { status: 400 });
+    }
+
+    if (deps.pluginState.enabled[pluginId] === false) {
+      return new Response("Plugin disabled", { status: 400 });
+    }
+    if (!deps.pluginRuntime.isRunning(pluginId)) {
+      return new Response("Plugin server runtime not running", { status: 503 });
+    }
+
+    const encoder = new TextEncoder();
+    let unsubscribe: (() => void) | null = null;
+    let heartbeat: ReturnType<typeof setInterval> | null = null;
+
+    const stream = new ReadableStream<Uint8Array>({
+      start(controller) {
+        const safeEnqueue = (chunk: Uint8Array) => {
+          try { controller.enqueue(chunk); } catch {}
+        };
+        const send = (sse: string) => safeEnqueue(encoder.encode(sse));
+        const close = () => {
+          try { controller.close(); } catch {}
+        };
+
+        unsubscribe = deps.pluginRuntime.subscribe(pluginId, send, close);
+        if (!unsubscribe) {
+          close();
+          return;
+        }
+
+        send(`: connected to ${pluginId}\n\n`);
+        heartbeat = setInterval(() => send(`: ping\n\n`), 25_000);
+      },
+      cancel() {
+        if (heartbeat) clearInterval(heartbeat);
+        if (unsubscribe) unsubscribe();
+      },
+    });
+
+    return new Response(stream, {
+      headers: {
+        "Content-Type": "text/event-stream; charset=utf-8",
+        "Cache-Control": "no-cache, no-transform",
+        Connection: "keep-alive",
+        "X-Accel-Buffering": "no",
+      },
+    });
+  }
+
   const pluginFrameMatch = url.pathname.match(/^\/plugins\/([^/]+)\/frame$/);
   if (req.method === "GET" && pluginFrameMatch) {
     const user = await authenticateRequest(req);

Overlord-Server/src/server/ws-notifications-plugin.ts

@@ -70,6 +70,12 @@ type CreateDeps = {
     clientId: string,
   ) => UserDeliveryTarget[];
   savePluginState: () => Promise<void>;
+  forwardPluginEventToRuntime?: (
+    clientId: string,
+    pluginId: string,
+    event: string,
+    payload: unknown,
+  ) => void;
 };
 
 function safeSendViewer(ws: ServerWebSocket<SocketData>, payload: unknown) {
@@ -441,6 +447,18 @@ export function createNotificationPluginHandlers(deps: CreateDeps) {
       if (pluginId && event) {
         bufferPluginUIEvent(clientId, pluginId, event, eventPayload ?? (error ? { error } : null));
       }
+      if (pluginId && event && deps.forwardPluginEventToRuntime) {
+        try {
+          deps.forwardPluginEventToRuntime(
+            clientId,
+            pluginId,
+            event,
+            eventPayload ?? (error ? { error } : null),
+          );
+        } catch (err) {
+          logger.warn(`[plugin] runtime dispatch failed: ${(err as Error).message}`);
+        }
+      }
     },
 
     drainPluginUIEvents,