fix(chat): narrow custom agent validation contract

Author: gtremper

docs/ai-chat/client-protocol.mdx

@@ -832,7 +832,9 @@ Custom actions (undo, rollback, edit) ride on the same `.in` channel using `kind
 }
 ```
 
-Actions wake the agent from suspension (same as messages) and fire the `onAction` hook — they are not turns, so `run()` and turn lifecycle hooks do not fire. If `onAction` returns a `StreamTextResult`, the response is auto-piped to the frontend (but still no `run()` or `onTurnComplete`). The `action` payload is validated against the agent's `actionSchema`. If the agent didn't register an `actionSchema` (or your `action` payload doesn't match it), validation fails the same way `metadata` does — `.in/append` returns `200 OK`, but the run trace shows `chat turn N [ERROR]` and the wire emits a `turn-complete` control record with no other chunks. See [Actions](/ai-chat/actions) for the agent-side schema setup.
+For managed `chat.agent()` tasks, actions wake the agent from suspension (same as messages) and fire the `onAction` hook — they are not turns, so `run()` and turn lifecycle hooks do not fire. If `onAction` returns a `StreamTextResult`, the response is auto-piped to the frontend (but still no `run()` or `onTurnComplete`). The `action` payload is validated against the agent's `actionSchema`. If the agent didn't register an `actionSchema` (or your `action` payload doesn't match it), validation fails the same way `metadata` does — `.in/append` returns `200 OK`, but the run trace shows `chat turn N [ERROR]` and the wire emits a `turn-complete` control record with no other chunks. See [Actions](/ai-chat/actions) for the agent-side schema setup.
+
+Raw `chat.customAgent()` tasks receive `action` as `unknown` and must validate it in their own loop.
 
 ### Regenerating the last response
 

docs/ai-chat/custom-agents.mdx

@@ -21,13 +21,19 @@ Inside the wrapper, pick one of two loop styles:
 
 ### 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.
+Use `chat.withClientData({ schema })` to validate `payload.metadata`. Custom agents parse the metadata on the initial payload and every later non-close input frame before passing it to `run`, `chat.messages`, or `chat.createSession`. Schema defaults and transforms are included in the value your code receives.
 
-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.
+This only validates `metadata`. A raw custom agent does not expose an action schema, so `payload.action` remains `unknown`. Validate the full frame or action payload in your own loop when you need that boundary.
+
+If validation fails for a submitted turn or an async read such as `wait()`, the SDK consumes and skips the invalid frame, writes an `Invalid client data` error followed by `turn-complete`, then waits for the next valid frame. The invalid value is not returned to the raw caller. The detailed validator error is available in the task log and `onClientDataValidationError`, but it is not sent to the client.
+
+This convenience path settles the invalid input before the read returns. If your raw loop needs to coordinate validation with persistence or settlement, omit `withClientData({ schema })` and validate the full wire frame in the loop instead. A messageless preload or continuation boot has no submitted turn to complete, so the SDK reports the error through the task log and callback 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.
+`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 an invalid 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()` stops the subscription from accepting new frames. A valid frame accepted before `off()` still finishes validation and is delivered to the handler. An invalid frame that finishes validation after `off()` is logged without calling the handler or error callback.
 
 ```ts
 import { chat } from "@trigger.dev/sdk/ai";

docs/ai-chat/reference.mdx

@@ -786,7 +786,7 @@ Send a custom action to the agent. Actions wake the agent from suspension and fi
 transport.sendAction(chatId: string, action: unknown): Promise<ReadableStream<UIMessageChunk>>
 ```
 
-The action payload is validated against the agent's `actionSchema` on the backend.
+For managed `chat.agent()` tasks, the action payload is validated against the agent's `actionSchema` on the backend. Raw `chat.customAgent()` tasks receive it as `unknown` and must validate it themselves.
 
 ```tsx
 // Undo button

packages/trigger-sdk/src/v3/ai.ts

@@ -1441,6 +1441,8 @@ type ChatCustomAgentClientDataErrorHandler = (event: {
   payload: ChatTaskWirePayload;
 }) => Promise<void> | void;
 
+const CHAT_CUSTOM_AGENT_CLIENT_DATA_ERROR_TEXT = "Invalid client data";
+
 const chatCustomAgentClientDataParserKey = locals.create<ChatCustomAgentClientDataParser>(
   "chat.customAgentClientDataParser"
 );
@@ -1503,13 +1505,14 @@ function getChatCustomAgentSyncSchemaParseFn(schema: TaskSchema): (value: unknow
 }
 
 async function writeChatCustomAgentClientDataErrorToStream(
-  payload: ChatTaskWirePayload,
-  error: unknown
+  payload: ChatTaskWirePayload
 ): Promise<void> {
-  const errorText = error instanceof Error ? error.message : "An unexpected error occurred";
   try {
     await withChatWriter((writer) => {
-      writer.write({ type: "error", errorText } as any);
+      writer.write({
+        type: "error",
+        errorText: CHAT_CUSTOM_AGENT_CLIENT_DATA_ERROR_TEXT,
+      } as any);
     });
     await chatWriteTurnComplete();
   } catch (signalError) {
@@ -1552,7 +1555,7 @@ async function reportChatCustomAgentClientDataError(
   if (!options.writeToStream) {
     return;
   }
-  await writeChatCustomAgentClientDataErrorToStream(payload, error);
+  await writeChatCustomAgentClientDataErrorToStream(payload);
 }
 
 type ChatCustomAgentPayloadValidationResult<TPayload extends ChatTaskWirePayload> =
@@ -1796,7 +1799,8 @@ const messagesInput: RealtimeDefinedInputStream<ChatTaskWirePayload> = {
       return subscribeToRawChatMessages(handler);
     }
 
-    return subscribeToValidatedChatMessages((payload) => handler(payload));
+    const deliver = (payload: ChatTaskWirePayload) => handler(payload);
+    return subscribeToValidatedChatMessages(deliver, { onAfterOff: deliver });
   },
   once(options) {
     const ctx = taskContext.ctx;
@@ -5594,6 +5598,7 @@ type ChatCustomAgentOptions<
    * error chunk followed by `turn-complete`. Messageless boots and active
    * subscriptions use `onClientDataValidationError` and the task log because
    * there is no submitted turn to complete or a response may still be streaming.
+   * This validates `metadata` only; raw `action` payloads remain `unknown`.
    */
   clientDataSchema?: TClientDataSchema;
   /**
@@ -5719,7 +5724,7 @@ function chatCustomAgent<
 
         // The head-start writer flushes before sending this signal. Writing
         // the terminal error now preserves stream order and closes the stitch.
-        await writeChatCustomAgentClientDataErrorToStream(payload, validated.error);
+        await writeChatCustomAgentClientDataErrorToStream(payload);
         return;
       }
 

packages/trigger-sdk/src/v3/chat.ts

@@ -95,7 +95,10 @@ export type ChatTaskWirePayload<TMessage extends UIMessage = UIMessage, TMetadat
     | "handover-prepare";
   messageId?: string;
   metadata?: TMetadata;
-  /** Custom action payload when `trigger` is `"action"`. Validated against `actionSchema` on the backend. */
+  /**
+   * Custom action payload when `trigger` is `"action"`. Managed agents validate
+   * this against `actionSchema`; raw custom agents must validate it themselves.
+   */
   action?: unknown;
   /** Whether this run is continuing an existing chat whose previous run ended. */
   continuation?: boolean;

packages/trigger-sdk/test/custom-agent-client-data-validation.test.ts

@@ -90,6 +90,7 @@ describe("chat.customAgent clientData validation", () => {
     };
     let started = false;
     const receivedClientData: unknown[] = [];
+    const validationErrors: unknown[] = [];
 
     const agent = chat
       .withClientData({
@@ -100,6 +101,9 @@ describe("chat.customAgent clientData validation", () => {
       })
       .customAgent({
         id: "custom-agent-client-data-invalid-frame",
+        onClientDataValidationError: ({ error }) => {
+          validationErrors.push(error);
+        },
         run: async (payload, { signal }) => {
           started = true;
           const session = chat.createSession(payload, {
@@ -126,8 +130,10 @@ describe("chat.customAgent clientData validation", () => {
 
       expect(receivedClientData).toHaveLength(0);
       expect(invalidTurn.chunks).toEqual([
-        expect.objectContaining({ type: "error", errorText: expect.any(String) }),
+        expect.objectContaining({ type: "error", errorText: "Invalid client data" }),
       ]);
+      expect(validationErrors).toHaveLength(1);
+      expect(validationErrors[0]).toBeInstanceOf(z.ZodError);
       expect(invalidTurn.rawChunks).toContainEqual(
         expect.objectContaining({ type: "trigger:turn-complete" })
       );
@@ -229,7 +235,7 @@ describe("chat.customAgent clientData validation", () => {
 
       expect(runCalls).toBe(0);
       expect(harness.allChunks).toEqual([
-        expect.objectContaining({ type: "error", errorText: expect.any(String) }),
+        expect.objectContaining({ type: "error", errorText: "Invalid client data" }),
       ]);
 
       clientData.userId = "user_123";
@@ -301,7 +307,7 @@ describe("chat.customAgent clientData validation", () => {
     }
   });
 
-  it("does not deliver frames or validation callbacks after chat.messages.on is removed", async () => {
+  it("does not report an invalid frame whose validation finishes after chat.messages.on is removed", async () => {
     const clientData = { blocked: false };
     const parserStarted = deferred();
     const releaseParser = deferred();
@@ -365,6 +371,68 @@ describe("chat.customAgent clientData validation", () => {
     }
   });
 
+  it("delivers a valid frame accepted before chat.messages.on is removed", async () => {
+    const clientData = { blocked: false };
+    const parserStarted = deferred();
+    const releaseParser = deferred();
+    const delivered = deferred();
+    let removeSubscription: (() => void) | undefined;
+    let receivedMetadata: unknown;
+    let handlerCalls = 0;
+    let started = false;
+
+    const agent = chat
+      .withClientData({
+        schema: async (value: unknown) => {
+          const blocked = (value as { blocked: boolean }).blocked;
+          if (blocked) {
+            parserStarted.resolve();
+            await releaseParser.promise;
+          }
+          return { blocked, parsed: true as const };
+        },
+      })
+      .customAgent({
+        id: "custom-agent-client-data-deliver-pending-after-off",
+        run: async (_payload, { signal }) => {
+          started = true;
+          const subscription = chat.messages.on(async (payload) => {
+            handlerCalls++;
+            receivedMetadata = payload.metadata;
+            await chat.writeTurnComplete();
+            delivered.resolve();
+          });
+          removeSubscription = () => subscription.off();
+          await new Promise<void>((resolve) => {
+            signal.addEventListener("abort", () => resolve(), { once: true });
+          });
+        },
+      });
+
+    const harness = mockChatAgent(agent, {
+      chatId: "custom-agent-client-data-deliver-pending-after-off-chat",
+      clientData,
+    });
+
+    try {
+      await waitFor(() => started);
+      clientData.blocked = true;
+      const send = harness.sendMessage(userMessage("hello", "message-1"));
+      await parserStarted.promise;
+
+      removeSubscription!();
+      releaseParser.resolve();
+
+      await send;
+      await delivered.promise;
+      expect(handlerCalls).toBe(1);
+      expect(receivedMetadata).toEqual({ blocked: true, parsed: true });
+    } finally {
+      releaseParser.resolve();
+      await harness.close();
+    }
+  });
+
   it("throws from chat.messages.peek when an object parser returns a promise", async () => {
     const clientData = { userId: "user_123" };
     let started = false;
@@ -464,7 +532,7 @@ describe("chat.customAgent clientData validation", () => {
 
       expect(receivedClientData).toEqual([{ attempt: 1 }]);
       expect(harness.allChunks).toContainEqual(
-        expect.objectContaining({ type: "error", errorText: expect.any(String) })
+        expect.objectContaining({ type: "error", errorText: "Invalid client data" })
       );
     } finally {
       releaseFirstTurn.resolve();
@@ -626,7 +694,7 @@ describe("chat.customAgent clientData validation", () => {
       expect(lateFrameParseCalls).toBe(1);
       expect(receivedSequences).toEqual([1]);
       expect(harness.allChunks).toContainEqual(
-        expect.objectContaining({ type: "error", errorText: "invalid late frame" })
+        expect.objectContaining({ type: "error", errorText: "Invalid client data" })
       );
     } finally {
       releaseFirstTurn.resolve();
@@ -754,7 +822,7 @@ describe("chat.customAgent clientData validation", () => {
 
       expect(runCalls).toBe(0);
       expect(handover.chunks).toEqual([
-        expect.objectContaining({ type: "error", errorText: expect.any(String) }),
+        expect.objectContaining({ type: "error", errorText: "Invalid client data" }),
       ]);
       expect(handover.rawChunks).toContainEqual(
         expect.objectContaining({ type: "trigger:turn-complete" })