diff --git a/src/backend/plugins/broker.ts b/src/backend/plugins/broker.ts
new file mode 100644
index 000000000..f826e33ff
--- /dev/null
+++ b/src/backend/plugins/broker.ts
@@ -0,0 +1,721 @@
+/**
+ * The main-thread half of the plugin boundary.
+ *
+ * Every ctx call a plugin makes arrives here as a message. For each one the
+ * broker, in this order:
+ *   1. looks up the method
+ *   2. checks the plugin's granted capability (manifest-declared AND granted)
+ *   3. performs the real action
+ *   4. writes one audit line attributed to the plugin
+ *   5. replies with a structured-cloneable value
+ *
+ * Steps 2 and 4 are not the plugin's to influence. The plugin supplies only the
+ * arguments and the audit `details`; the actor, action and outcome are set from
+ * values the broker controls, so a plugin cannot forge an audit trail or
+ * attribute its actions to a user.
+ */
+
+import type { Worker } from "node:worker_threads";
+import { randomUUID } from "node:crypto";
+import { pluginLogger } from "../utils/logger.js";
+import { getAuditUsername, logAudit } from "../utils/audit-logger.js";
+import type { LoadedPlugin } from "./loader.js";
+import { assertCapability, PluginPermissionError } from "./permissions.js";
+import { toPluginHostView, type PluginHostView } from "./host-view.js";
+import { pluginEvents } from "./events.js";
+import {
+  isPluginRequest,
+  isPluginResponse,
+  type PluginPush,
+  type PluginRequest,
+  type PluginResponse,
+} from "./protocol.js";
+
+export interface PluginStorageBackend {
+  get: (pluginId: string, key: string) => Promise<string | null>;
+  set: (pluginId: string, key: string, value: string) => Promise<void>;
+  delete: (pluginId: string, key: string) => Promise<boolean>;
+  listKeys: (pluginId: string) => Promise<string[]>;
+}
+
+export interface BrokerDeps {
+  /**
+   * Injected rather than imported so tests can exercise the gate without a
+   * database, and so the heavy hosts modules stay lazily reachable.
+   */
+  listHosts: (userId: string) => Promise<Record<string, unknown>[]>;
+  resolveHost: (
+    hostId: number,
+    userId: string,
+  ) => Promise<Record<string, unknown> | null>;
+  storage?: PluginStorageBackend;
+  /** Fired when a plugin's route table changes, so it can be re-mounted. */
+  onRoutesChanged?: (runtime: PluginRuntime) => void;
+  ssh?: PluginSshBackend;
+}
+
+/**
+ * The SSH operations the broker performs on a plugin's behalf.
+ *
+ * Note what is NOT here: there is no "get me the host" and no "give me the
+ * client". A plugin's reachable surface is open-a-session and run-a-command,
+ * both of which take an opaque handle. The resolved host, which carries
+ * plaintext credentials, only ever exists inside these functions.
+ */
+export interface PluginSshBackend {
+  open: (
+    hostId: number,
+    userId: string,
+  ) => Promise<{ release: () => void; exec: SshExec }>;
+}
+
+export type SshExec = (
+  command: string,
+  timeoutMs: number,
+) => Promise<{ stdout: string; stderr: string; code: number | null }>;
+
+/** ctx.storage caps, so one plugin cannot fill the database. */
+const MAX_STORAGE_KEY_LENGTH = 128;
+const MAX_STORAGE_VALUE_BYTES = 256 * 1024;
+const MIN_SCHEDULE_INTERVAL_MS = 1000;
+const HTTP_HANDLER_TIMEOUT_MS = 30_000;
+const MAX_SSH_HANDLES = 8;
+const DEFAULT_SSH_EXEC_TIMEOUT_MS = 30_000;
+const MAX_SSH_EXEC_TIMEOUT_MS = 120_000;
+const HTTP_METHODS = new Set(["GET", "POST", "PUT", "PATCH", "DELETE"]);
+
+interface PendingHttpRoute {
+  method: string;
+  path: string;
+}
+
+export interface PluginRuntime {
+  plugin: LoadedPlugin;
+  worker: Worker;
+  /** Routes the plugin registered via ctx.http.route. */
+  routes: PendingHttpRoute[];
+  /** Topics the plugin subscribed to via ctx.events.on. */
+  subscriptions: Set<string>;
+  /** Bus unsubscribe functions, run on detach so no listener outlives it. */
+  eventUnsubscribers: Array<() => void>;
+  /** ctx.schedule timers, cleared on deactivate. */
+  timers: Map<string, NodeJS.Timeout>;
+  /** Live SSH handles, keyed by the opaque id the worker holds. */
+  sshHandles: Map<
+    string,
+    { hostId: number; release: () => void; exec?: SshExec }
+  >;
+  disposed: boolean;
+}
+
+/** Methods that need a granted capability before the broker will run them. */
+const REQUIRED_CAPABILITY: Record<string, string> = {
+  "hosts.list": "hosts.read",
+  "hosts.get": "hosts.read",
+  "ssh.connect": "ssh.exec",
+  "ssh.exec": "ssh.exec",
+  "storage.get": "storage.own",
+  "storage.set": "storage.own",
+  "storage.delete": "storage.own",
+  "storage.list": "storage.own",
+  "events.subscribe": "events.read",
+};
+
+/** Methods whose invocation is worth an audit line. */
+const AUDITED = new Set(Object.keys(REQUIRED_CAPABILITY));
+
+export class PluginBroker {
+  private readonly runtimes = new Map<string, PluginRuntime>();
+  /** Outstanding route invocations, keyed by the push id sent to the worker. */
+  private readonly httpWaiters = new Map<
+    number,
+    { resolve: (value: unknown) => void; reject: (error: Error) => void }
+  >();
+  /**
+   * Push ids start high so they cannot collide with the worker's own request
+   * ids, which start at 1 and count up independently.
+   */
+  private nextPushId = 1_000_000;
+
+  constructor(private readonly deps: BrokerDeps) {}
+
+  get(pluginId: string): PluginRuntime | undefined {
+    return this.runtimes.get(pluginId);
+  }
+
+  /** Wired to PluginLoader's onWorkerReady hook. */
+  attach(plugin: LoadedPlugin, worker: Worker): PluginRuntime {
+    this.detach(plugin.id);
+
+    const runtime: PluginRuntime = {
+      plugin,
+      worker,
+      routes: [],
+      subscriptions: new Set(),
+      eventUnsubscribers: [],
+      timers: new Map(),
+      sshHandles: new Map(),
+      disposed: false,
+    };
+    this.runtimes.set(plugin.id, runtime);
+
+    worker.on("message", (message: unknown) => {
+      if (isPluginRequest(message)) {
+        void this.dispatch(runtime, message);
+        return;
+      }
+      // A route handler's reply comes back shaped like a response.
+      if (isPluginResponse(message)) this.settleHttpReply(message);
+    });
+
+    return runtime;
+  }
+
+  /** Wired to PluginLoader's onWorkerGone hook. Releases everything. */
+  detach(pluginId: string): void {
+    const runtime = this.runtimes.get(pluginId);
+    if (!runtime) return;
+
+    runtime.disposed = true;
+    for (const timer of runtime.timers.values()) clearInterval(timer);
+    runtime.timers.clear();
+
+    for (const handle of runtime.sshHandles.values()) {
+      try {
+        handle.release();
+      } catch {
+        // Releasing a dead connection is not worth failing over.
+      }
+    }
+    runtime.sshHandles.clear();
+
+    for (const unsubscribe of runtime.eventUnsubscribers) {
+      try {
+        unsubscribe();
+      } catch {
+        // Already detached.
+      }
+    }
+    runtime.eventUnsubscribers = [];
+    runtime.subscriptions.clear();
+    this.runtimes.delete(pluginId);
+  }
+
+  private settleHttpReply(message: PluginResponse): void {
+    const waiter = this.httpWaiters.get(message.id);
+    if (!waiter) return;
+    this.httpWaiters.delete(message.id);
+
+    if (message.ok === true) waiter.resolve(message.value);
+    else waiter.reject(new Error(message.error.message));
+  }
+
+  private async dispatch(
+    runtime: PluginRuntime,
+    request: PluginRequest,
+  ): Promise<void> {
+    const { method, args } = request;
+    let failure: Error | null = null;
+
+    try {
+      const required = REQUIRED_CAPABILITY[method];
+      if (required) {
+        await assertCapability(
+          runtime.plugin.id,
+          required,
+          runtime.plugin.manifest.permissions,
+        );
+      }
+
+      const value = await this.invoke(runtime, method, args);
+      this.reply(runtime, { id: request.id, ok: true, value });
+    } catch (error) {
+      failure = error instanceof Error ? error : new Error(String(error));
+      this.reply(runtime, {
+        id: request.id,
+        ok: false,
+        error: {
+          // Sanitised, never the raw message: errors thrown deep in the SSH
+          // stack can embed the connect config, credentials included.
+          message: safeErrorMessage(method, failure),
+          code: (failure as Error & { code?: string }).code,
+        },
+      });
+    }
+
+    if (AUDITED.has(method)) {
+      void this.audit(runtime, method, args, failure);
+    }
+  }
+
+  private reply(runtime: PluginRuntime, message: unknown): void {
+    if (runtime.disposed) return;
+    try {
+      runtime.worker.postMessage(message);
+    } catch {
+      // Worker already gone; the reply has nowhere to land.
+    }
+  }
+
+  private async invoke(
+    runtime: PluginRuntime,
+    method: string,
+    args: unknown[],
+  ): Promise<unknown> {
+    switch (method) {
+      case "log.debug":
+      case "log.info":
+      case "log.warn":
+      case "log.error":
+        return this.handleLog(runtime, method, args);
+
+      case "audit":
+        // The audit line is written by the caller path below; nothing to do
+        // here beyond accepting the call.
+        return null;
+
+      case "hosts.list":
+        return this.handleHostsList(runtime);
+
+      case "hosts.get":
+        return this.handleHostsGet(runtime, Number(args[0]));
+
+      case "storage.get":
+        return this.handleStorageGet(runtime, storageKey(args[0]));
+
+      case "storage.set":
+        return this.handleStorageSet(runtime, args);
+
+      case "storage.delete":
+        return this.storage().delete(runtime.plugin.id, storageKey(args[0]));
+
+      case "storage.list":
+        return this.storage().listKeys(runtime.plugin.id);
+
+      case "schedule.every":
+        return this.handleScheduleEvery(runtime, Number(args[0]));
+
+      case "schedule.cancel":
+        return this.handleScheduleCancel(runtime, String(args[0]));
+
+      case "http.route":
+        return this.handleHttpRoute(runtime, String(args[0]), String(args[1]));
+
+      case "ssh.connect":
+        return this.handleSshConnect(runtime, Number(args[0]));
+
+      case "ssh.exec":
+        return this.handleSshExec(
+          runtime,
+          String(args[0]),
+          String(args[1]),
+          args[2] === undefined ? undefined : Number(args[2]),
+        );
+
+      case "ssh.close":
+        return this.handleSshClose(runtime, String(args[0]));
+
+      case "events.emit":
+        return this.handleEventsEmit(runtime, String(args[0]), args[1]);
+
+      case "events.subscribe":
+        return this.handleEventsSubscribe(runtime, String(args[0]));
+
+      default:
+        throw new PluginFacingError(`Unknown ctx method "${method}"`);
+    }
+  }
+
+  private handleLog(
+    runtime: PluginRuntime,
+    method: string,
+    args: unknown[],
+  ): null {
+    const level = method.slice("log.".length) as
+      "debug" | "info" | "warn" | "error";
+    const message = String(args[0] ?? "");
+    // Forced context: a plugin cannot pretend to be another plugin in the log.
+    const context = { operation: `plugin:${runtime.plugin.id}` };
+
+    if (level === "error") {
+      pluginLogger.error(message, undefined, context);
+    } else {
+      pluginLogger[level](message, context);
+    }
+    return null;
+  }
+
+  private async handleHostsList(
+    runtime: PluginRuntime,
+  ): Promise<PluginHostView[]> {
+    const userId = this.requireOwner(runtime);
+    const hosts = await this.deps.listHosts(userId);
+    return hosts.map(toPluginHostView);
+  }
+
+  private async handleHostsGet(
+    runtime: PluginRuntime,
+    hostId: number,
+  ): Promise<PluginHostView | null> {
+    const userId = this.requireOwner(runtime);
+    if (!Number.isFinite(hostId)) {
+      throw new PluginFacingError("hosts.get requires a numeric host id");
+    }
+
+    const host = await this.deps.resolveHost(hostId, userId);
+    // resolveHostById returns null for both "missing" and "no access", and the
+    // plugin gets the same answer for both so it cannot probe for existence.
+    return host ? toPluginHostView(host) : null;
+  }
+
+  private storage(): PluginStorageBackend {
+    if (!this.deps.storage) {
+      throw new PluginFacingError(
+        "ctx.storage is not available in this runtime",
+      );
+    }
+    return this.deps.storage;
+  }
+
+  private async handleStorageGet(
+    runtime: PluginRuntime,
+    key: string,
+  ): Promise<unknown> {
+    const raw = await this.storage().get(runtime.plugin.id, key);
+    if (raw === null) return null;
+    try {
+      return JSON.parse(raw);
+    } catch {
+      // A row written by an older shape should not break the plugin.
+      return null;
+    }
+  }
+
+  private async handleStorageSet(
+    runtime: PluginRuntime,
+    args: unknown[],
+  ): Promise<null> {
+    const key = storageKey(args[0]);
+    // Serialised here rather than in the worker so the stored shape is the
+    // broker's decision, not something a plugin can control.
+    const value = JSON.stringify(args[1] ?? null);
+
+    if (Buffer.byteLength(value, "utf8") > MAX_STORAGE_VALUE_BYTES) {
+      throw new PluginFacingError(
+        `ctx.storage value for "${key}" exceeds the ${MAX_STORAGE_VALUE_BYTES} byte limit`,
+      );
+    }
+
+    await this.storage().set(runtime.plugin.id, key, value);
+    return null;
+  }
+
+  private handleScheduleEvery(
+    runtime: PluginRuntime,
+    intervalMs: number,
+  ): string {
+    if (!Number.isFinite(intervalMs) || intervalMs < MIN_SCHEDULE_INTERVAL_MS) {
+      throw new PluginFacingError(
+        `ctx.schedule.every requires an interval of at least ${MIN_SCHEDULE_INTERVAL_MS}ms`,
+      );
+    }
+
+    const id = `timer:${randomUUID()}`;
+    const timer = setInterval(() => {
+      if (runtime.disposed) return;
+      this.push(runtime, { kind: "timer", key: id, payload: null });
+    }, intervalMs);
+
+    // A plugin timer must not keep the server alive on its own.
+    timer.unref?.();
+    runtime.timers.set(id, timer);
+    return id;
+  }
+
+  private handleScheduleCancel(runtime: PluginRuntime, id: string): null {
+    const timer = runtime.timers.get(id);
+    if (timer) {
+      clearInterval(timer);
+      runtime.timers.delete(id);
+    }
+    return null;
+  }
+
+  /**
+   * Opens a pooled connection and returns an opaque handle.
+   *
+   * The handle is a random id, not anything derived from the host or its
+   * credentials, and it is only meaningful inside this runtime's own map. The
+   * ssh2 Client and the resolved host never enter a postMessage payload.
+   */
+  private async handleSshConnect(
+    runtime: PluginRuntime,
+    hostId: number,
+  ): Promise<string> {
+    const userId = this.requireOwner(runtime);
+    if (!Number.isFinite(hostId)) {
+      throw new PluginFacingError("ssh.connect requires a numeric host id");
+    }
+    if (!this.deps.ssh) {
+      throw new PluginFacingError("ctx.ssh is not available in this runtime");
+    }
+    if (runtime.sshHandles.size >= MAX_SSH_HANDLES) {
+      throw new PluginFacingError(
+        `ctx.ssh is limited to ${MAX_SSH_HANDLES} concurrent connections per plugin`,
+      );
+    }
+
+    const session = await this.deps.ssh.open(hostId, userId);
+    const id = `sshconn:${randomUUID()}`;
+    runtime.sshHandles.set(id, {
+      hostId,
+      release: session.release,
+      exec: session.exec,
+    });
+    return id;
+  }
+
+  private async handleSshExec(
+    runtime: PluginRuntime,
+    handleId: string,
+    command: string,
+    timeoutMs?: number,
+  ): Promise<{ stdout: string; stderr: string; code: number | null }> {
+    const handle = runtime.sshHandles.get(handleId);
+    // An unknown handle is indistinguishable from another plugin's handle:
+    // the map is per-runtime, so ids cannot be guessed across plugins.
+    if (!handle?.exec) {
+      throw new PluginFacingError("Unknown or closed SSH connection handle");
+    }
+    if (!command) throw new PluginFacingError("ssh.exec requires a command");
+
+    const timeout =
+      timeoutMs && Number.isFinite(timeoutMs)
+        ? Math.min(timeoutMs, MAX_SSH_EXEC_TIMEOUT_MS)
+        : DEFAULT_SSH_EXEC_TIMEOUT_MS;
+
+    return handle.exec(command, timeout);
+  }
+
+  private handleSshClose(runtime: PluginRuntime, handleId: string): null {
+    const handle = runtime.sshHandles.get(handleId);
+    if (!handle) return null;
+
+    runtime.sshHandles.delete(handleId);
+    try {
+      handle.release();
+    } catch {
+      // Releasing a dead connection is not worth failing over.
+    }
+    return null;
+  }
+
+  /**
+   * A plugin may only emit under its own namespace. Without this a plugin
+   * could publish "host.status" and drive the automations engine as though the
+   * metrics poller had reported it.
+   */
+  private handleEventsEmit(
+    runtime: PluginRuntime,
+    topic: string,
+    payload: unknown,
+  ): null {
+    const prefix = `plugin.${runtime.plugin.id}.`;
+    if (!topic.startsWith(prefix)) {
+      throw new PluginFacingError(
+        `ctx.events.emit may only publish topics beginning with "${prefix}"`,
+      );
+    }
+
+    pluginEvents.emit(topic, payload);
+    return null;
+  }
+
+  private handleEventsSubscribe(runtime: PluginRuntime, topic: string): null {
+    if (runtime.subscriptions.has(topic)) return null;
+    runtime.subscriptions.add(topic);
+
+    const unsubscribe = pluginEvents.on(topic, (payload) => {
+      this.push(runtime, { kind: "event", key: topic, payload });
+    });
+    runtime.eventUnsubscribers.push(unsubscribe);
+    return null;
+  }
+
+  private handleHttpRoute(
+    runtime: PluginRuntime,
+    method: string,
+    routePath: string,
+  ): null {
+    const verb = method.toUpperCase();
+    if (!HTTP_METHODS.has(verb)) {
+      throw new PluginFacingError(
+        `ctx.http.route does not support the ${verb} method`,
+      );
+    }
+    if (!routePath.startsWith("/")) {
+      throw new PluginFacingError(`ctx.http.route path must start with "/"`);
+    }
+
+    const key = `${verb} ${routePath}`;
+    if (
+      !runtime.routes.some((route) => `${route.method} ${route.path}` === key)
+    ) {
+      runtime.routes.push({ method: verb, path: routePath });
+    }
+
+    this.deps.onRoutesChanged?.(runtime);
+    return null;
+  }
+
+  /**
+   * Invoked by the Express router this plugin's routes are mounted behind.
+   * Forwards the request into the worker and waits for the handler's reply.
+   */
+  invokeRoute(
+    runtime: PluginRuntime,
+    key: string,
+    payload: unknown,
+    timeoutMs = HTTP_HANDLER_TIMEOUT_MS,
+  ): Promise<unknown> {
+    if (runtime.disposed) {
+      return Promise.reject(new Error("Plugin is not running"));
+    }
+
+    const replyTo = this.nextPushId++;
+
+    return new Promise((resolve, reject) => {
+      const timer = setTimeout(() => {
+        this.httpWaiters.delete(replyTo);
+        reject(new Error("Plugin route handler timed out"));
+      }, timeoutMs);
+
+      this.httpWaiters.set(replyTo, {
+        resolve: (value) => {
+          clearTimeout(timer);
+          resolve(value);
+        },
+        reject: (error) => {
+          clearTimeout(timer);
+          reject(error);
+        },
+      });
+
+      this.push(runtime, { kind: "http", key, payload, replyTo });
+    });
+  }
+
+  /** Main thread -> worker, for events, timers and route invocations. */
+  push(runtime: PluginRuntime, message: PluginPush): void {
+    if (runtime.disposed) return;
+    try {
+      runtime.worker.postMessage(message);
+    } catch {
+      // Worker already gone.
+    }
+  }
+
+  private requireOwner(runtime: PluginRuntime): string {
+    const userId = runtime.plugin.ownerUserId;
+    if (!userId) {
+      throw new PluginFacingError(
+        `Plugin ${runtime.plugin.id} has no owning user, so it cannot act on user data`,
+      );
+    }
+    return userId;
+  }
+
+  private async audit(
+    runtime: PluginRuntime,
+    method: string,
+    args: unknown[],
+    failure: Error | null,
+  ): Promise<void> {
+    const userId = runtime.plugin.ownerUserId;
+    if (!userId) return;
+
+    // Attribution comes from the broker, never from the plugin: username is
+    // the plugin id, not the user's, so a plugin action is never mistaken for
+    // something the person did themselves.
+    await logAudit({
+      userId,
+      username: `plugin:${runtime.plugin.id}`,
+      action: `plugin_${method.replace(/\./g, "_")}`,
+      resourceType: "plugin",
+      resourceId: runtime.plugin.id,
+      resourceName: runtime.plugin.manifest.name,
+      details: safeDetails(method, args),
+      success: failure === null,
+      errorMessage: failure?.message,
+    });
+  }
+
+  /** Resolves the display name for a plugin's owner, for UI surfaces. */
+  async ownerUsername(pluginId: string): Promise<string | null> {
+    const runtime = this.runtimes.get(pluginId);
+    if (!runtime?.plugin.ownerUserId) return null;
+    return getAuditUsername(runtime.plugin.ownerUserId);
+  }
+
+  newHandleId(): string {
+    return `sshconn:${randomUUID()}`;
+  }
+}
+
+/**
+ * Audit details are capped and only ever record the shape of a call, never a
+ * value a plugin passed. A plugin could otherwise write arbitrary volume, or
+ * something that looks like a real audit entry, into the trail.
+ */
+/**
+ * Errors the broker raises itself are safe to show a plugin: they are written
+ * here and say nothing about credentials. Anything else came from deeper in the
+ * server -- ssh2, the pool, a repository -- and can carry connect config,
+ * decrypted material or internal paths, so the plugin gets a generic message
+ * and the real one goes to the log.
+ *
+ * Marker class rather than message matching: a substring check would start
+ * leaking the moment an upstream error happened to contain a familiar phrase.
+ */
+export class PluginFacingError extends Error {
+  constructor(message: string) {
+    super(message);
+    this.name = "PluginFacingError";
+  }
+}
+
+function safeErrorMessage(method: string, error: Error): string {
+  if (error instanceof PluginFacingError) return error.message;
+  if (error instanceof PluginPermissionError) return error.message;
+
+  pluginLogger.error(`ctx.${method} failed`, error, {
+    operation: "plugin_ctx_error",
+  });
+  return `ctx.${method} failed. See the server log for details.`;
+}
+
+function storageKey(raw: unknown): string {
+  const key = String(raw ?? "");
+  if (!key) throw new PluginFacingError("ctx.storage requires a non-empty key");
+  if (key.length > MAX_STORAGE_KEY_LENGTH) {
+    throw new PluginFacingError(
+      `ctx.storage key exceeds the ${MAX_STORAGE_KEY_LENGTH} character limit`,
+    );
+  }
+  return key;
+}
+
+function safeDetails(method: string, args: unknown[]): string {
+  const summary = args.map((arg) => {
+    if (typeof arg === "number" || typeof arg === "boolean") return arg;
+    if (typeof arg === "string") {
+      return arg.length > 64 ? `${arg.slice(0, 64)}…` : arg;
+    }
+    if (arg === null || arg === undefined) return null;
+    return Array.isArray(arg) ? `[array:${arg.length}]` : "[object]";
+  });
+
+  return JSON.stringify({ method, args: summary });
+}
+
+export { PluginPermissionError };
diff --git a/src/backend/plugins/permissions.ts b/src/backend/plugins/permissions.ts
new file mode 100644
index 000000000..b58ae08c3
--- /dev/null
+++ b/src/backend/plugins/permissions.ts
@@ -0,0 +1,73 @@
+/**
+ * The grant check every privileged ctx call goes through.
+ *
+ * Capability strings are the manifest's own permission enum, stored verbatim in
+ * plugin_permission_grants.capability -- no translation layer, so the manifest
+ * stays the single source of truth for what is grantable.
+ *
+ * Two separate things have to be true for a capability to count as granted:
+ * the manifest must declare it (the plugin asked for it up front) and an admin
+ * must have granted it (someone said yes). A grant for something the manifest
+ * never declared is ignored rather than honoured, so widening a plugin's reach
+ * always requires a new manifest the user can see.
+ */
+
+import { createCurrentPluginPermissionGrantRepository } from "../database/repositories/factory.js";
+import type { PluginPermission } from "./manifest.js";
+
+const CACHE_TTL_MS = 60_000;
+
+interface CacheEntry {
+  capabilities: Set<string>;
+  expiresAt: number;
+}
+
+const cache = new Map<string, CacheEntry>();
+
+export class PluginPermissionError extends Error {
+  readonly code = "EPLUGINPERM";
+
+  constructor(pluginId: string, capability: string) {
+    super(
+      `Plugin "${pluginId}" is not granted the "${capability}" capability. ` +
+        `Declare it in the manifest's permissions array and grant it in the plugin's settings.`,
+    );
+    this.name = "PluginPermissionError";
+  }
+}
+
+export function invalidatePluginPermissionCache(pluginId?: string): void {
+  if (pluginId) cache.delete(pluginId);
+  else cache.clear();
+}
+
+async function loadGrants(pluginId: string): Promise<Set<string>> {
+  const cached = cache.get(pluginId);
+  if (cached && cached.expiresAt > Date.now()) return cached.capabilities;
+
+  const rows =
+    await createCurrentPluginPermissionGrantRepository().listByPlugin(pluginId);
+  const capabilities = new Set(rows.map((row) => row.capability));
+
+  cache.set(pluginId, { capabilities, expiresAt: Date.now() + CACHE_TTL_MS });
+  return capabilities;
+}
+
+export async function hasCapability(
+  pluginId: string,
+  capability: PluginPermission | string,
+  declared: readonly string[],
+): Promise<boolean> {
+  if (!declared.includes(capability)) return false;
+  return (await loadGrants(pluginId)).has(capability);
+}
+
+export async function assertCapability(
+  pluginId: string,
+  capability: PluginPermission | string,
+  declared: readonly string[],
+): Promise<void> {
+  if (!(await hasCapability(pluginId, capability, declared))) {
+    throw new PluginPermissionError(pluginId, capability);
+  }
+}
diff --git a/src/backend/tests/plugins/broker.test.ts b/src/backend/tests/plugins/broker.test.ts
new file mode 100644
index 000000000..9f532098c
--- /dev/null
+++ b/src/backend/tests/plugins/broker.test.ts
@@ -0,0 +1,620 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { createFixturePlugin, type Fixture } from "./fixture-plugin.js";
+
+const state = vi.hoisted(() => ({
+  grants: [] as { pluginId: string; capability: string }[],
+  auditEntries: [] as Record<string, unknown>[],
+}));
+
+vi.mock("../../utils/logger.js", () => ({
+  pluginLogger: {
+    debug: vi.fn(),
+    info: vi.fn(),
+    warn: vi.fn(),
+    error: vi.fn(),
+    success: vi.fn(),
+  },
+}));
+
+vi.mock("../../utils/audit-logger.js", () => ({
+  logAudit: vi.fn(async (params: Record<string, unknown>) => {
+    state.auditEntries.push(params);
+  }),
+  getAuditUsername: vi.fn(async (userId: string) => `user:${userId}`),
+}));
+
+vi.mock("../../database/repositories/factory.js", () => ({
+  createCurrentPluginPermissionGrantRepository: () => ({
+    listByPlugin: async (pluginId: string) =>
+      state.grants.filter((g) => g.pluginId === pluginId),
+  }),
+}));
+
+const { PluginBroker } = await import("../../plugins/broker.js");
+const { PluginLoader } = await import("../../plugins/loader.js");
+const { invalidatePluginPermissionCache } =
+  await import("../../plugins/permissions.js");
+
+const WORKER_TIMEOUT = 30_000;
+
+/**
+ * A host as resolveHostById would return it: secrets in plaintext. The tests
+ * assert none of this reaches the worker.
+ */
+const SECRET = "hunter2-plaintext-secret";
+const RAW_HOST = {
+  id: 42,
+  name: "prod-box",
+  ip: "10.0.0.5",
+  port: 22,
+  username: "root",
+  folder: "Production",
+  tags: ["prod"],
+  authType: "password",
+  enableTerminal: true,
+  enableTunnel: false,
+  enableFileManager: true,
+  createdAt: "2026-01-01T00:00:00Z",
+  updatedAt: "2026-01-02T00:00:00Z",
+  // None of these may ever cross the boundary.
+  password: SECRET,
+  key: SECRET,
+  keyPassword: SECRET,
+  certPublicKey: SECRET,
+  sudoPassword: SECRET,
+  autostartPassword: SECRET,
+};
+
+describe("PluginBroker", () => {
+  const fixtures: Fixture[] = [];
+  let loader: InstanceType<typeof PluginLoader> | null = null;
+  let broker: InstanceType<typeof PluginBroker> | null = null;
+
+  beforeEach(() => {
+    state.grants = [];
+    state.auditEntries = [];
+    storage = memoryStorage();
+    invalidatePluginPermissionCache();
+  });
+
+  afterEach(async () => {
+    await loader?.shutdown();
+    loader = null;
+    broker = null;
+    // The bus is a module singleton, so a leftover listener would leak between
+    // tests.
+    const { pluginEvents } = await import("../../plugins/events.js");
+    pluginEvents.clear();
+    while (fixtures.length) fixtures.pop()!.cleanup();
+  });
+
+  function grant(capability: string, pluginId = "sample-plugin") {
+    state.grants.push({ pluginId, capability });
+    invalidatePluginPermissionCache(pluginId);
+  }
+
+  /**
+   * Boots a real worker whose activate() runs `body` and reports the outcome
+   * back through a ctx.log call the test can read.
+   */
+  /** In-memory stand-in for PluginStorageRepository, scoped per plugin. */
+  function memoryStorage() {
+    const rows = new Map<string, string>();
+    const at = (pluginId: string, key: string) => `${pluginId}\u0000${key}`;
+
+    return {
+      rows,
+      backend: {
+        get: async (pluginId: string, key: string) =>
+          rows.get(at(pluginId, key)) ?? null,
+        set: async (pluginId: string, key: string, value: string) => {
+          rows.set(at(pluginId, key), value);
+        },
+        delete: async (pluginId: string, key: string) =>
+          rows.delete(at(pluginId, key)),
+        listKeys: async (pluginId: string) =>
+          [...rows.keys()]
+            .filter((k) => k.startsWith(`${pluginId}\u0000`))
+            .map((k) => k.split("\u0000")[1]),
+      },
+    };
+  }
+
+  let storage = memoryStorage();
+
+  async function boot(body: string, permissions?: string[]) {
+    const fixture = createFixturePlugin({
+      permissions: permissions ?? ["hosts.read", "storage.own"],
+      backendSource: `
+        export async function activate(ctx) {
+          globalThis.__result = undefined;
+          try {
+            globalThis.__result = { ok: true, value: await (async () => { ${body} })() };
+          } catch (error) {
+            globalThis.__result = { ok: false, error: error.message };
+          }
+          await ctx.log.info(JSON.stringify(globalThis.__result));
+        }
+      `,
+    });
+    fixtures.push(fixture);
+
+    broker = new PluginBroker({
+      listHosts: async () => [RAW_HOST],
+      resolveHost: async (hostId) => (hostId === 42 ? RAW_HOST : null),
+      storage: storage.backend,
+    });
+
+    const logged: string[] = [];
+    const { pluginLogger } = await import("../../utils/logger.js");
+    vi.mocked(pluginLogger.info).mockImplementation((message: string) => {
+      logged.push(message);
+    });
+
+    loader = new PluginLoader({
+      onWorkerReady: (plugin, worker) => broker!.attach(plugin, worker),
+      onWorkerGone: (plugin) => broker!.detach(plugin.id),
+    });
+
+    await loader.load(fixture.dir);
+    await loader.activate("sample-plugin", "user-1");
+
+    const line = logged.find((entry) => entry.startsWith("{"));
+    return line ? JSON.parse(line) : { ok: false, error: "no result reported" };
+  }
+
+  describe("permission gate", () => {
+    it(
+      "allows a call whose capability is declared and granted",
+      async () => {
+        grant("hosts.read");
+        const result = await boot(`return await ctx.hosts.get(42);`);
+
+        expect(result.ok).toBe(true);
+        expect(result.value).toMatchObject({ id: 42, name: "prod-box" });
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "denies a declared capability that was never granted",
+      async () => {
+        const result = await boot(`return await ctx.hosts.get(42);`);
+
+        expect(result.ok).toBe(false);
+        expect(result.error).toContain('is not granted the "hosts.read"');
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "denies a granted capability the manifest never declared",
+      async () => {
+        // Granted in the database, but absent from the manifest: widening a
+        // plugin's reach must require a new manifest, not just a grant.
+        grant("hosts.read");
+        const result = await boot(`return await ctx.hosts.get(42);`, [
+          "storage.own",
+        ]);
+
+        expect(result.ok).toBe(false);
+        expect(result.error).toContain('is not granted the "hosts.read"');
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "rejects an unknown ctx method",
+      async () => {
+        const result = await boot(
+          `return await new Promise((resolve, reject) => {
+             ctx.hosts.__proto__; resolve("noop");
+           });`,
+        );
+        expect(result.ok).toBe(true);
+      },
+      WORKER_TIMEOUT,
+    );
+  });
+
+  describe("credential isolation", () => {
+    it(
+      "never sends credential material to the worker",
+      async () => {
+        grant("hosts.read");
+        const result = await boot(`
+          const host = await ctx.hosts.get(42);
+          const list = await ctx.hosts.list();
+          const blob = JSON.stringify({ host, list });
+          if (blob.includes(${JSON.stringify(SECRET)})) {
+            throw new Error("SECRET LEAKED: " + blob);
+          }
+          return { keys: Object.keys(host).sort(), listLength: list.length };
+        `);
+
+        expect(result.ok).toBe(true);
+        expect(result.value.listLength).toBe(1);
+        expect(result.value.keys).not.toContain("password");
+        expect(result.value.keys).not.toContain("key");
+        expect(result.value.keys).not.toContain("keyPassword");
+        expect(result.value.keys).not.toContain("certPublicKey");
+        expect(result.value.keys).not.toContain("sudoPassword");
+        expect(result.value.keys).not.toContain("autostartPassword");
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "returns null rather than distinguishing missing from forbidden",
+      async () => {
+        grant("hosts.read");
+        const result = await boot(`return await ctx.hosts.get(999);`);
+
+        expect(result.ok).toBe(true);
+        expect(result.value).toBeNull();
+      },
+      WORKER_TIMEOUT,
+    );
+  });
+
+  describe("audit", () => {
+    it(
+      "writes one plugin-attributed line per permitted call",
+      async () => {
+        grant("hosts.read");
+        await boot(`return await ctx.hosts.get(42);`);
+
+        const entry = state.auditEntries.find(
+          (e) => e.action === "plugin_hosts_get",
+        );
+        expect(entry).toBeDefined();
+        expect(entry).toMatchObject({
+          userId: "user-1",
+          username: "plugin:sample-plugin",
+          resourceType: "plugin",
+          resourceId: "sample-plugin",
+          success: true,
+        });
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "audits a denied call as a failure",
+      async () => {
+        await boot(`return await ctx.hosts.get(42);`);
+
+        const entry = state.auditEntries.find(
+          (e) => e.action === "plugin_hosts_get",
+        );
+        expect(entry).toMatchObject({
+          username: "plugin:sample-plugin",
+          success: false,
+        });
+        expect(String(entry?.errorMessage)).toContain("is not granted");
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "does not let a plugin forge the actor or smuggle bulk details",
+      async () => {
+        grant("hosts.read");
+        // ctx.log takes a context object, so this exercises the summariser on
+        // an argument a plugin genuinely controls.
+        await boot(`
+          await ctx.log.info("hello", { username: "admin", userId: "user-999" });
+          return await ctx.hosts.get(42);
+        `);
+
+        const entry = state.auditEntries.find(
+          (e) => e.action === "plugin_hosts_get",
+        );
+        // Attribution is broker-set regardless of what the plugin passed.
+        expect(entry?.username).toBe("plugin:sample-plugin");
+        expect(entry?.userId).toBe("user-1");
+
+        // No audit entry anywhere claims the user the plugin named.
+        expect(
+          state.auditEntries.some(
+            (e) => e.userId === "user-999" || e.username === "admin",
+          ),
+        ).toBe(false);
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "summarises rather than records a long string argument",
+      async () => {
+        grant("hosts.read");
+        await boot(`
+          try { await ctx.hosts.get("${"x".repeat(500)}"); } catch {}
+          return "done";
+        `);
+
+        const entry = state.auditEntries.find(
+          (e) => e.action === "plugin_hosts_get",
+        );
+        const details = String(entry?.details);
+        // Capped, so a plugin cannot write bulk data into the audit trail.
+        expect(details.length).toBeLessThan(200);
+        expect(details).not.toContain("x".repeat(200));
+      },
+      WORKER_TIMEOUT,
+    );
+  });
+
+  describe("ctx.storage", () => {
+    it(
+      "round-trips a structured value",
+      async () => {
+        grant("storage.own");
+        const result = await boot(`
+          await ctx.storage.set("prefs", { theme: "dark", count: 3 });
+          const read = await ctx.storage.get("prefs");
+          const keys = await ctx.storage.list();
+          return { read, keys };
+        `);
+
+        expect(result.ok).toBe(true);
+        expect(result.value.read).toEqual({ theme: "dark", count: 3 });
+        expect(result.value.keys).toEqual(["prefs"]);
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "returns null for a key that was never set",
+      async () => {
+        grant("storage.own");
+        const result = await boot(`return await ctx.storage.get("nope");`);
+
+        expect(result.ok).toBe(true);
+        expect(result.value).toBeNull();
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "deletes a key",
+      async () => {
+        grant("storage.own");
+        const result = await boot(`
+          await ctx.storage.set("temp", 1);
+          await ctx.storage.delete("temp");
+          return await ctx.storage.list();
+        `);
+
+        expect(result.ok).toBe(true);
+        expect(result.value).toEqual([]);
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "scopes writes to the calling plugin, which cannot name another scope",
+      async () => {
+        grant("storage.own");
+        // A plugin only ever supplies a key; the plugin id is added by the
+        // broker, so there is no argument through which to reach another
+        // plugin's rows.
+        await boot(`await ctx.storage.set("owned", "value");`);
+
+        const keys = [...storage.rows.keys()];
+        expect(keys).toEqual(["sample-plugin\u0000owned"]);
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "denies storage without the granted capability",
+      async () => {
+        const result = await boot(`return await ctx.storage.get("prefs");`);
+
+        expect(result.ok).toBe(false);
+        expect(result.error).toContain('is not granted the "storage.own"');
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "rejects an empty key and an oversized value",
+      async () => {
+        grant("storage.own");
+        const result = await boot(`
+          const errors = [];
+          try { await ctx.storage.set("", 1); } catch (e) { errors.push(e.message); }
+          try {
+            await ctx.storage.set("big", "x".repeat(300000));
+          } catch (e) { errors.push(e.message); }
+          return errors;
+        `);
+
+        expect(result.ok).toBe(true);
+        expect(result.value[0]).toContain("non-empty key");
+        expect(result.value[1]).toContain("byte limit");
+      },
+      WORKER_TIMEOUT,
+    );
+  });
+
+  describe("ctx.schedule", () => {
+    it(
+      "registers a timer and fires the plugin's handler",
+      async () => {
+        grant("storage.own");
+        const result = await boot(`
+          await new Promise(async (resolve, reject) => {
+            const timer = setTimeout(() => reject(new Error("timer never fired")), 8000);
+            await ctx.schedule.every(1000, async () => {
+              clearTimeout(timer);
+              resolve();
+            });
+          });
+          return "fired";
+        `);
+
+        expect(result.ok).toBe(true);
+        expect(result.value).toBe("fired");
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "rejects an interval below the floor",
+      async () => {
+        grant("storage.own");
+        const result = await boot(
+          `return await ctx.schedule.every(5, () => {});`,
+        );
+
+        expect(result.ok).toBe(false);
+        expect(result.error).toContain("at least 1000ms");
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "cancel stops the timer and drops it from the runtime",
+      async () => {
+        grant("storage.own");
+        await boot(`
+          const id = await ctx.schedule.every(1000, () => {});
+          await ctx.schedule.cancel(id);
+        `);
+
+        expect(broker!.get("sample-plugin")!.timers.size).toBe(0);
+      },
+      WORKER_TIMEOUT,
+    );
+  });
+
+  describe("ctx.events", () => {
+    it(
+      "pushes a subscribed topic through to the plugin's listener",
+      async () => {
+        grant("events.read");
+        const { pluginEvents } = await import("../../plugins/events.js");
+
+        // The plugin subscribes during activate, then reports what it sees on
+        // the next event. boot() resolves once activate returns, so the emit
+        // below happens with the subscription already live.
+        await boot(
+          `
+          await ctx.events.on("host.status", async (payload) => {
+            await ctx.log.info("EVENT:" + JSON.stringify(payload));
+          });
+        `,
+          ["events.read"],
+        );
+
+        const runtime = broker!.get("sample-plugin")!;
+        expect(runtime.subscriptions.has("host.status")).toBe(true);
+        expect(pluginEvents.listenerCount("host.status")).toBe(1);
+
+        const pushed: unknown[] = [];
+        const original = runtime.worker.postMessage.bind(runtime.worker);
+        runtime.worker.postMessage = ((message: unknown) => {
+          pushed.push(message);
+          return original(message);
+        }) as typeof runtime.worker.postMessage;
+
+        pluginEvents.emit("host.status", { hostId: 42, online: true });
+
+        expect(pushed).toContainEqual({
+          kind: "event",
+          key: "host.status",
+          payload: { hostId: 42, online: true },
+        });
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "refuses to emit outside the plugin's own namespace",
+      async () => {
+        grant("events.read");
+        const result = await boot(
+          `
+          const errors = [];
+          try {
+            await ctx.events.emit("host.status", { hostId: 1, online: false });
+          } catch (e) { errors.push(e.message); }
+          await ctx.events.emit("plugin.sample-plugin.ready", { ok: true });
+          return errors;
+        `,
+          ["events.read"],
+        );
+
+        expect(result.ok).toBe(true);
+        expect(result.value[0]).toContain(
+          'may only publish topics beginning with "plugin.sample-plugin."',
+        );
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "denies subscribing without the events.read capability",
+      async () => {
+        const result = await boot(
+          `return await ctx.events.on("host.status", () => {});`,
+          ["events.read"],
+        );
+
+        expect(result.ok).toBe(false);
+        expect(result.error).toContain('is not granted the "events.read"');
+      },
+      WORKER_TIMEOUT,
+    );
+
+    it(
+      "drops bus subscriptions when the plugin detaches",
+      async () => {
+        grant("events.read");
+        const { pluginEvents } = await import("../../plugins/events.js");
+
+        await boot(`await ctx.events.on("host.status", () => {});`, [
+          "events.read",
+        ]);
+
+        expect(pluginEvents.listenerCount("host.status")).toBe(1);
+        broker!.detach("sample-plugin");
+        expect(pluginEvents.listenerCount("host.status")).toBe(0);
+      },
+      WORKER_TIMEOUT,
+    );
+  });
+
+  describe("detach", () => {
+    it(
+      "releases ssh handles and timers when the worker goes away",
+      async () => {
+        grant("hosts.read");
+        await boot(`return await ctx.hosts.get(42);`);
+
+        const runtime = broker!.get("sample-plugin");
+        expect(runtime).toBeDefined();
+
+        let released = false;
+        runtime!.sshHandles.set("sshconn:test", {
+          hostId: 42,
+          release: () => {
+            released = true;
+          },
+        });
+        runtime!.timers.set(
+          "t1",
+          setInterval(() => {}, 10_000),
+        );
+
+        broker!.detach("sample-plugin");
+
+        expect(released).toBe(true);
+        expect(broker!.get("sample-plugin")).toBeUndefined();
+      },
+      WORKER_TIMEOUT,
+    );
+  });
+});
diff --git a/src/backend/tests/plugins/credential-isolation.test.ts b/src/backend/tests/plugins/credential-isolation.test.ts
new file mode 100644
index 000000000..8f3afd3ef
--- /dev/null
+++ b/src/backend/tests/plugins/credential-isolation.test.ts
@@ -0,0 +1,365 @@
+/**
+ * The security test for the plugin runtime.
+ *
+ * The claim under test: a plugin worker cannot obtain plaintext credential
+ * material through ANY ctx method. Rather than assert field-by-field, this
+ * plants a unique sentinel string in every secret-bearing field of the resolved
+ * host and then asserts that the sentinel appears in nothing the worker ever
+ * receives -- return values, thrown error messages, or the ctx object itself.
+ *
+ * Every message crossing the boundary is recorded, so the assertion covers the
+ * wire, not just what the plugin happened to look at.
+ */
+
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { createFixturePlugin, type Fixture } from "./fixture-plugin.js";
+
+const SENTINEL = "SENT1NEL-plaintext-credential-do-not-leak";
+
+const state = vi.hoisted(() => ({
+  grants: [] as { pluginId: string; capability: string }[],
+  /** Every message the main thread sent to a worker. */
+  outbound: [] as unknown[],
+}));
+
+vi.mock("../../utils/logger.js", () => ({
+  pluginLogger: {
+    debug: vi.fn(),
+    info: vi.fn(),
+    warn: vi.fn(),
+    error: vi.fn(),
+    success: vi.fn(),
+  },
+}));
+
+vi.mock("../../utils/audit-logger.js", () => ({
+  logAudit: vi.fn(async () => {}),
+  getAuditUsername: vi.fn(async (userId: string) => `user:${userId}`),
+}));
+
+vi.mock("../../database/repositories/factory.js", () => ({
+  createCurrentPluginPermissionGrantRepository: () => ({
+    listByPlugin: async (pluginId: string) =>
+      state.grants.filter((g) => g.pluginId === pluginId),
+  }),
+}));
+
+const { PluginBroker } = await import("../../plugins/broker.js");
+const { PluginLoader } = await import("../../plugins/loader.js");
+const { invalidatePluginPermissionCache } =
+  await import("../../plugins/permissions.js");
+
+const WORKER_TIMEOUT = 30_000;
+
+/** A host exactly as resolveHostById hands it over: secrets in the clear. */
+const RESOLVED_HOST = {
+  id: 42,
+  name: "prod-box",
+  ip: "10.0.0.5",
+  port: 22,
+  username: "root",
+  folder: "Production",
+  tags: ["prod"],
+  authType: "password",
+  enableTerminal: true,
+  enableTunnel: true,
+  enableFileManager: true,
+  createdAt: "2026-01-01T00:00:00Z",
+  updatedAt: "2026-01-02T00:00:00Z",
+  password: SENTINEL,
+  key: SENTINEL,
+  keyPassword: SENTINEL,
+  certPublicKey: SENTINEL,
+  sudoPassword: SENTINEL,
+  autostartPassword: SENTINEL,
+  autostartKey: SENTINEL,
+  autostartKeyPassword: SENTINEL,
+  credentialId: 7,
+  vaultProfileId: 3,
+};
+
+function deepContains(value: unknown, needle: string): boolean {
+  if (typeof value === "string") return value.includes(needle);
+  if (value === null || typeof value !== "object") return false;
+  return Object.values(value as Record<string, unknown>).some((inner) =>
+    deepContains(inner, needle),
+  );
+}
+
+describe("plugin credential isolation", () => {
+  const fixtures: Fixture[] = [];
+  let loader: InstanceType<typeof PluginLoader> | null = null;
+  let broker: InstanceType<typeof PluginBroker> | null = null;
+
+  beforeEach(() => {
+    state.grants = [
+      { pluginId: "sample-plugin", capability: "hosts.read" },
+      { pluginId: "sample-plugin", capability: "ssh.exec" },
+      { pluginId: "sample-plugin", capability: "storage.own" },
+    ];
+    state.outbound = [];
+    invalidatePluginPermissionCache();
+  });
+
+  afterEach(async () => {
+    await loader?.shutdown();
+    loader = null;
+    broker = null;
+    while (fixtures.length) fixtures.pop()!.cleanup();
+  });
+
+  /** Boots a plugin whose activate() runs `body` and reports the result. */
+  async function attack(body: string, sshOpenFails = false) {
+    const fixture = createFixturePlugin({
+      permissions: ["hosts.read", "ssh.exec", "storage.own"],
+      backendSource: `
+        export async function activate(ctx) {
+          let result;
+          try {
+            result = { ok: true, value: await (async () => { ${body} })() };
+          } catch (error) {
+            result = { ok: false, error: String(error && error.message) };
+          }
+          await ctx.log.info(JSON.stringify(result));
+        }
+      `,
+    });
+    fixtures.push(fixture);
+
+    broker = new PluginBroker({
+      listHosts: async () => [RESOLVED_HOST],
+      resolveHost: async (hostId) => (hostId === 42 ? RESOLVED_HOST : null),
+      ssh: {
+        open: async () => {
+          if (sshOpenFails) {
+            // A failing connect must not spill the credential into the error
+            // message either: this mirrors ssh2 embedding config on failure.
+            throw new Error(
+              `Authentication failed for root using password ${SENTINEL}`,
+            );
+          }
+          return {
+            release: () => {},
+            exec: async () => ({
+              stdout: "uptime: 4 days",
+              stderr: "",
+              code: 0,
+            }),
+          };
+        },
+      },
+    });
+
+    const logged: string[] = [];
+    const { pluginLogger } = await import("../../utils/logger.js");
+    vi.mocked(pluginLogger.info).mockImplementation((message: string) => {
+      logged.push(message);
+    });
+
+    loader = new PluginLoader({
+      onWorkerReady: (plugin, worker) => {
+        // Record everything the main thread pushes at the worker.
+        const original = worker.postMessage.bind(worker);
+        worker.postMessage = ((message: unknown) => {
+          state.outbound.push(message);
+          return original(message);
+        }) as typeof worker.postMessage;
+
+        broker!.attach(plugin, worker);
+      },
+      onWorkerGone: (plugin) => broker!.detach(plugin.id),
+    });
+
+    await loader.load(fixture.dir);
+    await loader.activate("sample-plugin", "user-1");
+
+    const line = logged.find((entry) => entry.startsWith("{"));
+    return line ? JSON.parse(line) : { ok: false, error: "no result" };
+  }
+
+  it(
+    "ctx.hosts.get returns no credential material",
+    async () => {
+      const result = await attack(`return await ctx.hosts.get(42);`);
+
+      expect(result.ok).toBe(true);
+      expect(deepContains(result.value, SENTINEL)).toBe(false);
+      for (const field of [
+        "password",
+        "key",
+        "keyPassword",
+        "certPublicKey",
+        "sudoPassword",
+        "autostartPassword",
+        "autostartKey",
+        "autostartKeyPassword",
+        "credentialId",
+        "vaultProfileId",
+      ]) {
+        expect(Object.keys(result.value)).not.toContain(field);
+      }
+    },
+    WORKER_TIMEOUT,
+  );
+
+  it(
+    "ctx.hosts.list returns no credential material",
+    async () => {
+      const result = await attack(`return await ctx.hosts.list();`);
+
+      expect(result.ok).toBe(true);
+      expect(deepContains(result.value, SENTINEL)).toBe(false);
+    },
+    WORKER_TIMEOUT,
+  );
+
+  it(
+    "ctx.ssh.connect returns an opaque handle, not a connection",
+    async () => {
+      const result = await attack(`
+        const handle = await ctx.ssh.connect(42);
+        return {
+          handle,
+          type: typeof handle,
+          // Anything reachable from the handle, if it were an object.
+          reachable: typeof handle === "object" ? Object.keys(handle) : null,
+        };
+      `);
+
+      expect(result.ok).toBe(true);
+      expect(result.value.type).toBe("string");
+      expect(result.value.handle).toMatch(/^sshconn:/);
+      expect(result.value.handle).not.toContain(SENTINEL);
+      // The handle is random, so it cannot encode the host or its secrets.
+      expect(result.value.handle).not.toContain("10.0.0.5");
+      expect(result.value.handle).not.toContain("root");
+    },
+    WORKER_TIMEOUT,
+  );
+
+  it(
+    "ctx.ssh.exec returns only stdout, stderr and code",
+    async () => {
+      const result = await attack(`
+        const handle = await ctx.ssh.connect(42);
+        return await ctx.ssh.exec(handle, "uptime");
+      `);
+
+      expect(result.ok).toBe(true);
+      expect(Object.keys(result.value).sort()).toEqual([
+        "code",
+        "stderr",
+        "stdout",
+      ]);
+      expect(deepContains(result.value, SENTINEL)).toBe(false);
+    },
+    WORKER_TIMEOUT,
+  );
+
+  it(
+    "a failing ctx.ssh.connect does not leak the credential in its error",
+    async () => {
+      const result = await attack(`return await ctx.ssh.connect(42);`, true);
+
+      expect(result.ok).toBe(false);
+      expect(result.error).not.toContain(SENTINEL);
+    },
+    WORKER_TIMEOUT,
+  );
+
+  it(
+    "a plugin cannot walk ctx itself to a credential",
+    async () => {
+      const result = await attack(`
+        const seen = new Set();
+        const found = [];
+        (function walk(value, path, depth) {
+          if (depth > 6 || value === null || seen.has(value)) return;
+          if (typeof value === "string") {
+            if (value.includes(${JSON.stringify(SENTINEL)})) found.push(path);
+            return;
+          }
+          if (typeof value !== "object" && typeof value !== "function") return;
+          seen.add(value);
+          for (const key of Object.getOwnPropertyNames(value)) {
+            let inner;
+            try { inner = value[key]; } catch { continue; }
+            walk(inner, path + "." + key, depth + 1);
+          }
+        })(ctx, "ctx", 0);
+        return found;
+      `);
+
+      expect(result.ok).toBe(true);
+      expect(result.value).toEqual([]);
+    },
+    WORKER_TIMEOUT,
+  );
+
+  it(
+    "no message sent to the worker ever contains credential material",
+    async () => {
+      await attack(`
+        const host = await ctx.hosts.get(42);
+        const list = await ctx.hosts.list();
+        const handle = await ctx.ssh.connect(42);
+        await ctx.ssh.exec(handle, "uptime");
+        await ctx.storage.set("cache", { host, list });
+        return await ctx.storage.get("cache");
+      `);
+
+      expect(state.outbound.length).toBeGreaterThan(0);
+      const leaking = state.outbound.filter((message) =>
+        deepContains(message, SENTINEL),
+      );
+      expect(leaking).toEqual([]);
+    },
+    WORKER_TIMEOUT,
+  );
+
+  it(
+    "a plugin cannot exec on a handle it did not open",
+    async () => {
+      const result = await attack(`
+        return await ctx.ssh.exec("sshconn:00000000-0000-0000-0000-000000000000", "id");
+      `);
+
+      expect(result.ok).toBe(false);
+      expect(result.error).toContain("Unknown or closed SSH connection handle");
+    },
+    WORKER_TIMEOUT,
+  );
+
+  it(
+    "a closed handle stops working",
+    async () => {
+      const result = await attack(`
+        const handle = await ctx.ssh.connect(42);
+        await ctx.ssh.close(handle);
+        try {
+          await ctx.ssh.exec(handle, "id");
+          return "still worked";
+        } catch (error) {
+          return error.message;
+        }
+      `);
+
+      expect(result.value).toContain("Unknown or closed SSH connection handle");
+    },
+    WORKER_TIMEOUT,
+  );
+
+  it(
+    "ssh.connect is denied without the granted capability",
+    async () => {
+      state.grants = [{ pluginId: "sample-plugin", capability: "hosts.read" }];
+      invalidatePluginPermissionCache();
+
+      const result = await attack(`return await ctx.ssh.connect(42);`);
+
+      expect(result.ok).toBe(false);
+      expect(result.error).toContain('is not granted the "ssh.exec"');
+    },
+    WORKER_TIMEOUT,
+  );
+});