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.

packages/server/src/server/plugins/plugin-process.ts

@@ -135,9 +135,8 @@ export function createPluginWorker(options: {
     handler: (input: unknown, context: never) => unknown,
   ) => {
     if ("event" in contract) {
-      registerSubscription(contract, (input, context) =>
-        handler(contract.input.parse(input), context as never),
-      );
+      // 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));

packages/server/src/server/plugins/plugin-subscriptions.e2e.test.ts

@@ -34,6 +34,11 @@ const nestedPush = defineSubscription({
   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({}),
@@ -73,6 +78,9 @@ export default function contribute(server: PluginServerContext) {
       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) }));
@@ -200,6 +208,19 @@ test("a subscription delivers JSON and binary events in order and release aborts
   }
 }, 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;