diff --git a/packages/server/src/server/plugins/runtime.posix.test.ts b/packages/server/src/server/plugins/runtime.posix.test.ts index 01a7cf5217..bed5e98a00 100644 --- a/packages/server/src/server/plugins/runtime.posix.test.ts +++ b/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", diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index acd292e895..4f02efad2e 100644 --- a/packages/server/src/server/session.test.ts +++ b/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 void>(); + const pluginRuntime = { + subscribe: () => () => {}, + invokePluginRpc: (_pluginId: string, method: string) => + new Promise((resolve) => outputs.set(method, resolve)), + } as unknown as NonNullable; + 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 = { diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index c6cae8b74c..81c22082c7 100644 --- a/packages/server/src/server/session.ts +++ b/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, output: unknown, ): Promise { + 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");