fix(sdk,core): address review feedback on chat.agent hardening

Fold the failed turn's wire message into the error-path snapshot and
onTurnComplete event so an early pre-run throw cannot strand it, and stop
passing raw metadata as the parsed clientData on errored turns. Mark the
session streaming before subscribing in sendAction so a reload mid-action
resumes. Keep the per-append X-Part-Id from being overridden by a
transport-wide header, and align the server-side append part id entropy
with the browser transport. Adds tests for the error-turn snapshot,
sendAction streaming state, and the X-Part-Id header precedence.

Author: ericallam

packages/core/src/v3/apiClient/index.ts

@@ -1279,8 +1279,9 @@ export class ApiClient {
   ) {
     // Generated once per logical append, outside zodfetch, so its internal
     // retries reuse the same part id and the server-side dedupe collapses a
-    // retried POST whose first attempt actually committed.
-    const partId = nanoid(7);
+    // retried POST whose first attempt actually committed. Full-length nanoid
+    // (~126 bits) to match the browser transport's randomUUID entropy.
+    const partId = nanoid();
     return zodfetch(
       AppendToStreamResponseBody,
       `${this.baseUrl}/realtime/v1/sessions/${encodeURIComponent(sessionIdOrExternalId)}/${io}/append`,

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

@@ -7583,6 +7583,17 @@ function chatAgent<
                 // Best-effort — if stream write fails, let the run continue anyway
               }
 
+              // The submit-message merge into the accumulator may not have run
+              // yet (a pre-run hook threw), so fold the wire message in for the
+              // error event + snapshot — the cursor has already advanced past it,
+              // so otherwise it survives in neither the snapshot nor the `.in` tail.
+              const erroredWireMessage = (currentWirePayload as { message?: TUIMessage }).message;
+              const erroredUIMessages =
+                erroredWireMessage &&
+                !accumulatedUIMessages.some((m) => m.id === erroredWireMessage.id)
+                  ? [...accumulatedUIMessages, erroredWireMessage]
+                  : accumulatedUIMessages;
+
               // Fire onTurnComplete on the error path too — the docs promise it
               // runs "after every turn, successful or errored" so customers can
               // mark the turn failed. `responseMessage` is undefined/partial and
@@ -7596,15 +7607,18 @@ function chatAgent<
                         ctx,
                         chatId: currentWirePayload.chatId,
                         messages: accumulatedMessages,
-                        uiMessages: accumulatedUIMessages,
+                        uiMessages: erroredUIMessages,
                         newMessages: [],
-                        newUIMessages: [],
+                        newUIMessages: erroredWireMessage ? [erroredWireMessage] : [],
                         responseMessage: undefined,
                         rawResponseMessage: undefined,
                         turn,
                         runId: ctx.run.id,
                         chatAccessToken: "",
-                        clientData: currentWirePayload.metadata as inferSchemaIn<TClientDataSchema>,
+                        // Parsed `clientData` isn't reliably in scope here (parsing
+                        // may itself be the failure), and the raw metadata is the
+                        // wrong shape — leave it undefined on the error path.
+                        clientData: undefined,
                         stopped: false,
                         continuation,
                         previousRunId,
@@ -7642,7 +7656,7 @@ function chatAgent<
                   await writeChatSnapshot<TUIMessage>(sessionIdForSnapshot, {
                     version: 1,
                     savedAt: Date.now(),
-                    messages: accumulatedUIMessages,
+                    messages: erroredUIMessages,
                     lastOutEventId: errorTurnCompleteResult?.lastEventId,
                   });
                 } catch (error) {

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

@@ -1005,6 +1005,65 @@ describe("TriggerChatTransport", () => {
       expect(actionBody.payload.trigger).toBe("action");
       expect(actionBody.payload.action).toEqual({ type: "undo" });
     });
+
+    it("marks the session streaming and notifies before subscribing", async () => {
+      global.fetch = vi.fn().mockImplementation(async (url: string | URL) => {
+        const urlStr = typeof url === "string" ? url : url.toString();
+        if (isSessionStreamAppendUrl(urlStr)) return defaultAppendResponse();
+        if (isSessionOutSubscribeUrl(urlStr)) return defaultSseResponse();
+        throw new Error(`Unexpected URL: ${urlStr}`);
+      });
+
+      const onSessionChange = vi.fn();
+      const transport = new TriggerChatTransport({
+        task: "my-chat-task",
+        accessToken: () => "pat",
+        onSessionChange,
+        sessions: { "chat-act-stream": { publicAccessToken: "p" } },
+      });
+
+      const stream = await transport.sendAction("chat-act-stream", { type: "undo" });
+      // isStreaming:true must be observed during the action — otherwise a reload
+      // mid-action sees a persisted isStreaming:false and never resumes.
+      expect(
+        onSessionChange.mock.calls.some(([, session]) => session && session.isStreaming === true)
+      ).toBe(true);
+      await drainChunks(stream);
+    });
+  });
+
+  describe("append idempotency header", () => {
+    it("the per-append X-Part-Id wins over a transport-wide headers override", async () => {
+      let appendPartId: string | undefined;
+      global.fetch = vi.fn().mockImplementation(async (url: string | URL, init?: RequestInit) => {
+        const urlStr = typeof url === "string" ? url : url.toString();
+        if (isSessionStreamAppendUrl(urlStr)) {
+          appendPartId = (init!.headers as Record<string, string>)["X-Part-Id"];
+          return defaultAppendResponse();
+        }
+        if (isSessionOutSubscribeUrl(urlStr)) return defaultSseResponse();
+        throw new Error(`Unexpected URL: ${urlStr}`);
+      });
+
+      const transport = new TriggerChatTransport({
+        task: "my-chat-task",
+        accessToken: () => "pat",
+        headers: { "X-Part-Id": "STATIC-OVERRIDE" },
+        sessions: { "chat-hdr": { publicAccessToken: "p" } },
+      });
+
+      const stream = await transport.sendMessages({
+        trigger: "submit-message",
+        chatId: "chat-hdr",
+        messageId: undefined,
+        messages: [createUserMessage("hi")],
+        abortSignal: undefined,
+      });
+      await drainChunks(stream);
+
+      expect(appendPartId).toBeDefined();
+      expect(appendPartId).not.toBe("STATIC-OVERRIDE");
+    });
   });
 
   describe("reconnectToStream", () => {

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

@@ -944,6 +944,11 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
       this.activeStreams.delete(chatId);
     }
 
+    // Mark streaming + persist so a reload mid-action resumes (reconnectToStream
+    // no-ops when the persisted session says isStreaming: false).
+    state.isStreaming = true;
+    this.notifySessionChange(chatId, state);
+
     return this.subscribeToSessionStream(state, undefined, chatId);
   };
 
@@ -1123,14 +1128,14 @@ export class TriggerChatTransport implements ChatTransport<UIMessage> {
   ): Promise<void> {
     const ctx: ChatTransportEndpointContext = { endpoint: "in", chatId };
     const url = `${this.resolveBaseURL(ctx)}/realtime/v1/sessions/${encodeURIComponent(chatId)}/in/append`;
+    // extraHeaders first so the fixed headers below win — a transport-wide
+    // X-Part-Id must not override the per-append idempotency key.
     const headers: Record<string, string> = {
+      ...this.extraHeaders,
       "Content-Type": "application/json",
       Authorization: `Bearer ${token}`,
       "x-trigger-source": "sdk",
-      // Idempotency key: the server skips appends whose part id it has
-      // already committed, so a retried POST can't duplicate the record.
       "X-Part-Id": partId ?? crypto.randomUUID(),
-      ...this.extraHeaders,
     };
     const response = await this.doFetch(ctx, url, { method: "POST", headers, body });
     if (!response.ok) {

packages/trigger-sdk/test/mockChatAgent.test.ts

@@ -1140,6 +1140,63 @@ describe("mockChatAgent", () => {
     }
   });
 
+  it("error turn: onTurnComplete fires with the error and the failed message is snapshotted", async () => {
+    const onTurnComplete = vi.fn();
+    const agent = chat.agent({
+      id: "mockChatAgent.error-turn-run",
+      onTurnComplete,
+      run: async () => {
+        throw new Error("boom in run");
+      },
+    });
+    const harness = mockChatAgent(agent, { chatId: "err-run" });
+    try {
+      await harness.sendMessage(userMessage("hello", "u-err-run"));
+      await new Promise((r) => setTimeout(r, 50));
+      expect(onTurnComplete).toHaveBeenCalledTimes(1);
+      const evt = onTurnComplete.mock.calls[0]![0];
+      expect(evt.error).toBeInstanceOf(Error);
+      expect(evt.finishReason).toBe("error");
+      expect(evt.responseMessage).toBeUndefined();
+      expect(evt.uiMessages.some((m: any) => m.id === "u-err-run")).toBe(true);
+      const snap = harness.getSnapshot();
+      expect(snap?.messages.some((m) => m.id === "u-err-run")).toBe(true);
+    } finally {
+      await harness.close();
+    }
+  });
+
+  it("error turn: a pre-merge hook throw still snapshots the failed user message", async () => {
+    const model = new MockLanguageModelV3({
+      doStream: async () => ({ stream: textStream("never reached") }),
+    });
+    const onTurnComplete = vi.fn();
+    const agent = chat.agent({
+      id: "mockChatAgent.error-turn-prehook",
+      // onValidateMessages fires BEFORE the wire message is merged into the
+      // accumulator, so the message lands in the snapshot only because the
+      // error path folds the wire message back in.
+      onValidateMessages: async () => {
+        throw new Error("boom in validate");
+      },
+      onTurnComplete,
+      run: async ({ messages, signal }) => streamText({ model, messages, abortSignal: signal }),
+    });
+    const harness = mockChatAgent(agent, { chatId: "err-prehook" });
+    try {
+      await harness.sendMessage(userMessage("validate me", "u-err-prehook"));
+      await new Promise((r) => setTimeout(r, 50));
+      expect(onTurnComplete).toHaveBeenCalledTimes(1);
+      const evt = onTurnComplete.mock.calls[0]![0];
+      expect(evt.error).toBeInstanceOf(Error);
+      expect(evt.uiMessages.some((m: any) => m.id === "u-err-prehook")).toBe(true);
+      const snap = harness.getSnapshot();
+      expect(snap?.messages.some((m) => m.id === "u-err-prehook")).toBe(true);
+    } finally {
+      await harness.close();
+    }
+  });
+
   it("seeds locals before run() via setupLocals (DI pattern)", async () => {
     type FakeDb = { findUser(id: string): Promise<{ id: string; name: string }> };
     const dbKey = locals.create<FakeDb>("test-db");