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.

packages/server/src/server/plugins/runtime.posix.test.ts

@@ -1424,7 +1424,9 @@ export default function contribute(server: { registerProvider(provider: Provider
       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") });
+      await expect(rejection).resolves.toMatchObject({
+        message: expect.stringContaining("timed out"),
+      });
       const subscribe = sent.find((message) => message.type === "subscribe");
       expect(sent).toContainEqual({
         type: "unsubscribe",

packages/server/src/server/session.test.ts

@@ -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(),
@@ -624,6 +626,41 @@ test("losing daemon.manage releases the session's plugin subscriptions", async (
   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"]> = {

packages/server/src/server/session.ts

@@ -2309,7 +2309,10 @@ export class Session {
       void this.delivery
         .releaseFamily("plugin-rpc")
         .catch((error) =>
-          this.sessionLogger.error({ err: error }, "Failed to release revoked plugin subscriptions"),
+          this.sessionLogger.error(
+            { err: error },
+            "Failed to release revoked plugin subscriptions",
+          ),
         );
     }
   }
@@ -2661,13 +2664,16 @@ export class Session {
     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({
-        type: "plugin.rpc.invoke.response",
-        payload: { requestId: request.requestId, output },
-      });
+      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}`,
@@ -2711,13 +2717,15 @@ export class Session {
     };
     subscription.attach({
       event: (event) => {
+        const message: SessionOutboundMessage = {
+          type: "plugin.rpc.event",
+          payload: { subscriptionId: owner.responseId, event },
+        };
         if (!(event instanceof Uint8Array)) {
-          owner.emit({
-            type: "plugin.rpc.event",
-            payload: { subscriptionId: owner.responseId, event },
-          });
+          owner.emit(message);
           return;
         }
+        if (!this.authorization.allowsOutbound(message)) return;
         if (this.onBinaryMessage === null) {
           subscription.release();
           close("This connection cannot carry binary plugin events");