feat(chat): runtime clientData validation for custom agents

.changeset/quiet-chats-validate.md

@@ -0,0 +1,5 @@
+---
+"@trigger.dev/sdk": patch
+---
+
+Custom chat agents now validate and parse client data declared with `chat.withClientData({ schema })` before passing it to agent code.

docs/ai-chat/client-protocol.mdx

@@ -771,7 +771,7 @@ type ChatTaskWirePayload<TMessage extends UIMessage = UIMessage, TMetadata = unk
 ```
 
 <Note>
-  **`metadata` is the wire envelope for `clientData`.** The agent's `clientData` (typed via `chat.withClientData({ schema })`) is read from this field at run boot. If the agent declares e.g. `{ userId: string, model?: string }`, then every `kind: "message"` payload — and the `triggerConfig.basePayload` you sent at session create — must carry a matching `metadata.userId`. The agent rejects messages whose metadata fails schema validation.
+  **`metadata` is the wire envelope for `clientData`.** The agent's `clientData` (typed via `chat.withClientData({ schema })`) is read from this field at run boot. If the agent declares e.g. `{ userId: string, model?: string }`, then the `triggerConfig.basePayload` you sent at session create and every non-close `kind: "message"` payload must carry a matching `metadata.userId`. Invalid metadata is not passed to agent code. Async reads produce an error chunk followed by `turn-complete`; raw `chat.messages.on()` subscriptions use `onClientDataValidationError` and the task log so they do not end an active response.
 </Note>
 
 ### Sending a message

docs/ai-chat/custom-agents.mdx

@@ -19,61 +19,93 @@ Inside the wrapper, pick one of two loop styles:
 - **[Managed loop](#managed-loop-chatcreatesession)** — `chat.createSession()` yields turns; the SDK handles stop signals, accumulation, idle suspend/resume, and turn-complete signaling. You write the turn body.
 - **[Hand-rolled loop](#hand-rolled-loop-with-primitives)** — you write the loop itself with `chat.messages`, `MessageAccumulator`, `pipeAndCapture`, and `writeTurnComplete`. The right choice when you need complete control over `.toUIMessageStream()` (e.g. `onFinish`, `originalMessages`) beyond what `chat.setUIMessageStreamOptions()` provides, or you're implementing a custom protocol.
 
+### Validating client data
+
+Use `chat.withClientData({ schema })` to validate `payload.metadata`. Custom agents parse the initial payload and later message and action frames before passing them to `run`, `chat.messages`, or `chat.createSession`. Schema defaults and transforms are included in the value your code receives. Close frames are not validated.
+
+If validation fails for a submitted turn or an async read such as `wait()`, the SDK writes an error chunk followed by `turn-complete` and skips the invalid frame. The invalid value never reaches your run handler or turn loop. The task then waits for the next valid frame. A messageless preload or continuation boot has no submitted turn to complete, so the SDK reports the error through the task log and `onClientDataValidationError` while it waits.
+
+An invalid [head-start handover](/ai-chat/fast-starts#handover-with-custom-agents) boot fails closed. The SDK waits for the warm handler to finish so stream ordering stays intact. A handover skip ends the run. A real handover writes the validation error and `turn-complete` after the warm output, then ends the run. Without a schema, metadata is passed through unchanged.
+
+`chat.messages.on()` is different because a subscribed frame can arrive while the current response is still streaming. Ending the turn at that point would cut off the response. While the subscription is active, the SDK skips the frame, logs the validation error, and calls `onClientDataValidationError` if you set it. This also applies to the steering subscription created by `chat.createSession({ pendingMessages })`. Calling `off()` prevents queued validation from invoking your handler or error callback.
+
+```ts
+import { chat } from "@trigger.dev/sdk/ai";
+import { z } from "zod";
+
+export const myChat = chat
+  .withClientData({ schema: z.object({ userId: z.string() }) })
+  .customAgent({
+    id: "my-chat",
+    onClientDataValidationError: ({ error, payload }) => {
+      console.warn("Invalid client data", { error, trigger: payload.trigger });
+    },
+    run: async (payload) => {
+      // ...
+    },
+  });
+```
+
+`chat.messages.peek()` validates synchronously and throws validation errors to the caller. If your schema only supports asynchronous parsing, use `once()`, `wait()`, or `waitWithIdleTimeout()` instead.
+
 ## Managed loop: chat.createSession()
 
 `chat.createSession()` gives you an async iterator of `ChatTurn` objects. Each turn arrives with the accumulated history, a combined stop+cancel signal, and helpers to finish the turn:
 
 ```ts trigger/my-chat.ts
-import { chat, type ChatTaskWirePayload } from "@trigger.dev/sdk/ai";
+import { chat } from "@trigger.dev/sdk/ai";
 import { streamText, stepCountIs } from "ai";
 import { anthropic } from "@ai-sdk/anthropic";
-
-export const myChat = chat.customAgent({
-  id: "my-chat",
-  run: async (payload: ChatTaskWirePayload, { signal }) => {
-    // One-time initialization — plain code, no hooks. Upsert, not create:
-    // continuation runs boot with the row already in place.
-    const clientData = payload.metadata as { userId: string };
-    await db.chat.upsert({
-      where: { id: payload.chatId },
-      create: { id: payload.chatId, userId: clientData.userId },
-      update: {},
-    });
-
-    const session = chat.createSession(payload, {
-      signal,
-      idleTimeoutInSeconds: 60,
-      timeout: "1h",
-    });
-
-    for await (const turn of session) {
-      // Persist the incoming user message BEFORE streaming — this is your
-      // onTurnStart equivalent. Without it, a page reload mid-stream
-      // restores the assistant text (replayed from the session) but loses
-      // the user message that prompted it.
-      await db.chat.update({
-        where: { id: turn.chatId },
-        data: { messages: turn.uiMessages },
+import { z } from "zod";
+
+export const myChat = chat
+  .withClientData({ schema: z.object({ userId: z.string() }) })
+  .customAgent({
+    id: "my-chat",
+    run: async (payload, { signal }) => {
+      // One-time initialization — plain code, no hooks. Upsert, not create:
+      // continuation runs boot with the row already in place.
+      const clientData = payload.metadata!;
+      await db.chat.upsert({
+        where: { id: payload.chatId },
+        create: { id: payload.chatId, userId: clientData.userId },
+        update: {},
       });
 
-      const result = streamText({
-        model: anthropic("claude-sonnet-4-5"),
-        messages: turn.messages,
-        abortSignal: turn.signal,
-        stopWhen: stepCountIs(15),
+      const session = chat.createSession(payload, {
+        signal,
+        idleTimeoutInSeconds: 60,
+        timeout: "1h",
       });
 
-      // Pipe, capture, accumulate, and signal turn-complete — all in one call
-      await turn.complete(result);
-
-      // Persist the full exchange after the turn — your onTurnComplete equivalent
-      await db.chat.update({
-        where: { id: turn.chatId },
-        data: { messages: turn.uiMessages },
-      });
-    }
-  },
-});
+      for await (const turn of session) {
+        // Persist the incoming user message BEFORE streaming — this is your
+        // onTurnStart equivalent. Without it, a page reload mid-stream
+        // restores the assistant text (replayed from the session) but loses
+        // the user message that prompted it.
+        await db.chat.update({
+          where: { id: turn.chatId },
+          data: { messages: turn.uiMessages },
+        });
+
+        const result = streamText({
+          model: anthropic("claude-sonnet-4-5"),
+          messages: turn.messages,
+          abortSignal: turn.signal,
+          stopWhen: stepCountIs(15),
+        });
+
+        // Pipe, capture, accumulate, and signal turn-complete — all in one call
+        await turn.complete(result);
+
+        // Persist the full exchange after the turn — your onTurnComplete equivalent
+        await db.chat.update({
+          where: { id: turn.chatId },
+          data: { messages: turn.uiMessages },
+        });
+      }
+    },
+  });
 ```
 
 <Warning>
@@ -102,7 +134,7 @@ Each turn yielded by the iterator provides:
 | `number`            | `number`                          | Turn number (0-indexed)                                  |
 | `chatId`            | `string`                          | Chat session ID                                          |
 | `trigger`           | `string`                          | What triggered this turn                                 |
-| `clientData`        | `unknown`                         | Client data from the transport                           |
+| `clientData`        | Schema output or `unknown`        | Parsed client data when `withClientData` is configured   |
 | `messages`          | `ModelMessage[]`                  | Full accumulated model messages — pass to `streamText`   |
 | `uiMessages`        | `UIMessage[]`                     | Full accumulated UI messages — use for persistence       |
 | `signal`            | `AbortSignal`                     | Combined stop+cancel signal (fresh each turn)            |

docs/ai-chat/reference.mdx

@@ -546,7 +546,7 @@ Use this when you need [`InferChatUIMessage`](#inferchatuimessage) / typed `data
 
 ## `chat.withClientData`
 
-Returns a [`ChatBuilder`](/ai-chat/types#chatbuilder) with a fixed client data schema. All hooks and `run` get typed `clientData` without passing `clientDataSchema` in `.agent()` options.
+Returns a [`ChatBuilder`](/ai-chat/types#chatbuilder) with a fixed client data schema. Managed-agent hooks and `run` get typed `clientData` without passing `clientDataSchema` in `.agent()` options. Custom agents parse `payload.metadata` on the initial payload and later input frames before passing it to user code.
 
 ```ts
 chat.withClientData<TSchema>({ schema: TSchema }): ChatBuilder<UIMessage, TSchema>;
@@ -556,6 +556,8 @@ chat.withClientData<TSchema>({ schema: TSchema }): ChatBuilder<UIMessage, TSchem
 | --------- | ------------ | -------------------------------------------------- |
 | `schema`  | `TaskSchema` | Zod, ArkType, Valibot, or any supported schema lib |
 
+For `chat.customAgent()`, invalid client data is skipped. Async reads emit an error chunk followed by `turn-complete`. A `chat.messages.on()` subscription uses the task's `onClientDataValidationError` callback and task log instead, so an active response is not ended early. Without a schema, metadata is passed through unchanged.
+
 Full guide: [Typed client data](/ai-chat/types#typed-client-data-with-chatwithclientdata).
 
 ## `ChatWithUIMessageConfig`

docs/ai-chat/types.mdx

@@ -140,7 +140,7 @@ You can also import `InferChatUIMessage` from `@trigger.dev/sdk/ai` in non-React
 
 ## Typed client data with `chat.withClientData`
 
-`chat.withClientData({ schema })` returns a [ChatBuilder](#chatbuilder) that fixes the client data schema. All hooks and `run` receive typed `clientData` without needing `clientDataSchema` in `.agent()` options.
+`chat.withClientData({ schema })` returns a [ChatBuilder](#chatbuilder) that fixes the client data schema. Managed-agent hooks and `run` receive typed `clientData` without needing `clientDataSchema` in `.agent()` options. A `.customAgent()` run receives the parsed schema output in `payload.metadata`, and `chat.createSession()` yields it as `turn.clientData`.
 
 ```ts
 import { chat } from "@trigger.dev/sdk/ai";
@@ -167,6 +167,8 @@ export const myChat = chat
   });
 ```
 
+The schema runs at runtime for both `.agent()` and `.customAgent()`. Custom agents validate the initial payload and later `chat.messages` frames. Invalid frames are not passed to user code. Async reads emit an error chunk followed by `turn-complete`; `chat.messages.on()` reports through `onClientDataValidationError` and the task log so it does not end an active response. Without a schema, metadata is passed through unchanged.
+
 ## ChatBuilder
 
 Both `chat.withUIMessage()` and `chat.withClientData()` return a **ChatBuilder** — a chainable object that accumulates configuration before creating the agent with `.agent()`.