From 63302287feb3405fdd499161547e0d80566623c1 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 2 Nov 2026 15:42:39 +0700 Subject: [PATCH] Fix plugin RPCs with schema transforms (#8550) * fix(plugins): parse RPC values once at the receiver Sending parsed values caused transforms to run again on incompatible schema outputs. Keep the wire values in schema-input form and parse only where consumed. * fix(plugins): let omitted RPC inputs reach schema defaults * fix(plugins): accept omitted RPC outputs for client defaults --- packages/plugin/src/client/host.test.ts | 26 ++++--- packages/plugin/src/rpc.ts | 3 +- packages/protocol/src/messages.ts | 4 +- .../tests/validation/ws-outbound.test.ts | 7 +- .../src/server/plugins/plugin-process.ts | 3 +- .../plugins/plugin-subscriptions.e2e.test.ts | 74 +++++++++++++++++++ .../src/server/plugins/runtime.posix.test.ts | 6 +- public-docs/plugins/reference.md | 2 +- 8 files changed, 105 insertions(+), 20 deletions(-) diff --git a/packages/plugin/src/client/host.test.ts b/packages/plugin/src/client/host.test.ts index 5b5333fcd1..aad5eed8fd 100644 --- a/packages/plugin/src/client/host.test.ts +++ b/packages/plugin/src/client/host.test.ts @@ -24,17 +24,23 @@ describe("plugin RPC contracts", () => { expectTypeOf>().toEqualTypeOf<{ value: number }>(); }); - it("validates input before invoking the daemon", async () => { - let invocations = 0; - const invoke = async () => { - invocations += 1; - return { value: 2 }; + it("sends input unchanged for the plugin to parse", async () => { + let parses = 0; + const transformed = defineRpc({ + name: "transformed", + input: z.string().transform((text) => { + parses += 1; + return Number(text); + }), + output: z.number(), + }); + const invoke = async (_method: string, input: unknown) => { + expect(input).toBe("2"); + expect(parses).toBe(0); + return transformed.input.parseAsync(input); }; - - await expect(callPluginRpc(contract, invoke, { value: "wrong" })).rejects.toBeInstanceOf( - z.ZodError, - ); - expect(invocations).toBe(0); + await expect(callPluginRpc(transformed, invoke, "2")).resolves.toBe(2); + expect(parses).toBe(1); }); it("validates output before resolving", async () => { diff --git a/packages/plugin/src/rpc.ts b/packages/plugin/src/rpc.ts index 856b11de61..fd550b101f 100644 --- a/packages/plugin/src/rpc.ts +++ b/packages/plugin/src/rpc.ts @@ -39,8 +39,7 @@ export async function callPluginRpc Promise, input: PluginWireInput>, ): Promise> { - const parsedInput = await contract.input.parseAsync(input); - const output = await invoke(contract.name, parsedInput); + const output = await invoke(contract.name, input); return contract.output.parseAsync(output); } diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 1934d1072b..5d931334d0 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -1603,7 +1603,7 @@ export const PluginRpcInvokeRequestSchema = z.object({ requestId: z.string(), pluginId: PluginIdSchema, method: z.string().min(1), - input: z.unknown(), + input: z.unknown().optional(), }); export const PluginRpcSubscribeRequestSchema = z.object({ @@ -6969,7 +6969,7 @@ export const PluginRpcInvokeResponseSchema = z.object({ type: z.literal("plugin.rpc.invoke.response"), payload: z.object({ requestId: z.string(), - output: z.unknown(), + output: z.unknown().optional(), }), }); diff --git a/packages/protocol/tests/validation/ws-outbound.test.ts b/packages/protocol/tests/validation/ws-outbound.test.ts index ed147d89ad..18b0cd999f 100644 --- a/packages/protocol/tests/validation/ws-outbound.test.ts +++ b/packages/protocol/tests/validation/ws-outbound.test.ts @@ -5,7 +5,7 @@ import { fileURLToPath, pathToFileURL } from "node:url"; import { createJiti } from "jiti"; import type { z } from "zod"; import { describe, expect, it } from "vitest"; -import { WSOutboundMessageSchema } from "../../src/messages.js"; +import { PluginRpcInvokeResponseSchema, WSOutboundMessageSchema } from "../../src/messages.js"; import { validateWSOutboundMessage } from "../../src/validation/ws-outbound.js"; interface GeneratedSchema { @@ -401,3 +401,8 @@ const SourceSchema = z.object({ }); }); }); + +it("accepts omitted RPC output so the receiving client can apply schema defaults", () => { + const message = { type: "plugin.rpc.invoke.response", payload: { requestId: "default-output" } }; + expect(PluginRpcInvokeResponseSchema.safeParse(message).success).toBe(true); +}); diff --git a/packages/server/src/server/plugins/plugin-process.ts b/packages/server/src/server/plugins/plugin-process.ts index 9afbddba5a..1c0d6bc476 100644 --- a/packages/server/src/server/plugins/plugin-process.ts +++ b/packages/server/src/server/plugins/plugin-process.ts @@ -139,7 +139,7 @@ export function createPluginWorker(options: { registerSubscription(contract, handler as PluginSubscriptionHandler); return; } - register(contract, (input, context) => handler(contract.input.parse(input), context as never)); + register(contract, (input, context) => handler(input, context as never)); }) as PluginServerContext["handle"]; function register(contract: PluginRpcContract, handler: RpcHandler): void { @@ -593,7 +593,6 @@ export function createPluginWorker(options: { if (!paseo) throw new Error("Plugin Paseo API is unavailable"); return registered.handler(input, { paseo }); }) - .then((output) => registered.contract.output.parseAsync(output)) .then((output: unknown) => { assertPluginWireValue(output, "output"); return output; diff --git a/packages/server/src/server/plugins/plugin-subscriptions.e2e.test.ts b/packages/server/src/server/plugins/plugin-subscriptions.e2e.test.ts index e5f427eae2..97af53c364 100644 --- a/packages/server/src/server/plugins/plugin-subscriptions.e2e.test.ts +++ b/packages/server/src/server/plugins/plugin-subscriptions.e2e.test.ts @@ -8,6 +8,9 @@ import { type WebSocketLike, } from "@getpaseo/client/internal/daemon-client"; import { afterEach, expect, test } from "vitest"; +import { defineRpc } from "@getpaseo/plugin"; +import { callPluginRpc } from "@getpaseo/plugin/client/host"; +import { z } from "zod"; import { resolveDaemonVersion } from "../daemon-version.js"; import { createTestPaseoDaemon, type TestPaseoDaemon } from "../test-utils/paseo-daemon.js"; @@ -39,6 +42,13 @@ const repeated = defineSubscription({ input: z.object({ times: z.string().transform(Number) }), event: z.string().transform((text) => text.length), }); +const transformed = defineRpc({ + name: "transformed", + input: z.object({ when: z.string().transform(async (text) => new Date(text)) }), + output: z.object({ n: z.string().transform(async (text) => Number(text)) }), +}); +const defaulted = defineRpc({ name: "defaulted", input: z.string().default("x"), output: z.string() }); +const defaultedOutput = defineRpc({ name: "defaulted-output", input: z.object({}), output: z.string().default("x") }); const watchers = defineRpc({ name: "watchers", input: z.object({}), @@ -81,6 +91,9 @@ export default function contribute(server: PluginServerContext) { server.handle(repeated, async ({ times }, { push }) => { await push(typeof times === "number" ? "x".repeat(times) : "input arrived untransformed"); }); + server.handle(transformed, ({ when }) => ({ n: String(when.getUTCFullYear()) })); + server.handle(defaulted, (input) => input); + server.handle(defaultedOutput, () => undefined); server.handle(watchers, () => ({ starts, active })); server.handle(thumbnail, ({ size }) => Buffer.alloc(size, 7)); server.handle(nestedOutput, () => ({ data: new Uint8Array(1) })); @@ -352,3 +365,64 @@ export default function contribute(server: PluginServerContext) { await harness.close(); } }, 60_000); + +test("plain RPC parses async input and output transforms once at the receiver", async () => { + const transformed = defineRpc({ + name: "transformed", + input: z.object({ when: z.string().transform(async (text) => new Date(text)) }), + output: z.object({ n: z.string().transform(async (text) => Number(text)) }), + }); + + const harness = await startHarness(); + try { + await expect( + callPluginRpc( + transformed, + (method, input) => harness.client.invokePluginRpc(PLUGIN_ID, method, input), + { when: "2026-01-01T00:00:00.000Z" }, + ), + ).resolves.toEqual({ n: 2026 }); + } finally { + await harness.close(); + } +}); + +test("plain RPC applies a top-level input default in the plugin process", async () => { + const contract = defineRpc({ + name: "defaulted", + input: z.string().default("x"), + output: z.string(), + }); + const harness = await startHarness(); + try { + await expect( + callPluginRpc( + contract, + (method, input) => harness.client.invokePluginRpc(PLUGIN_ID, method, input), + undefined, + ), + ).resolves.toBe("x"); + } finally { + await harness.close(); + } +}); + +test("plain RPC applies a top-level output default on the client", async () => { + const contract = defineRpc({ + name: "defaulted-output", + input: z.object({}), + output: z.string().default("x"), + }); + const harness = await startHarness(); + try { + await expect( + callPluginRpc( + contract, + (method, input) => harness.client.invokePluginRpc(PLUGIN_ID, method, input), + {}, + ), + ).resolves.toBe("x"); + } finally { + await harness.close(); + } +}, 30_000); diff --git a/packages/server/src/server/plugins/runtime.posix.test.ts b/packages/server/src/server/plugins/runtime.posix.test.ts index bed5e98a00..1b2e7192c7 100644 --- a/packages/server/src/server/plugins/runtime.posix.test.ts +++ b/packages/server/src/server/plugins/runtime.posix.test.ts @@ -1853,7 +1853,7 @@ export function inspect() { void Surface; return {}; }`, await runtime.stopAll(); }); - it("rejects a handler result that does not match its RPC output schema", async () => { + it("forwards a handler result for the client to validate against its RPC output schema", async () => { const directory = await createPlugin( "invalid-output", `import { z } from "zod"; @@ -1872,7 +1872,9 @@ export default function contribute(plugin: any) { await runtime.startPlugin("invalid-output", directory); - await expect(runtime.invoke("invalid-output", "broken", {})).rejects.toThrow(); + await expect(runtime.invoke("invalid-output", "broken", {})).resolves.toEqual({ + value: "wrong", + }); await runtime.stopAll(); }); diff --git a/public-docs/plugins/reference.md b/public-docs/plugins/reference.md index 0f5b3eb40a..463cbd4957 100644 --- a/public-docs/plugins/reference.md +++ b/public-docs/plugins/reference.md @@ -2222,7 +2222,7 @@ export default function contribute(server: PluginServerContext) { } ``` -Inputs and outputs are validated on both sides. RPC names start with a lowercase letter and contain lowercase letters, numbers, dots, hyphens, or underscores. `useRpc()` returns a typed async function. Use TanStack Query for request state, caching, and mutations. +Inputs are parsed once in the plugin process; outputs are parsed once on the client. Send schema inputs from the client and return output-schema inputs from the handler; schema transforms run at the receiving side. RPC names start with a lowercase letter and contain lowercase letters, numbers, dots, hyphens, or underscores. `useRpc()` returns a typed async function. Use TanStack Query for request state, caching, and mutations. Backend handlers receive the same `PaseoApi` as `{ paseo }`. Their connection belongs to the subprocess and closes when the plugin stops. It does not subscribe to timelines or catalog events until plugin code subscribes. Follow the [SDK event contract](/docs/sdk/events) for cleanup and timeline replacements. Backend code can use Node APIs and dependencies installed in the plugin directory.