Let plugins stream live events and binary data to the app (#8527)

* feat(plugins): stream events and bytes between plugins and the app Plugins can declare a subscription with defineSubscription, handle it with server.handle(contract, (input, { push, signal })), and listen from a surface with useSubscription or client.subscribe. A top-level Uint8Array in an event, an RPC output, or an RPC input crosses the wire as a binary frame (opcodes 0x20-0x2f) without JSON or base64; a nested one fails validation on the sender. Each subscriber runs its own handler, owned by the session that subscribed, so release, disconnect, plugin stop and reload need no plugin code. The runtime buffers pushes until the subscribe response has gone out, so a client never sees an event for a subscription it does not know yet. Gated on server_info.features.pluginSubscriptions; binary outputs reach only clients that advertise plugin_binary_payloads. * fix(plugins): close subscription gaps found in review - Losing daemon.manage now releases the session's plugin subscriptions; binary events bypass the outbound permission check, so streams would otherwise outlive the revocation. - A subscribe that times out sends unsubscribe, since the plugin may still start the handler after the daemon gave up. - The byte limit leaves room for the largest frame header and routing, and lives in assertPluginWireValue so events, outputs and inputs share it. - A release right after the host accepted keeps `ready` resolved, so a handler that returns at once reports closed rather than an error. - Session trace logging tolerates loggers without isLevelEnabled. * fix(plugins): withhold binary plugin results after plugin access is revoked A binary RPC output or event skipped the outbound permission check its JSON form passes, so an RPC in flight when daemon.manage was revoked still delivered its bytes. Each frame now checks the permission of the JSON message it stands in for. * test(plugins): cover subscriptions in a real browser instead of JSDOM The app subscription tests mounted hooks in JSDOM, which docs/testing.md rules out. The adapter and contract behavior is now a plain unit test, and a Playwright spec drives the diff-watch and media-stream examples: JSON events following file changes, streamed bytes, and an upload. * test(plugins): add the plain subscription unit test and the browser spec * fix(plugins): apply subscription schema transforms once Both ends parsed each subscription value, and the sender sent its parse output, so the receiver parsed a schema's output as fresh input. An event schema like z.string().transform((s) => s.length) closed the stream as invalid. The sender now validates and sends the value as given; the receiver's parse produces what the handler or callback sees. * fix(plugins): parse a subscription's input once in the plugin process server.handle wrapped subscription handlers in a second input parse, so a transforming input schema rejected its own parsed output before the handler ran.
2026-11-03 07:03:18 +00:00 · 2026-11-02 14:10:15 +07:00
parent 70cb34927a
commit d4ac64f335
80 changed files with 3237 additions and 87 deletions
@@ -142,6 +142,7 @@ const INBOUND_PERMISSION = {
  "plugin.reload.request": "daemon.manage",
  "plugin.remove.request": "daemon.manage",
  "plugin.rpc.invoke.request": "daemon.manage",
+  "plugin.rpc.subscribe.request": "daemon.manage",
  "plugin.source.install.request": "daemon.manage",
  "plugin.source.status.request": "daemon.manage",
  "plugin.source.update.request": "daemon.manage",
@@ -378,6 +379,9 @@ const OUTBOUND_PERMISSION = {
  "plugin.reload.response": "daemon.manage",
  "plugin.remove.response": "daemon.manage",
  "plugin.rpc.invoke.response": "daemon.manage",
+  "plugin.rpc.subscribe.response": "daemon.manage",
+  "plugin.rpc.event": "daemon.manage",
+  "plugin.rpc.closed": "daemon.manage",
  "plugin.source.install.response": "daemon.manage",
  "plugin.source.status.response": "daemon.manage",
  "plugin.source.update.response": "daemon.manage",
@@ -34,7 +34,13 @@ const emptyUsageRuntime = {
  getUsageSourceRegistrations: () => [],
  fetchUsage: async () => undefined,
  discoverUsage: async () => [],
-} satisfies Pick<TestPluginRuntime, "getUsageSourceRegistrations" | "fetchUsage" | "discoverUsage">;
+  startSubscription: () => {
+    throw new Error("No subscriptions in this test");
+  },
+} satisfies Pick<
+  TestPluginRuntime,
+  "getUsageSourceRegistrations" | "fetchUsage" | "discoverUsage" | "startSubscription"
+>;

 async function createPlugin(id: string, source: string): Promise<string> {
  const directory = await mkdtemp(path.join(tmpdir(), "paseo-plugin-service-"));
@@ -25,6 +25,7 @@ import { readPluginManifest } from "./manifest.js";
 import { expandTilde } from "../../utils/path.js";
 import { runPluginBuild } from "./preparation.js";
 import { PluginRuntime } from "./runtime.js";
+import type { PluginRuntimeSubscription } from "./runtime-subscription.js";
 import { BuiltinPluginLoader, type BuiltinPlugin } from "./builtin/index.js";
 import type { PluginProviderMetadata } from "./plugin-process-protocol.js";
 import { readPluginProviderIcon } from "./provider-icon.js";
@@ -48,6 +49,7 @@ interface PluginRuntimePort {
  wait?: PluginLifecycle["wait"];
  catalog: PluginRuntime["catalog"];
  invoke(pluginId: string, method: string, input: unknown): Promise<unknown>;
+  startSubscription: PluginRuntime["startSubscription"];
  getLogs(pluginId: string): PluginLogEntry[];
  clearLogs(pluginId: string): void;
  getProviderRegistrations?(pluginId: string): readonly PluginProviderMetadata[];
@@ -539,6 +541,10 @@ export class PluginService {
    return this.runtime.invoke(pluginId, method, input);
  }

+  subscribePluginRpc(pluginId: string, method: string, input: unknown): PluginRuntimeSubscription {
+    return this.runtime.startSubscription(pluginId, method, input);
+  }
+
  async stopAllPlugins(): Promise<void> {
    this.globalStartsBlocked = true;
    const stopping = this.stopAll();
@@ -60,6 +60,14 @@ export type PluginProcessRequest =
  | { type: "usage.fetch"; requestId: string; sourceId: string; input: unknown }
  | { type: "usage.discover"; requestId: string; sourceId: string; scope: UsageScope }
  | { type: "invoke"; requestId: string; method: string; input: unknown }
+  | {
+      type: "subscribe";
+      requestId: string;
+      subscriptionId: string;
+      method: string;
+      input: unknown;
+    }
+  | { type: "unsubscribe"; subscriptionId: string }
  | {
      type: "provider.connect";
      providerId: string;
@@ -83,6 +91,7 @@ export type PluginProcessMessage =
  | {
      type: "ready";
      methods: string[];
+      subscriptions?: string[];
      providers: PluginProviderMetadata[];
      usageSources?: PluginUsageSourceMetadata[];
      hooks?: string[];
@@ -90,6 +99,8 @@ export type PluginProcessMessage =
  | { type: "result"; requestId: string; output: unknown }
  | { type: "error"; requestId: string; error: string }
  | { type: "fatal"; error: string }
+  | { type: "subscription.event"; subscriptionId: string; event: unknown }
+  | { type: "subscription.closed"; subscriptionId: string; error?: string }
  | {
      type: "provider.connected";
      connectionId: string;
@@ -240,6 +251,16 @@ export const PluginProcessRequestSchema: z.ZodType<PluginProcessRequest> = z.dis
        input: z.unknown(),
      })
      .strict(),
+    z
+      .object({
+        type: z.literal("subscribe"),
+        requestId: z.string().min(1),
+        subscriptionId: z.string().min(1),
+        method: z.string().min(1),
+        input: z.unknown(),
+      })
+      .strict(),
+    z.object({ type: z.literal("unsubscribe"), subscriptionId: z.string().min(1) }).strict(),
    z
      .object({
        type: z.literal("provider.connect"),
@@ -272,6 +293,7 @@ export const PluginProcessMessageSchema: z.ZodType<PluginProcessMessage> = z.dis
      .object({
        type: z.literal("ready"),
        methods: z.array(z.string()),
+        subscriptions: z.array(z.string()).optional(),
        providers: z.array(providerMetadataSchema),
        usageSources: z.array(usageSourceMetadataSchema).optional(),
        hooks: hooksSchema.optional(),
@@ -284,6 +306,20 @@ export const PluginProcessMessageSchema: z.ZodType<PluginProcessMessage> = z.dis
      .object({ type: z.literal("error"), requestId: z.string().min(1), error: z.string() })
      .strict(),
    z.object({ type: z.literal("fatal"), error: z.string() }).strict(),
+    z
+      .object({
+        type: z.literal("subscription.event"),
+        subscriptionId: z.string().min(1),
+        event: z.unknown(),
+      })
+      .strict(),
+    z
+      .object({
+        type: z.literal("subscription.closed"),
+        subscriptionId: z.string().min(1),
+        error: z.string().optional(),
+      })
+      .strict(),
    z
      .object({
        type: z.literal("provider.connected"),
@@ -7,8 +7,21 @@ import {
  type PluginProcessRequest,
 } from "./plugin-process-protocol.js";
 import type { UsageSourceRegistration } from "@getpaseo/plugin/server/usage";
-import type { SettingsDefinition, PluginRpcContract } from "@getpaseo/plugin";
-import type { PluginHandlerContext, PluginServerContribution } from "@getpaseo/plugin/server";
+import type {
+  SettingsDefinition,
+  PluginRpcContract,
+  PluginSubscriptionContract,
+} from "@getpaseo/plugin";
+import type {
+  PluginHandlerContext,
+  PluginServerContext,
+  PluginServerContribution,
+} from "@getpaseo/plugin/server";
+import { assertPluginWireValue } from "@getpaseo/protocol/binary-frames/index";
+import {
+  PluginSubscriptionHandlers,
+  type PluginSubscriptionHandler,
+} from "./subscription-handlers.js";
 import { fileURLToPath } from "node:url";
 import type { ZodType } from "zod";
 import {
@@ -66,6 +79,7 @@ export function createPluginWorker(options: {
    send({ type: "hooks.changed", hooks: hooks.catalog() });
  }, agentTools);
  const handlers = new Map<string, RegisteredRpc>();
+  const subscriptions = new PluginSubscriptionHandlers({ send, sendAndWait });
  const providers = new Map<string, ProviderRegistration>();
  const usageSources = new Map<string, UsageSourceRegistration>();
  const providerConnections = new Map<
@@ -98,12 +112,36 @@ export function createPluginWorker(options: {
    if (!/^[a-z][a-z0-9._-]*$/.test(normalized)) {
      throw new Error(`Invalid plugin RPC method: ${method}`);
    }
-    if (handlers.has(normalized)) {
+    if (handlers.has(normalized) || subscriptions.has(normalized)) {
      throw new Error(`Duplicate plugin RPC method: ${normalized}`);
    }
    return normalized;
  }

+  function registerSubscription(
+    contract: PluginSubscriptionContract,
+    handler: PluginSubscriptionHandler,
+  ): void {
+    if (typeof handler !== "function") {
+      throw new Error(`Plugin subscription ${contract.name} must provide a handler`);
+    }
+    const method = validateMethod(contract.name);
+    subscriptions.register({ ...contract, name: method }, handler);
+  }
+
+  // One name serves either an RPC or a subscription; the contract says which.
+  const handle = ((
+    contract: PluginRpcContract | PluginSubscriptionContract,
+    handler: (input: unknown, context: never) => unknown,
+  ) => {
+    if ("event" in contract) {
+      // The subscription handlers parse the input once and pass the result here.
+      registerSubscription(contract, handler as PluginSubscriptionHandler);
+      return;
+    }
+    register(contract, (input, context) => handler(contract.input.parse(input), context as never));
+  }) as PluginServerContext["handle"];
+
  function register(contract: PluginRpcContract, handler: RpcHandler): void {
    if (typeof handler !== "function") {
      throw new Error(`Plugin RPC ${contract.name} must provide a handler`);
@@ -284,8 +322,7 @@ export function createPluginWorker(options: {
        )
      : null;
    const contributedCleanup = contribute({
-      handle: (contract, handler) =>
-        register(contract, (input, context) => handler(contract.input.parse(input), context)),
+      handle,
      registerProvider,
      registerUsageSource,
      registerSettings,
@@ -311,6 +348,7 @@ export function createPluginWorker(options: {
    send({
      type: "ready",
      methods: [...handlers.keys()].sort(),
+      subscriptions: subscriptions.catalog(),
      hooks: hooks.catalog(),
      providers: [...providers.values()]
        .sort((left, right) => left.id.localeCompare(right.id))
@@ -327,6 +365,7 @@ export function createPluginWorker(options: {
      .catch((error) => console.error("Plugin API cleanup failed", error));
    hooks.close();
    agentTools.close();
+    subscriptions.releaseAll();
    for (const pending of pendingProviderConnections.values()) pending.tombstoned = true;
    const currentCleanup = cleanup;
    cleanup = null;
@@ -365,7 +404,8 @@ export function createPluginWorker(options: {
      message.type === "provider.status" ||
      message.type === "provider.catalog_key" ||
      message.type === "usage.fetch" ||
-      message.type === "usage.discover"
+      message.type === "usage.discover" ||
+      message.type === "subscribe"
    ) {
      send({ type: "error", requestId: message.requestId, error: "Plugin is stopping" });
    } else if (message.type === "provider.connect") {
@@ -422,6 +462,14 @@ export function createPluginWorker(options: {
    return false;
  }

+  function handleSubscriptionMessage(
+    message: Extract<PluginProcessRequest, { type: "subscribe" | "unsubscribe" }>,
+  ): void {
+    if (message.type === "unsubscribe") subscriptions.release(message.subscriptionId);
+    else if (paseo) subscriptions.start(message, paseo);
+    else send({ type: "error", requestId: message.requestId, error: "Plugin API unavailable" });
+  }
+
  function handleMessage(rawMessage: unknown): void {
    const parsed = PluginProcessRequestSchema.safeParse(rawMessage);
    if (!parsed.success) {
@@ -525,6 +573,11 @@ export function createPluginWorker(options: {
      handleHookMessage(message);
      return;
    }
+    if (message.type === "invoke") invoke(message);
+    else handleSubscriptionMessage(message);
+  }
+
+  function invoke(message: Extract<PluginProcessRequest, { type: "invoke" }>): void {
    const registered = handlers.get(message.method);
    if (!registered) {
      send({
@@ -541,6 +594,10 @@ export function createPluginWorker(options: {
        return registered.handler(input, { paseo });
      })
      .then((output) => registered.contract.output.parseAsync(output))
+      .then((output: unknown) => {
+        assertPluginWireValue(output, "output");
+        return output;
+      })
      .then(
        (output) => send({ type: "result", requestId: message.requestId, output }),
        (error) =>
@@ -0,0 +1,354 @@
+import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { WebSocket } from "ws";
+import {
+  DaemonClient,
+  type OwnedSubscription,
+  type WebSocketLike,
+} from "@getpaseo/client/internal/daemon-client";
+import { afterEach, expect, test } from "vitest";
+import { resolveDaemonVersion } from "../daemon-version.js";
+import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo-daemon.js";
+
+const PLUGIN_ID = "subscriptions";
+
+// Written as an author writes a plugin: one contract per name, handlers that read like English.
+const SERVER_ENTRY = `
+import { defineRpc, defineSubscription } from "@getpaseo/plugin";
+import type { PluginServerContext } from "@getpaseo/plugin/server";
+import { z } from "zod";
+
+const ticks = defineSubscription({
+  name: "ticks",
+  input: z.object({ count: z.number().int(), bytes: z.boolean() }),
+  event: z.union([z.instanceof(Uint8Array), z.object({ n: z.number() })]),
+});
+const failing = defineSubscription({
+  name: "failing",
+  input: z.object({ when: z.enum(["before-ack", "after-ack"]) }),
+  event: z.object({}),
+});
+const nestedPush = defineSubscription({
+  name: "nested-push",
+  input: z.object({}),
+  event: z.object({ data: z.instanceof(Uint8Array) }),
+});
+const repeated = defineSubscription({
+  name: "repeated",
+  input: z.object({ times: z.string().transform(Number) }),
+  event: z.string().transform((text) => text.length),
+});
+const watchers = defineRpc({
+  name: "watchers",
+  input: z.object({}),
+  output: z.object({ starts: z.number(), active: z.number() }),
+});
+const thumbnail = defineRpc({
+  name: "thumbnail",
+  input: z.object({ size: z.number().int() }),
+  output: z.instanceof(Uint8Array),
+});
+const nestedOutput = defineRpc({
+  name: "nested-output",
+  input: z.object({}),
+  output: z.object({ data: z.instanceof(Uint8Array) }),
+});
+const reverse = defineRpc({
+  name: "reverse",
+  input: z.instanceof(Uint8Array),
+  output: z.instanceof(Uint8Array),
+});
+
+export default function contribute(server: PluginServerContext) {
+  let starts = 0;
+  let active = 0;
+  server.handle(ticks, ({ count, bytes }, { push, signal }) => {
+    starts += 1;
+    active += 1;
+    signal.addEventListener("abort", () => { active -= 1; }, { once: true });
+    for (let n = 0; n < count; n += 1) void push(bytes ? new Uint8Array([n, n + 1]) : { n });
+  });
+  server.handle(failing, ({ when }) => {
+    if (when === "before-ack") throw new Error("refused before acknowledging");
+    return Promise.reject(new Error("failed after acknowledging"));
+  });
+  server.handle(nestedPush, (_input, { push }) => {
+    void push({ data: new Uint8Array(1) }).catch((error) => {
+      console.error("nested push", error.message);
+    });
+  });
+  server.handle(repeated, async ({ times }, { push }) => {
+    await push(typeof times === "number" ? "x".repeat(times) : "input arrived untransformed");
+  });
+  server.handle(watchers, () => ({ starts, active }));
+  server.handle(thumbnail, ({ size }) => Buffer.alloc(size, 7));
+  server.handle(nestedOutput, () => ({ data: new Uint8Array(1) }));
+  server.handle(reverse, (bytes) => Uint8Array.from(bytes).reverse());
+  return () => undefined;
+}
+`;
+
+type Received =
+  | { type: "event"; event: unknown }
+  | { type: "bytes"; bytes: number[] }
+  | { type: "closed"; error?: string };
+
+const roots: string[] = [];
+
+afterEach(async () => {
+  await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })));
+});
+
+interface Harness {
+  daemon: TestPaseoDaemon;
+  client: DaemonClient;
+  sockets: WebSocket[];
+  close(): Promise<void>;
+}
+
+async function writePlugin(id: string, serverEntry: string): Promise<string> {
+  const pluginDirectory = await mkdtemp(path.join(tmpdir(), "paseo-plugin-subscriptions-"));
+  roots.push(pluginDirectory);
+  await writeFile(
+    path.join(pluginDirectory, "paseo-plugin.json"),
+    JSON.stringify({ id, requirements: { paseo: `>=${resolveDaemonVersion(import.meta.url)}` } }),
+  );
+  await writeFile(path.join(pluginDirectory, "index.server.ts"), serverEntry);
+  return pluginDirectory;
+}
+
+async function startHarness(): Promise<Harness> {
+  const pluginDirectory = await writePlugin(PLUGIN_ID, SERVER_ENTRY);
+  const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
+  const sockets: WebSocket[] = [];
+  const client = new DaemonClient({
+    url: `ws://127.0.0.1:${daemon.port}/ws`,
+    clientId: `clid_plugin_subscriptions_${Date.now()}`,
+    appVersion: resolveDaemonVersion(import.meta.url),
+    reconnect: { enabled: true, baseDelayMs: 10, maxDelayMs: 50 },
+    webSocketFactory: (url) => {
+      const socket = new WebSocket(url);
+      sockets.push(socket);
+      return socket as unknown as WebSocketLike;
+    },
+  });
+  await client.connect();
+  await client.patchDaemonConfig({ pluginsEnabled: true });
+  await expect(client.installDirectoryPlugin(pluginDirectory)).resolves.toMatchObject({
+    id: PLUGIN_ID,
+    status: "running",
+  });
+  return {
+    daemon,
+    client,
+    sockets,
+    async close() {
+      await client.close().catch(() => undefined);
+      await daemon.close();
+    },
+  };
+}
+
+function collect(subscription: OwnedSubscription<unknown>): Received[] {
+  const received: Received[] = [];
+  subscription.subscribe({
+    snapshot: () => undefined,
+    update: (message) => {
+      if (message.type === "plugin.rpc.event")
+        received.push({ type: "event", event: message.payload.event });
+      if (message.type === "plugin.rpc.closed")
+        received.push({
+          type: "closed",
+          ...(message.payload.error ? { error: message.payload.error } : {}),
+        });
+    },
+    binary: (payload) => received.push({ type: "bytes", bytes: [...payload] }),
+  });
+  return received;
+}
+
+function watcherCount(client: DaemonClient) {
+  return client.invokePluginRpc(PLUGIN_ID, "watchers", {});
+}
+
+test("a subscription delivers JSON and binary events in order and release aborts its handler", async () => {
+  const harness = await startHarness();
+  const { client } = harness;
+  try {
+    const json = client.observePluginRpc(PLUGIN_ID, "ticks", { count: 3, bytes: false });
+    const jsonEvents = collect(json);
+    await json.ready;
+    const binary = client.observePluginRpc(PLUGIN_ID, "ticks", { count: 2, bytes: true });
+    const binaryEvents = collect(binary);
+    await binary.ready;
+
+    await expect
+      .poll(() => jsonEvents)
+      .toEqual([
+        { type: "event", event: { n: 0 } },
+        { type: "event", event: { n: 1 } },
+        { type: "event", event: { n: 2 } },
+      ]);
+    await expect
+      .poll(() => binaryEvents)
+      .toEqual([
+        { type: "bytes", bytes: [0, 1] },
+        { type: "bytes", bytes: [1, 2] },
+      ]);
+    expect(json.subscriptionId).not.toBe(binary.subscriptionId);
+    await expect(watcherCount(client)).resolves.toEqual({ starts: 2, active: 2 });
+
+    await json.release();
+    await expect(watcherCount(client)).resolves.toEqual({ starts: 2, active: 1 });
+    await binary.release();
+    await expect(watcherCount(client)).resolves.toEqual({ starts: 2, active: 0 });
+  } finally {
+    await harness.close();
+  }
+}, 60_000);
+
+test("a schema transform runs once: the handler gets the parsed input and the wire carries the pushed event", async () => {
+  const harness = await startHarness();
+  try {
+    const subscription = harness.client.observePluginRpc(PLUGIN_ID, "repeated", { times: "2" });
+    const received = collect(subscription);
+    await subscription.ready;
+    // The client parses "xx" into 2; parsing it on the plugin side too would reject it.
+    await expect.poll(() => received).toEqual([{ type: "event", event: "xx" }, { type: "closed" }]);
+  } finally {
+    await harness.close();
+  }
+}, 60_000);
+
+test("dropping the connection releases subscriptions and reconnecting runs the handler again", async () => {
+  const harness = await startHarness();
+  const { client, sockets } = harness;
+  try {
+    const subscription = client.observePluginRpc(PLUGIN_ID, "ticks", { count: 1, bytes: false });
+    const events = collect(subscription);
+    await subscription.ready;
+    await expect.poll(() => events).toEqual([{ type: "event", event: { n: 0 } }]);
+    const firstId = subscription.subscriptionId;
+
+    sockets.at(-1)!.terminate();
+    await expect
+      .poll(() => subscription.subscriptionId, { timeout: 10_000 })
+      .toSatisfy((id: unknown) => typeof id === "string" && id !== firstId);
+    await expect
+      .poll(() => events)
+      .toEqual([
+        { type: "event", event: { n: 0 } },
+        { type: "event", event: { n: 0 } },
+      ]);
+    // The first handler's signal aborted when its connection dropped; one handler runs now.
+    await expect(watcherCount(client)).resolves.toEqual({ starts: 2, active: 1 });
+    await subscription.release();
+  } finally {
+    await harness.close();
+  }
+}, 60_000);
+
+test("a handler that throws before acknowledging rejects; one that fails later closes", async () => {
+  const harness = await startHarness();
+  const { client } = harness;
+  try {
+    const refused = client.observePluginRpc(PLUGIN_ID, "failing", { when: "before-ack" });
+    await expect(refused.ready).rejects.toThrow("refused before acknowledging");
+
+    const later = client.observePluginRpc(PLUGIN_ID, "failing", { when: "after-ack" });
+    const events = collect(later);
+    await later.ready;
+    await expect
+      .poll(() => events)
+      .toEqual([{ type: "closed", error: "failed after acknowledging" }]);
+    await later.release();
+
+    const invalid = client.observePluginRpc(PLUGIN_ID, "ticks", { count: "many" });
+    await expect(invalid.ready).rejects.toThrow();
+    const unknown = client.observePluginRpc(PLUGIN_ID, "watchers", {});
+    await expect(unknown.ready).rejects.toThrow("does not contribute subscription watchers");
+  } finally {
+    await harness.close();
+  }
+}, 60_000);
+
+test("reloading the plugin closes its subscriptions with an error", async () => {
+  const harness = await startHarness();
+  const { client } = harness;
+  try {
+    const subscription = client.observePluginRpc(PLUGIN_ID, "ticks", { count: 0, bytes: false });
+    const events = collect(subscription);
+    await subscription.ready;
+    await client.reloadPlugin(PLUGIN_ID);
+    await expect
+      .poll(() => events)
+      .toEqual([{ type: "closed", error: `Plugin stopped: ${PLUGIN_ID}` }]);
+    await subscription.release();
+    // The reloaded plugin starts fresh and serves new subscriptions.
+    const again = client.observePluginRpc(PLUGIN_ID, "ticks", { count: 1, bytes: false });
+    const againEvents = collect(again);
+    await again.ready;
+    await expect.poll(() => againEvents).toEqual([{ type: "event", event: { n: 0 } }]);
+    await again.release();
+  } finally {
+    await harness.close();
+  }
+}, 60_000);
+
+test("RPC bytes ride as binary frames in both directions and nested bytes fail validation", async () => {
+  const harness = await startHarness();
+  const { client } = harness;
+  try {
+    const output = await client.invokePluginRpc(PLUGIN_ID, "thumbnail", { size: 4 });
+    expect(output).toBeInstanceOf(Uint8Array);
+    expect([...(output as Uint8Array)]).toEqual([7, 7, 7, 7]);
+
+    const reversed = await client.invokePluginRpc(PLUGIN_ID, "reverse", new Uint8Array([1, 2, 3]));
+    expect([...(reversed as Uint8Array)]).toEqual([3, 2, 1]);
+
+    await expect(client.invokePluginRpc(PLUGIN_ID, "nested-output", {})).rejects.toThrow(
+      "output.data is a Uint8Array. Bytes cross the wire only as the whole value",
+    );
+    await expect(
+      client.invokePluginRpc(PLUGIN_ID, "reverse", { data: new Uint8Array(1) }),
+    ).rejects.toThrow("input.data is a Uint8Array");
+
+    const nested = client.observePluginRpc(PLUGIN_ID, "nested-push", {});
+    await nested.ready;
+    await expect
+      .poll(async () => JSON.stringify(await client.getPluginLogs(PLUGIN_ID)))
+      .toContain("event.data is a Uint8Array");
+    await nested.release();
+  } finally {
+    await harness.close();
+  }
+}, 60_000);
+
+test("one name cannot serve both an RPC and a subscription", async () => {
+  const harness = await startHarness();
+  try {
+    const directory = await writePlugin(
+      "collision",
+      `
+import { defineRpc, defineSubscription } from "@getpaseo/plugin";
+import type { PluginServerContext } from "@getpaseo/plugin/server";
+import { z } from "zod";
+
+export default function contribute(server: PluginServerContext) {
+  server.handle(defineRpc({ name: "watch", input: z.object({}), output: z.null() }), () => null);
+  server.handle(
+    defineSubscription({ name: "watch", input: z.object({}), event: z.object({}) }),
+    () => undefined,
+  );
+  return () => undefined;
+}
+`,
+    );
+    await expect(harness.client.installDirectoryPlugin(directory)).rejects.toThrow(
+      "Duplicate plugin RPC method: watch",
+    );
+  } finally {
+    await harness.close();
+  }
+}, 60_000);
@@ -0,0 +1,108 @@
+import { mkdtemp, rm, writeFile } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { DaemonClient } from "@getpaseo/client/internal/daemon-client";
+import { afterEach, expect, test } from "vitest";
+import { resolveDaemonVersion } from "../daemon-version.js";
+import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js";
+
+const PUSH_BYTES = 256 * 1024;
+const TOTAL_BYTES = 100 * 1024 * 1024;
+const PUSHES = TOTAL_BYTES / PUSH_BYTES;
+const MIN_MEGABYTES_PER_SECOND = 200;
+
+// One buffer, pushed until the total is out: what a capture loop looks like.
+const SERVER_ENTRY = `
+import { defineSubscription } from "@getpaseo/plugin";
+import type { PluginServerContext } from "@getpaseo/plugin/server";
+import { z } from "zod";
+
+const stream = defineSubscription({
+  name: "stream",
+  input: z.object({ pushes: z.number().int(), bytes: z.number().int() }),
+  event: z.instanceof(Uint8Array),
+});
+
+export default function contribute(server: PluginServerContext) {
+  server.handle(stream, async ({ pushes, bytes }, { push, signal }) => {
+    const chunk = new Uint8Array(bytes).fill(7);
+    for (let sent = 0; sent < pushes && !signal.aborted; sent += 1) await push(chunk);
+    await new Promise((resolve) => signal.addEventListener("abort", resolve, { once: true }));
+  });
+  return () => undefined;
+}
+`;
+
+const roots: string[] = [];
+
+afterEach(async () => {
+  await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true })));
+});
+
+test(`a plugin streams ${TOTAL_BYTES / 1024 / 1024} MiB in ${PUSH_BYTES / 1024} KiB pushes at ${MIN_MEGABYTES_PER_SECOND} MB/s without copying per frame`, async () => {
+  const pluginDirectory = await mkdtemp(path.join(tmpdir(), "paseo-plugin-stream-"));
+  roots.push(pluginDirectory);
+  await writeFile(
+    path.join(pluginDirectory, "paseo-plugin.json"),
+    JSON.stringify({
+      id: "stream",
+      requirements: { paseo: `>=${resolveDaemonVersion(import.meta.url)}` },
+    }),
+  );
+  await writeFile(path.join(pluginDirectory, "index.server.ts"), SERVER_ENTRY);
+  const daemon = await createTestPaseoDaemon({ mcpEnabled: false });
+  const client = new DaemonClient({
+    url: `ws://127.0.0.1:${daemon.port}/ws`,
+    clientId: `clid_plugin_stream_${Date.now()}`,
+    appVersion: resolveDaemonVersion(import.meta.url),
+  });
+  try {
+    await client.connect();
+    await client.patchDaemonConfig({ pluginsEnabled: true });
+    await client.installDirectoryPlugin(pluginDirectory);
+
+    let received = 0;
+    let frames = 0;
+    let copied = 0;
+    let finished!: () => void;
+    const done = new Promise<void>((resolve) => {
+      finished = resolve;
+    });
+    const baseline = process.memoryUsage().arrayBuffers;
+    let peak = baseline;
+    const started = performance.now();
+    const subscription = client.observePluginRpc("stream", "stream", {
+      pushes: PUSHES,
+      bytes: PUSH_BYTES,
+    });
+    subscription.subscribe({
+      snapshot: () => undefined,
+      update: () => undefined,
+      binary: (payload) => {
+        received += payload.byteLength;
+        frames += 1;
+        // The payload is a view into the frame the socket received, not a copy of it.
+        if (payload.buffer.byteLength - payload.byteLength > 1024) copied += 1;
+        if (frames % 40 === 0) peak = Math.max(peak, process.memoryUsage().arrayBuffers);
+        if (received >= TOTAL_BYTES) finished();
+      },
+    });
+    await done;
+    const seconds = (performance.now() - started) / 1000;
+    const megabytesPerSecond = TOTAL_BYTES / 1e6 / seconds;
+    process.stderr.write(
+      `[plugin stream] ${frames} frames, ${(TOTAL_BYTES / 1e6).toFixed(1)} MB in ${(seconds * 1000).toFixed(0)} ms: ${megabytesPerSecond.toFixed(0)} MB/s, peak external buffers +${((peak - baseline) / 1024 / 1024).toFixed(1)} MiB\n`,
+    );
+
+    expect(frames).toBe(PUSHES);
+    expect(received).toBe(TOTAL_BYTES);
+    expect(copied).toBe(0);
+    expect(megabytesPerSecond).toBeGreaterThanOrEqual(MIN_MEGABYTES_PER_SECOND);
+    // Memory held while streaming stays bounded by in-flight frames, not by the payload total.
+    expect(peak - baseline).toBeLessThan(TOTAL_BYTES / 2);
+    await subscription.release();
+  } finally {
+    await client.close().catch(() => undefined);
+    await daemon.close();
+  }
+}, 120_000);
@@ -0,0 +1,56 @@
+/** Receives a plugin subscription's events once its owner attached. */
+export interface PluginSubscriptionSink {
+  event(event: unknown): void;
+  /** The plugin ended the subscription, or stopped. Not called after `release()`. */
+  closed(error?: string): void;
+}
+
+/** A started plugin subscription, from the daemon's side of the subprocess. */
+export interface PluginRuntimeSubscription {
+  /** Resolves when the plugin acknowledged; rejects when it refused. */
+  readonly ready: Promise<void>;
+  /** Delivers what arrived so far, then everything that follows. */
+  attach(sink: PluginSubscriptionSink): void;
+  release(): void;
+}
+
+type Pending = { type: "event"; event: unknown } | { type: "closed"; error?: string };
+
+/**
+ * Holds events until the owner attaches, so the owner can answer the subscribe request before
+ * the first event goes out.
+ */
+export class BufferedPluginSubscription {
+  private sink: PluginSubscriptionSink | null = null;
+  private pending: Pending[] = [];
+  private ended = false;
+
+  attach(sink: PluginSubscriptionSink): void {
+    if (this.sink) throw new Error("Plugin subscription is already attached");
+    this.sink = sink;
+    const pending = this.pending;
+    this.pending = [];
+    for (const entry of pending) {
+      if (entry.type === "event") sink.event(entry.event);
+      else sink.closed(entry.error);
+    }
+  }
+
+  event(event: unknown): void {
+    if (this.ended) return;
+    if (this.sink) this.sink.event(event);
+    else this.pending.push({ type: "event", event });
+  }
+
+  closed(error?: string): void {
+    if (this.ended) return;
+    this.ended = true;
+    if (this.sink) this.sink.closed(error);
+    else this.pending.push({ type: "closed", ...(error === undefined ? {} : { error }) });
+  }
+
+  released(): void {
+    this.ended = true;
+    this.pending = [];
+  }
+}
@@ -1399,6 +1399,44 @@ export default function contribute(server: { registerProvider(provider: Provider
    }
  });

+  it("releases a subscription in the plugin when its subscribe times out", async () => {
+    vi.useFakeTimers();
+    try {
+      const directory = await createPlugin(
+        "slow-subscribe",
+        `export default function contribute(plugin: unknown) { void plugin; return () => undefined; }`,
+      );
+      const child = createReloadChild("slow-subscribe", []);
+      const sent: Array<{ type: string; subscriptionId?: string }> = [];
+      const originalSend = child.send.bind(child);
+      child.send = (message, callback) => {
+        sent.push(message);
+        if (message.type !== "initialize") return originalSend(message, callback);
+        callback?.(null);
+        queueMicrotask(() =>
+          child.emitMessage({ type: "ready", methods: [], subscriptions: ["feed"], providers: [] }),
+        );
+        return true;
+      };
+      const runtime = createTestRuntime({ spawnChild: () => child });
+      await runtime.startPlugin("slow-subscribe", directory);
+
+      const subscription = runtime.startSubscription("slow-subscribe", "feed", {});
+      const rejection = subscription.ready.catch((error: Error) => error);
+      await vi.advanceTimersByTimeAsync(30_000);
+      await expect(rejection).resolves.toMatchObject({
+        message: expect.stringContaining("timed out"),
+      });
+      const subscribe = sent.find((message) => message.type === "subscribe");
+      expect(sent).toContainEqual({
+        type: "unsubscribe",
+        subscriptionId: subscribe?.subscriptionId,
+      });
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
  it("escalates a plugin that ignores graceful shutdown from TERM to KILL", async () => {
    vi.useFakeTimers();
    try {
@@ -34,6 +34,10 @@ import type {
  PluginUsageSourceMetadata,
 } from "./plugin-process-protocol.js";
 import { PluginProcessMessageSchema } from "./plugin-process-protocol.js";
+import {
+  BufferedPluginSubscription,
+  type PluginRuntimeSubscription,
+} from "./runtime-subscription.js";
 import { PluginSessionSocket } from "./session-socket.js";
 import { InternalPluginChild } from "./internal-child.js";
 import { evaluateBundle } from "./bundle-evaluator.js";
@@ -75,6 +79,8 @@ interface LoadedPlugin {
  /** Lets a client that already holds this bundle skip downloading it again. */
  clientBundleHash: string;
  methods: ReadonlySet<string>;
+  subscriptionMethods: ReadonlySet<string>;
+  subscriptions: Map<string, BufferedPluginSubscription>;
  hooks: string[];
  providers: readonly PluginProviderMetadata[];
  usageSources: readonly PluginUsageSourceMetadata[];
@@ -514,6 +520,40 @@ export class PluginRuntime {
    return this.request(loaded, { type: "invoke", requestId: randomUUID(), method, input });
  }

+  /** Starts one subscriber's handler in the plugin process. */
+  startSubscription(pluginId: string, method: string, input: unknown): PluginRuntimeSubscription {
+    const loaded = this.plugins.get(pluginId);
+    if (!loaded) throw new Error(`Plugin is not available: ${pluginId}`);
+    if (!loaded.subscriptionMethods.has(method))
+      throw new Error(`Plugin ${pluginId} does not contribute subscription ${method}`);
+    const subscriptionId = randomUUID();
+    const subscription = new BufferedPluginSubscription();
+    loaded.subscriptions.set(subscriptionId, subscription);
+    const ready = this.request(loaded, {
+      type: "subscribe",
+      requestId: randomUUID(),
+      subscriptionId,
+      method,
+      input,
+    }).then(
+      () => undefined,
+      (error: unknown) => {
+        loaded.subscriptions.delete(subscriptionId);
+        throw error;
+      },
+    );
+    return {
+      ready,
+      attach: (sink) => subscription.attach(sink),
+      release: () => {
+        subscription.released();
+        if (!loaded.subscriptions.delete(subscriptionId)) return;
+        if (loaded.child?.connected)
+          void send(loaded.child, { type: "unsubscribe", subscriptionId }).catch(() => undefined);
+      },
+    };
+  }
+
  emit<Name extends keyof PluginHookEvents>(name: Name, event: PluginHookEvents[Name]): void {
    for (const loaded of this.plugins.values()) {
      if (!loaded.hooks.includes(name)) {
@@ -727,6 +767,11 @@ export class PluginRuntime {
          if (message.type === "hook") {
            void send(child, { type: "hook.cancel", requestId }).catch(() => {});
          }
+          // The handler may still start after the daemon gave up on it.
+          if (message.type === "subscribe") {
+            const { subscriptionId } = message;
+            void send(child, { type: "unsubscribe", subscriptionId }).catch(() => {});
+          }
          reject(new Error(`Plugin RPC timed out: ${pluginId}.${message.type}`));
        },
        message.type === "tool.call" ? TOOL_CALL_TIMEOUT_MS : REQUEST_TIMEOUT_MS,
@@ -781,6 +826,8 @@ export class PluginRuntime {
        clientBundleHash: hashClientBundle(clientBundle),
        requirements: manifest.requirements,
        methods: new Set(),
+        subscriptionMethods: new Set(),
+        subscriptions: new Map(),
        hooks: [],
        providers: [],
        usageSources: [],
@@ -911,6 +958,8 @@ export class PluginRuntime {
      clientBundleHash: hashClientBundle(clientBundle),
      requirements,
      methods: new Set(ready.methods),
+      subscriptionMethods: new Set(ready.subscriptions ?? []),
+      subscriptions: new Map(),
      hooks: ready.hooks ?? [],
      providers: ready.providers ?? [],
      usageSources: ready.usageSources ?? [],
@@ -1009,6 +1058,16 @@ export class PluginRuntime {
      this.handleProviderMessage(loaded, message);
      return;
    }
+    if (message.type === "subscription.event") {
+      loaded.subscriptions.get(message.subscriptionId)?.event(message.event);
+      return;
+    }
+    if (message.type === "subscription.closed") {
+      const subscription = loaded.subscriptions.get(message.subscriptionId);
+      loaded.subscriptions.delete(message.subscriptionId);
+      subscription?.closed(message.error);
+      return;
+    }
    if (message.type !== "result" && message.type !== "error") return;
    const pending = loaded.pending.get(message.requestId);
    if (!pending) return;
@@ -1367,12 +1426,16 @@ export class PluginRuntime {
    this.appendLog(loaded.id, "stdout", "[paseo] Plugin stopped");
  }

+  /** Rejects in-flight requests and closes open subscriptions with the reason. */
  private rejectPending(loaded: LoadedPlugin, message: string): void {
    for (const invocation of loaded.pending.values()) {
      clearTimeout(invocation.timeout);
      invocation.reject(new Error(message));
    }
    loaded.pending.clear();
+    const subscriptions = [...loaded.subscriptions.values()];
+    loaded.subscriptions.clear();
+    for (const subscription of subscriptions) subscription.closed(message);
  }

  private appendLog(pluginId: string, stream: PluginLogEntry["stream"], message: string): void {
@@ -0,0 +1,55 @@
+import { defineSubscription } from "@getpaseo/plugin";
+import { subscribePluginRpc } from "@getpaseo/plugin/client/host";
+import type { PluginSubscriptionContext } from "@getpaseo/plugin/server";
+import type { PaseoApi } from "@getpaseo/client";
+import { expect, test } from "vitest";
+import { z } from "zod";
+import type { PluginProcessMessage } from "./plugin-process-protocol.js";
+import { PluginSubscriptionHandlers } from "./subscription-handlers.js";
+
+const lengths = defineSubscription({
+  name: "lengths",
+  input: z.object({ times: z.string().transform(Number) }),
+  event: z.string().transform((text) => text.length),
+});
+
+/** The plugin process's handlers wired to the client helper through the channel between them. */
+function connect(handler: (input: unknown, context: PluginSubscriptionContext) => unknown) {
+  const handlers = new PluginSubscriptionHandlers({
+    send: (message) => deliver(message),
+    sendAndWait: async (message) => deliver(message),
+  });
+  handlers.register(lengths, handler);
+  let observer: Parameters<Parameters<typeof subscribePluginRpc>[1]>[2] | null = null;
+  function deliver(message: PluginProcessMessage) {
+    if (message.type === "subscription.event") observer?.event(message.event);
+    if (message.type === "subscription.closed") observer?.closed();
+  }
+  return (method: string, input: unknown, next: NonNullable<typeof observer>) => {
+    observer = next;
+    handlers.start(
+      { type: "subscribe", requestId: "r", subscriptionId: "s", method, input },
+      {} as PaseoApi,
+    );
+    return Object.assign(() => handlers.release("s"), {
+      ready: Promise.resolve(),
+      release: async () => handlers.release("s"),
+    });
+  };
+}
+
+test("an event schema that transforms is applied once, on the receiving side", async () => {
+  const handled: unknown[] = [];
+  const received: number[] = [];
+  const closed: unknown[] = [];
+  const subscribe = connect(async (input, { push }) => {
+    handled.push(input);
+    await push("abc");
+  });
+  subscribePluginRpc(lengths, subscribe, { times: "2" }, (length) => received.push(length), {
+    onClosed: (result) => closed.push(result),
+  });
+  await expect.poll(() => received).toEqual([3]);
+  expect(handled).toEqual([{ times: 2 }]);
+  expect(closed).toEqual([{}]);
+});
@@ -0,0 +1,132 @@
+import type { PluginSubscriptionContract } from "@getpaseo/plugin";
+import type { PluginSubscriptionContext } from "@getpaseo/plugin/server";
+import type { PaseoApi } from "@getpaseo/client";
+import { assertPluginWireValue } from "@getpaseo/protocol/binary-frames/index";
+import type { PluginProcessMessage, PluginProcessRequest } from "./plugin-process-protocol.js";
+
+export type PluginSubscriptionHandler = (
+  input: unknown,
+  context: PluginSubscriptionContext,
+) => unknown;
+
+interface ActiveSubscription {
+  controller: AbortController;
+}
+
+interface SubscriptionChannel {
+  send(message: PluginProcessMessage): void;
+  sendAndWait(message: PluginProcessMessage): Promise<void>;
+}
+
+type SubscribeRequest = Extract<PluginProcessRequest, { type: "subscribe" }>;
+
+/**
+ * The plugin subprocess side of subscriptions: one handler call per subscriber, an abort signal
+ * that fires on release, and `push` that validates each event and resolves when it left the
+ * process.
+ */
+export class PluginSubscriptionHandlers {
+  private readonly registered = new Map<
+    string,
+    { contract: PluginSubscriptionContract; handler: PluginSubscriptionHandler }
+  >();
+  private readonly active = new Map<string, ActiveSubscription>();
+  private readonly channel: SubscriptionChannel;
+
+  constructor(channel: SubscriptionChannel) {
+    this.channel = channel;
+  }
+
+  has(name: string): boolean {
+    return this.registered.has(name);
+  }
+
+  register(contract: PluginSubscriptionContract, handler: PluginSubscriptionHandler): void {
+    this.registered.set(contract.name, { contract, handler });
+  }
+
+  catalog(): string[] {
+    return [...this.registered.keys()].sort();
+  }
+
+  start(message: SubscribeRequest, paseo: PaseoApi): void {
+    const { requestId, subscriptionId } = message;
+    const registered = this.registered.get(message.method);
+    if (!registered) {
+      this.channel.send({
+        type: "error",
+        requestId,
+        error: `Unknown subscription: ${message.method}`,
+      });
+      return;
+    }
+    const parsed = registered.contract.input.safeParse(message.input);
+    if (!parsed.success) {
+      this.channel.send({
+        type: "error",
+        requestId,
+        error: `Invalid input for ${message.method}: ${parsed.error.message}`,
+      });
+      return;
+    }
+    const controller = new AbortController();
+    const { signal } = controller;
+    let acknowledge = () => {};
+    const acknowledged = new Promise<void>((resolve) => {
+      acknowledge = resolve;
+    });
+    const push = async (event: unknown): Promise<void> => {
+      if (signal.aborted) return;
+      // The receiver parses into the schema's output; sending that output would parse it twice.
+      registered.contract.event.parse(event);
+      assertPluginWireValue(event, "event");
+      // Events pushed while the handler runs follow the acknowledgement.
+      await acknowledged;
+      if (signal.aborted) return;
+      await this.channel.sendAndWait({ type: "subscription.event", subscriptionId, event });
+    };
+    this.active.set(subscriptionId, { controller });
+    let result: unknown;
+    try {
+      result = registered.handler(parsed.data, { paseo, push, signal });
+    } catch (error) {
+      this.active.delete(subscriptionId);
+      controller.abort();
+      this.channel.send({ type: "error", requestId, error: describeError(error) });
+      return;
+    }
+    this.channel.send({ type: "result", requestId, output: null });
+    acknowledge();
+    if (result instanceof Promise)
+      result.then(
+        () => this.close(subscriptionId),
+        (error: unknown) => this.close(subscriptionId, describeError(error)),
+      );
+  }
+
+  /** The daemon released the subscription; it needs no closed message back. */
+  release(subscriptionId: string): void {
+    const active = this.active.get(subscriptionId);
+    if (!active) return;
+    this.active.delete(subscriptionId);
+    active.controller.abort();
+  }
+
+  releaseAll(): void {
+    for (const subscriptionId of this.active.keys()) this.release(subscriptionId);
+  }
+
+  private close(subscriptionId: string, error?: string): void {
+    if (!this.active.has(subscriptionId)) return;
+    this.release(subscriptionId);
+    this.channel.send({
+      type: "subscription.closed",
+      subscriptionId,
+      ...(error === undefined ? {} : { error }),
+    });
+  }
+}
+
+function describeError(error: unknown): string {
+  return error instanceof Error ? error.message : String(error);
+}
@@ -348,6 +348,7 @@ interface SessionForTestOptions {
  messages?: unknown[];
  targetedMessages?: Array<{ source: object; message: SessionOutboundMessage }>;
  binaryMessages?: Uint8Array[];
+  clientCapabilities?: SessionOptions["clientCapabilities"];
  pluginRuntime?: SessionOptions["pluginRuntime"];
  orchestrationSkills?: SessionOptions["orchestrationSkills"];
  workspaceLabelService?: WorkspaceLabelService;
@@ -402,6 +403,7 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session {
            messages.push(message),
        }),
    onBinaryMessage: createBinaryMessageHandler(options.binaryMessages),
+    clientCapabilities: options.clientCapabilities,
    logger,
    downloadTokenStore: options.downloadTokenStore ?? asDownloadTokenStore(),
    pushNotifications: options.pushNotifications ?? asPushNotifications(),
@@ -601,6 +603,64 @@ test("routes plugin requests and releases its owned catalog subscription on clea
  expect(releasePluginSubscription).toHaveBeenCalledOnce();
 });

+test("losing daemon.manage releases the session's plugin subscriptions", async () => {
+  const messages: SessionOutboundMessage[] = [];
+  const release = vi.fn();
+  const pluginRuntime = {
+    subscribe: () => () => {},
+    subscribePluginRpc: () => ({ ready: Promise.resolve(), attach: () => {}, release }),
+  } as unknown as NonNullable<SessionOptions["pluginRuntime"]>;
+  const session = createSessionForTest({ messages, pluginRuntime });
+
+  await session.handleMessage({
+    type: "plugin.rpc.subscribe.request",
+    requestId: "watch",
+    pluginId: "example",
+    method: "screen.capture",
+    input: {},
+  });
+  expect(messages.map((message) => message.type)).toContain("plugin.rpc.subscribe.response");
+  session.setPermissions(["daemon.read"]);
+
+  // Bytes have no outbound permission check, so the stream itself has to end.
+  await vi.waitFor(() => expect(release).toHaveBeenCalledOnce());
+});
+
+test("losing daemon.manage during a plugin RPC withholds its JSON and binary results", async () => {
+  const messages: SessionOutboundMessage[] = [];
+  const binaryMessages: Uint8Array[] = [];
+  const outputs = new Map<string, (output: unknown) => void>();
+  const pluginRuntime = {
+    subscribe: () => () => {},
+    invokePluginRpc: (_pluginId: string, method: string) =>
+      new Promise((resolve) => outputs.set(method, resolve)),
+  } as unknown as NonNullable<SessionOptions["pluginRuntime"]>;
+  const session = createSessionForTest({
+    messages,
+    binaryMessages,
+    pluginRuntime,
+    clientCapabilities: { [CLIENT_CAPS.pluginBinaryPayloads]: true },
+  });
+  const invoke = (method: string) =>
+    session.handleMessage({
+      type: "plugin.rpc.invoke.request",
+      requestId: method,
+      pluginId: "example",
+      method,
+      input: {},
+    });
+
+  const pending = [invoke("capture.json"), invoke("capture.bytes")];
+  await vi.waitFor(() => expect(outputs.size).toBe(2));
+  session.setPermissions(["daemon.read"]);
+  outputs.get("capture.json")?.({ width: 1 });
+  outputs.get("capture.bytes")?.(new Uint8Array([1, 2, 3]));
+  await Promise.all(pending);
+
+  expect(messages.filter((message) => message.type === "plugin.rpc.invoke.response")).toEqual([]);
+  expect(binaryMessages).toEqual([]);
+});
+
 test("leaves out the client bundles a catalog request already holds", async () => {
  const messages: SessionOutboundMessage[] = [];
  const pluginRuntime: NonNullable<SessionOptions["pluginRuntime"]> = {
@@ -50,6 +50,11 @@ import type {
 import { TerminalSessionController } from "../terminal/terminal-session-controller.js";
 import type { TerminalActivity } from "@getpaseo/protocol/terminal-activity";
 import type { BinaryFrame } from "@getpaseo/protocol/binary-frames/index";
+import {
+  encodePluginEventFrame,
+  encodePluginOutputFrame,
+  PluginFrameOpcode,
+} from "@getpaseo/protocol/binary-frames/index";
 import { CursorError } from "./pagination/cursor.js";
 import { SortablePager, type SortSpec } from "./pagination/sortable-pager.js";
 import { matchesAgentHistoryQuery } from "./agent-history-search.js";
@@ -521,6 +526,11 @@ export interface SessionOptions {
    subscribeSettings?(listener: (pluginId: string, settingsId: string) => void): () => void;
    catalog(): Array<{ id: string; clientBundle: string; clientBundleHash?: string }>;
    invokePluginRpc(pluginId: string, method: string, input: unknown): Promise<unknown>;
+    subscribePluginRpc(
+      pluginId: string,
+      method: string,
+      input: unknown,
+    ): import("./plugins/runtime-subscription.js").PluginRuntimeSubscription;
    listUsageReports(options?: {
      forceRefresh?: boolean;
      reportIds?: string[];
@@ -926,7 +936,7 @@ export class Session {
    this.workspaceFilesSession = new WorkspaceFilesSession({
      host: {
        emit: (msg, source) => this.emitForSource(msg, source),
-        emitBinary: (frame, source) => this.emitBinaryForFileTransfer(frame, source),
+        emitBinary: (frame, source) => this.emitBinaryReply(frame, source),
        hasBinaryChannel: () => this.onBinaryMessage !== null,
      },
      downloadTokenStore,
@@ -2221,13 +2231,15 @@ export class Session {
      this.peakInflightRequests = this.inflightRequests;
    }
    try {
-      this.sessionLogger.trace(
-        {
-          messageType: msg.type,
-          payloadBytes: JSON.stringify(msg).length,
-        },
-        "agent.session.inbound",
-      );
+      // Stringifying a binary plugin input costs its size; test logger stubs lack isLevelEnabled.
+      if (this.sessionLogger.isLevelEnabled?.("trace"))
+        this.sessionLogger.trace(
+          {
+            messageType: msg.type,
+            payloadBytes: JSON.stringify(msg).length,
+          },
+          "agent.session.inbound",
+        );
      if (!this.authorization.allowsInbound(msg)) {
        const requestId = sessionRequestId(msg);
        if (requestId) {
@@ -2292,6 +2304,17 @@ export class Session {
          this.sessionLogger.error({ err: error }, "Failed to release revoked browser hosting"),
        );
    }
+    // Binary plugin events skip the outbound permission check, so revocation ends the streams.
+    if (!this.authorization.allowsPermission("daemon.manage")) {
+      void this.delivery
+        .releaseFamily("plugin-rpc")
+        .catch((error) =>
+          this.sessionLogger.error(
+            { err: error },
+            "Failed to release revoked plugin subscriptions",
+          ),
+        );
+    }
  }

  public getPermissions(): DaemonPermission[] {
@@ -2454,6 +2477,7 @@ export class Session {
      this.dispatchOrchestrationSkillsMessage(msg) ??
      this.dispatchPluginDirectoryMessage(msg) ??
      this.dispatchPluginMessage(msg) ??
+      this.dispatchPluginRpcMessage(msg) ??
      this.dispatchHostServiceMessage(msg) ??
      this.dispatchMiscMessage(msg);
    if (promise) await promise;
@@ -2621,21 +2645,98 @@ export class Session {
      });
      return undefined;
    }
+    return undefined;
+  }
+
+  private dispatchPluginRpcMessage(msg: SessionInboundMessage): Promise<void> | undefined {
    if (msg.type === "plugin.rpc.invoke.request") {
      if (!this.pluginRuntime) throw new Error("Plugin service is unavailable");
      return this.pluginRuntime
        .invokePluginRpc(msg.pluginId, msg.method, msg.input)
-        .then((output) => {
-          this.emit({
-            type: "plugin.rpc.invoke.response",
-            payload: { requestId: msg.requestId, output },
-          });
-          return undefined;
-        });
+        .then((output) => this.replyPluginRpc(msg, output));
    }
+    if (msg.type === "plugin.rpc.subscribe.request") return this.subscribePluginRpc(msg);
    return undefined;
  }

+  /** A `Uint8Array` output is the binary form of `plugin.rpc.invoke.response`. */
+  private async replyPluginRpc(
+    request: Extract<SessionInboundMessage, { type: "plugin.rpc.invoke.request" }>,
+    output: unknown,
+  ): Promise<void> {
+    const response: SessionOutboundMessage = {
+      type: "plugin.rpc.invoke.response",
+      payload: { requestId: request.requestId, output },
+    };
+    if (!(output instanceof Uint8Array)) {
+      this.emit(response);
+      return;
+    }
+    // The frame is the binary form of the response, so it needs the response's permission.
+    if (!this.authorization.allowsOutbound(response)) return;
+    if (!this.supports(CLIENT_CAPS.pluginBinaryPayloads) || this.onBinaryMessage === null)
+      throw new Error(
+        `Update Paseo to receive the binary output of plugin RPC ${request.pluginId}.${request.method}`,
+      );
+    await this.emitBinaryReply(
+      encodePluginOutputFrame(request.requestId, output),
+      this.delivery.currentSource,
+    );
+  }
+
+  private async subscribePluginRpc(
+    request: Extract<SessionInboundMessage, { type: "plugin.rpc.subscribe.request" }>,
+  ): Promise<void> {
+    if (!this.pluginRuntime) throw new Error("Plugin service is unavailable");
+    const subscription = this.pluginRuntime.subscribePluginRpc(
+      request.pluginId,
+      request.method,
+      request.input,
+    );
+    const owner = this.delivery.begin("plugin-rpc", undefined, () => subscription.release());
+    try {
+      await subscription.ready;
+    } catch (error) {
+      await owner.release();
+      throw error;
+    }
+    if (owner.signal.aborted) return;
+    this.emit({
+      type: "plugin.rpc.subscribe.response",
+      payload: { requestId: request.requestId, subscriptionId: owner.responseId },
+    });
+    const close = (reason?: string) => {
+      owner.emit({
+        type: "plugin.rpc.closed",
+        payload: {
+          subscriptionId: owner.responseId,
+          ...(reason === undefined ? {} : { error: reason }),
+        },
+      });
+      void owner.release().catch(() => undefined);
+    };
+    subscription.attach({
+      event: (event) => {
+        const message: SessionOutboundMessage = {
+          type: "plugin.rpc.event",
+          payload: { subscriptionId: owner.responseId, event },
+        };
+        if (!(event instanceof Uint8Array)) {
+          owner.emit(message);
+          return;
+        }
+        if (!this.authorization.allowsOutbound(message)) return;
+        if (this.onBinaryMessage === null) {
+          subscription.release();
+          close("This connection cannot carry binary plugin events");
+          return;
+        }
+        owner.emitBinary(encodePluginEventFrame(owner.responseId, event));
+      },
+      closed: close,
+    });
+  }
+
  private dispatchPluginDirectoryMessage(msg: SessionInboundMessage): Promise<void> | undefined {
    if (msg.type === "plugin.source.install.request") {
      if (!this.pluginRuntime) throw new Error("Plugin service is unavailable");
@@ -3272,6 +3373,22 @@ export class Session {
      this.connectedHostSession?.binary(binaryFrame.frame, source);
      return;
    }
+    if (binaryFrame.kind === "plugin") {
+      const frame = binaryFrame.frame;
+      // The binary form of plugin.rpc.invoke.request; it authorizes and replies like the JSON form.
+      if (frame.opcode !== PluginFrameOpcode.Input) return;
+      await this.handleMessage(
+        {
+          type: "plugin.rpc.invoke.request",
+          requestId: frame.requestId,
+          pluginId: frame.pluginId,
+          method: frame.method,
+          input: frame.payload,
+        },
+        source,
+      );
+      return;
+    }
    if (!this.authorization.allowsPermission("workspace.write")) {
      return;
    }
@@ -8690,9 +8807,9 @@ export class Session {
    }
  }

-  private async emitBinaryForFileTransfer(frame: Uint8Array, source?: object): Promise<void> {
+  private async emitBinaryReply(frame: Uint8Array, source?: object): Promise<void> {
    if (source && this.onBinaryMessageToSource) {
-      this.delivery.authorizeFileReply(frame, source);
+      this.delivery.authorizeBinaryReply(frame, source);
      await this.onBinaryMessageToSource(source, frame);
      return;
    }
@@ -56,6 +56,11 @@ export interface OwnedSubscription {
  release(): Promise<void>;
 }

+const BINARY_REPLY_REQUESTS: ReadonlySet<string> = new Set([
+  "file_explorer_request",
+  "plugin.rpc.invoke.request",
+]);
+
 // One registration per context kind; per-session registrations outlive closed sessions in Node.
 const requests = new AsyncLocalStorage<RequestOwner>();
 const projectionSource = new AsyncLocalStorage<ProjectionOwner>();
@@ -394,12 +399,13 @@ export class SessionDelivery {
    if (owner) await this.releaseOwner(owner);
  }

-  authorizeFileReply(frame: Uint8Array, socket: object): void {
+  /** Lets a binary frame answer the current request, for requests whose reply can be bytes. */
+  authorizeBinaryReply(frame: Uint8Array, socket: object): void {
    const owner = this.currentRequest;
    if (
      owner?.active &&
      owner.source.socket === socket &&
-      owner.request.type === "file_explorer_request"
+      BINARY_REPLY_REQUESTS.has(owner.request.type)
    )
      this.proofs.set(frame, owner);
  }
@@ -2066,6 +2066,8 @@ export class VoiceAssistantWebSocketServer {
        pluginThemes: true,
        pluginSettings: true,
        pluginTimelineItems: true,
+        // COMPAT(pluginSubscriptions): added in v0.15.0, remove gate after 2027-05-02.
+        pluginSubscriptions: true,
        // COMPAT(skillManagement): added in v0.4.0, remove gate after 2027-08-16.
        skillManagement: true,
        // COMPAT(idleRuntimeCleanupSetting): added in v0.13.0, remove gate after 2027-04-26.