fix!: parse payloads once per boundary (validate on send, transmit original)

Implements D1 from the v8 review remediation spec. Previously both sides
of every payload boundary ran the same Standard Schema and the sender
transmitted the PARSED value; the receiver parsed again, so a transforming
schema (z.coerce.*, .transform(...)) was applied twice — silent data
corruption. Now each boundary parses exactly once, on the receiving side:
the sender still validates (failing early with the existing typed error)
but transmits the caller's ORIGINAL value, discarding the parsed result.

Send-side sites reclassified (validate, transmit original):
- client startWorkflow / executeWorkflow / signalWithStart workflow args
  (resolveDefinitionAndValidateInput no longer returns a parsed value)
- client signalWithStart signal args
- client handle signal/query/update proxy inputs (buildValidatedProxy)
- client schedule.create args
- worker workflow-side activity proxy inputs (both throwing and
  Result-shaped wrappers in activities-proxy.ts)
- worker child-workflow args (startChildWorkflow / executeChildWorkflow)
- worker continueAsNew args
- worker workflow output (declareWorkflow returns the implementation's
  original value after validating it)
- worker activity handler output (declareActivitiesHandler)
- worker query/update handler outputs (bindQueryHandler/bindUpdateHandler)
- worker contract-error data (contractErrorToApplicationFailure details[0])

Receive-side sites confirmed as the single boundary parse (unchanged):
- worker workflow/signal/query/update input handlers, activity handler
  input, middleware-substituted inputs (a patched input is raw from the
  pipeline's perspective and parsed once)
- client executeWorkflow / handle.result() / query / update result parsing
- worker parent-side child-workflow result parsing, activity-result
  parsing in the workflow proxy, contract-error rehydration

Adds transform-schema tests pinning the new wire format at unit level in
both packages plus an end-to-end integration case in the client suite,
and rewrites the old "intentional double-validation" rationale comments.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: btravers

packages/client/src/__tests__/client.spec.ts

@@ -158,6 +158,33 @@ describe("Client Package - Integration Tests", () => {
     });
   });
 
+  describe("Wire format (D1) — transforms apply exactly once per boundary", () => {
+    it("sends the original input, receiver parses once; output parsed once by the client", async ({
+      client,
+    }) => {
+      // GIVEN — input schema appends "!", output schema doubles. The client
+      // validates-and-discards on send; the workflow-side parse is the only
+      // input transform, and the client-side result parse is the only output
+      // transform. Double application would yield "hello!!" / 84.
+      const input = { text: "hello" };
+
+      // WHEN
+      const result = await client.executeWorkflow("transformWorkflow", {
+        workflowId: `transform-${Date.now()}`,
+        args: input,
+      });
+
+      // THEN
+      expect(result).toBeOk();
+      if (result.isOk()) {
+        expect(result.value).toEqual({
+          handlerSaw: "hello!", // exactly one "!" — parsed once, on receive
+          doubled: 42, // 21 doubled exactly once, by the client's parse
+        });
+      }
+    });
+  });
+
   describe("Workflow with Activities", () => {
     it("should execute workflow with activity", async ({ client }) => {
       // GIVEN

packages/client/src/__tests__/test.contract.ts

@@ -60,6 +60,22 @@ export const testContract = defineContract({
       },
     }),
 
+    // Workflow with transforming input/output schemas — exercises the D1
+    // wire format: the sender validates but transmits the ORIGINAL value;
+    // the receiver parses, so each transform applies exactly once.
+    transformWorkflow: defineWorkflow({
+      input: z.object({
+        text: z.string().transform((s) => `${s}!`),
+      }),
+      output: z.object({
+        // What the (receive-side-parsed) input looked like to the handler.
+        handlerSaw: z.string(),
+        // The handler returns the pre-transform number; only the client's
+        // receive-side parse doubles it.
+        doubled: z.number().transform((n) => n * 2),
+      }),
+    }),
+
     // Workflow with activities
     workflowWithActivity: defineWorkflow({
       input: z.object({

packages/client/src/__tests__/test.workflows.ts

@@ -7,6 +7,8 @@ import {
   defineUpdate,
 } from "@temporalio/workflow";
 
+import { testContract } from "./test.contract.js";
+
 // Define activity types manually based on the contract
 type Activities = {
   logMessage(args: { message: string }): Promise<{}>;
@@ -56,6 +58,22 @@ export async function interactiveWorkflow(args: { initialValue: number }) {
   };
 }
 
+// Mirrors what `@temporal-contract/worker`'s `declareWorkflow` does at the
+// D1 wire boundary: the client transmits the caller's ORIGINAL args, the
+// receiving side parses them exactly once, and the return value is handed to
+// Temporal untransformed (the client parses the output on receive).
+export async function transformWorkflow(args: { text: string }) {
+  const parsed = await testContract.workflows.transformWorkflow.input["~standard"].validate(args);
+  if (parsed.issues) {
+    throw new Error(`transformWorkflow input validation failed`);
+  }
+  const input = parsed.value as { text: string };
+  return {
+    handlerSaw: input.text,
+    doubled: 21,
+  };
+}
+
 export async function workflowWithActivity(args: { message: string }) {
   const processed = await activities.processMessage({ message: args.message });
   await activities.logMessage({ message: `Activity result: ${processed.processed}` });

packages/client/src/client.spec.ts

@@ -1320,6 +1320,201 @@ describe("TypedClient", () => {
   });
 });
 
+describe("TypedClient — wire format (validate on send, parse on receive)", () => {
+  // D1: each payload boundary parses exactly once, on the receiving side.
+  // The client validates what it sends (failing early with the typed
+  // validation error) but transmits the caller's ORIGINAL value; parsed
+  // results are only used on the receive side (workflow/query/update
+  // results). Transforming schemas make the two sides observable.
+  const transformContract = defineContract({
+    taskQueue: "wire-q",
+    workflows: {
+      transformer: defineWorkflow({
+        // Asymmetric transform: input type is `string`, parsed type is `number`.
+        input: z.string().transform((s) => s.length),
+        output: z.number().transform((n) => n * 2),
+        signals: {
+          ping: { input: z.string().transform((s) => s.length) },
+        },
+        queries: {
+          peek: {
+            input: z.string().transform((s) => s.length),
+            output: z.number().transform((n) => n * 2),
+          },
+        },
+        updates: {
+          poke: {
+            input: z.string().transform((s) => s.length),
+            output: z.number().transform((n) => n * 2),
+          },
+        },
+      }),
+    },
+  });
+
+  let wireClient: TypedClient<typeof transformContract>;
+
+  beforeEach(() => {
+    vi.clearAllMocks();
+    const rawClient = { workflow: mockWorkflow, schedule: mockSchedule } as unknown as Client;
+    wireClient = TypedClient.createOrThrow(transformContract, rawClient);
+  });
+
+  it("startWorkflow transmits the ORIGINAL args, not the parsed value", async () => {
+    mockWorkflow.start.mockResolvedValue({ workflowId: "wf-1" });
+
+    const result = await wireClient.startWorkflow("transformer", {
+      workflowId: "wf-1",
+      args: "hello",
+    });
+
+    expect(result).toBeOk();
+    expect(mockWorkflow.start).toHaveBeenCalledWith("transformer", {
+      workflowId: "wf-1",
+      taskQueue: "wire-q",
+      args: ["hello"], // original string — not 5 (the parsed length)
+    });
+  });
+
+  it("startWorkflow still rejects invalid input before dispatch", async () => {
+    const result = await wireClient.startWorkflow("transformer", {
+      workflowId: "wf-1",
+      // @ts-expect-error testing runtime validation
+      args: 42,
+    });
+
+    expect(result).toBeErr();
+    if (result.isErr()) {
+      expect(result.error).toBeInstanceOf(WorkflowValidationError);
+    }
+    expect(mockWorkflow.start).not.toHaveBeenCalled();
+  });
+
+  it("executeWorkflow transmits the ORIGINAL args and parses the result exactly once", async () => {
+    // The wire carries the producer's original (pre-transform) value; the
+    // client applies the output transform on receive.
+    mockWorkflow.execute.mockResolvedValue(21);
+
+    const result = await wireClient.executeWorkflow("transformer", {
+      workflowId: "wf-2",
+      args: "hello",
+    });
+
+    expect(mockWorkflow.execute).toHaveBeenCalledWith("transformer", {
+      workflowId: "wf-2",
+      taskQueue: "wire-q",
+      args: ["hello"],
+    });
+    expect(result).toBeOk();
+    if (result.isOk()) {
+      expect(result.value).toBe(42); // 21 doubled once — not 84
+    }
+  });
+
+  it("signalWithStart transmits the ORIGINAL workflow and signal args", async () => {
+    mockWorkflow.signalWithStart.mockResolvedValue({
+      workflowId: "wf-3",
+      signaledRunId: "run-1",
+    });
+
+    const result = await wireClient.signalWithStart("transformer", {
+      workflowId: "wf-3",
+      args: "hello",
+      signalName: "ping",
+      signalArgs: "hey",
+    });
+
+    expect(result).toBeOk();
+    expect(mockWorkflow.signalWithStart).toHaveBeenCalledWith("transformer", {
+      workflowId: "wf-3",
+      taskQueue: "wire-q",
+      args: ["hello"],
+      signal: "ping",
+      signalArgs: ["hey"], // original string — not 3
+    });
+  });
+
+  it("handle.result() parses what the handle returns (receive side)", async () => {
+    const rawHandle = {
+      workflowId: "wf-4",
+      result: vi.fn().mockResolvedValue(21),
+      query: vi.fn(),
+      signal: vi.fn(),
+      executeUpdate: vi.fn(),
+    };
+    mockWorkflow.getHandle.mockReturnValue(rawHandle);
+
+    const handleResult = await wireClient.getHandle("transformer", "wf-4");
+    if (!handleResult.isOk()) throw new Error("expected Ok");
+    const result = await handleResult.value.result();
+
+    expect(result).toBeOk();
+    if (result.isOk()) {
+      expect(result.value).toBe(42);
+    }
+  });
+
+  it("handle.signals.* transmits the ORIGINAL signal args", async () => {
+    const rawHandle = {
+      workflowId: "wf-5",
+      result: vi.fn(),
+      query: vi.fn(),
+      signal: vi.fn().mockResolvedValue(undefined),
+      executeUpdate: vi.fn(),
+    };
+    mockWorkflow.getHandle.mockReturnValue(rawHandle);
+
+    const handleResult = await wireClient.getHandle("transformer", "wf-5");
+    if (!handleResult.isOk()) throw new Error("expected Ok");
+    const result = await handleResult.value.signals.ping("hey");
+
+    expect(result).toBeOk();
+    expect(rawHandle.signal).toHaveBeenCalledWith("ping", "hey");
+  });
+
+  it("handle.queries.* transmits the ORIGINAL input and parses the result once", async () => {
+    const rawHandle = {
+      workflowId: "wf-6",
+      result: vi.fn(),
+      query: vi.fn().mockResolvedValue(21),
+      signal: vi.fn(),
+      executeUpdate: vi.fn(),
+    };
+    mockWorkflow.getHandle.mockReturnValue(rawHandle);
+
+    const handleResult = await wireClient.getHandle("transformer", "wf-6");
+    if (!handleResult.isOk()) throw new Error("expected Ok");
+    const result = await handleResult.value.queries.peek("hey");
+
+    expect(rawHandle.query).toHaveBeenCalledWith("peek", "hey");
+    expect(result).toBeOk();
+    if (result.isOk()) {
+      expect(result.value).toBe(42);
+    }
+  });
+
+  it("handle.updates.* transmits the ORIGINAL input and parses the result once", async () => {
+    const rawHandle = {
+      workflowId: "wf-7",
+      result: vi.fn(),
+      query: vi.fn(),
+      signal: vi.fn(),
+      executeUpdate: vi.fn().mockResolvedValue(21),
+    };
+    mockWorkflow.getHandle.mockReturnValue(rawHandle);
+
+    const handleResult = await wireClient.getHandle("transformer", "wf-7");
+    if (!handleResult.isOk()) throw new Error("expected Ok");
+    const result = await handleResult.value.updates.poke("hey");
+
+    expect(rawHandle.executeUpdate).toHaveBeenCalledWith("poke", { args: ["hey"] });
+    expect(result).toBeOk();
+    if (result.isOk()) {
+      expect(result.value).toBe(42);
+    }
+  });
+});
+
 describe("TypedClient — workflow contract errors", () => {
   const erroredContract = defineContract({
     taskQueue: "test-queue",