fix(sdk): clean incomplete parts from the recovered error partial

Apply cleanupAbortedParts to the partial recovered on a source-stream failure,
in both the managed and manual loops. A transport error leaves the same
incomplete-part state as a user stop (streaming text not finalized, dangling
tool calls), so the partial is cleaned the same way: text is kept and
unfinished tool parts are dropped, keeping the persisted UI history and the
model view consistent. Also reset the error event's newMessages delta when the
model-message conversion fails, so it can't advertise messages that were rolled
back.

Author: matt-aitken

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

@@ -7936,6 +7936,9 @@ function chatAgent<
             let partialResponse: TUIMessage | undefined =
               capturedPartialResponse ??
               ((await assemblePartialFromChunks(turnBufferedChunks)) as TUIMessage | undefined);
+            if (partialResponse) {
+              partialResponse = cleanupAbortedParts(partialResponse);
+            }
 
             // Build the complete error UI state. A HITL/tool continuation partial
             // reuses an existing assistant id, so replace it in place (like the
@@ -7999,6 +8002,7 @@ function chatAgent<
                 locals.set(chatCurrentUIMessagesKey, accumulatedUIMessages);
               } catch {
                 // Keep the prior model accumulator if conversion fails.
+                erroredNewModelMessages = [];
               }
             }
 
@@ -9840,17 +9844,8 @@ function createChatSession(
                 // Surface a genuine stream failure to the caller. A user stop
                 // (status "aborted") falls through so the partial is accumulated.
                 if (captured.status === "error") {
-                  // Preserve the partial the model streamed before the failure:
-                  // accumulate it (mirroring the stop path) so `turn.uiMessages`
-                  // reflects it and the caller can persist it after catching,
-                  // then rethrow. Without this the partial pipeAndCapture
-                  // reconstructed is silently dropped on rethrow. Fold in any
-                  // data parts queued via chat.response / writer.write() this
-                  // turn, same as the success path below. cleanupAbortedParts is
-                  // intentionally skipped: it only runs on a user stop, not on a
-                  // hard transport error (which is not an abort).
                   if (captured.message) {
-                    const partial = captured.message;
+                    const partial = cleanupAbortedParts(captured.message);
                     const queuedParts = locals.get(chatResponsePartsKey);
                     if (queuedParts && queuedParts.length > 0) {
                       (partial as any).parts = [...(partial.parts ?? []), ...queuedParts];

packages/trigger-sdk/test/chat-agent-source-stream-error.test.ts

@@ -48,13 +48,17 @@ function erroringSource(errorMessage: string) {
     { type: "text-start", id: "t1" },
     { type: "text-delta", id: "t1", delta: "partial answer" },
   ];
+  return sourceFromChunks(partialChunks, errorMessage);
+}
+
+function sourceFromChunks(chunks: unknown[], errorMessage: string) {
   return {
     toUIMessageStream() {
       let i = 0;
       return new ReadableStream({
         pull(controller) {
-          if (i < partialChunks.length) {
-            controller.enqueue(partialChunks[i++]);
+          if (i < chunks.length) {
+            controller.enqueue(chunks[i++]);
           } else {
             controller.error(new Error(errorMessage));
           }
@@ -319,6 +323,48 @@ describe("chat.agent managed loop — source-stream failure", () => {
       await harness.close();
     }
   });
+
+  it("cleans dangling tool parts from the recovered partial while keeping its text", async () => {
+    const turnCompletes: TurnCompleteEvent<unknown, UIMessage>[] = [];
+
+    const agent = chat.agent({
+      id: "chatAgent.error-partial-cleanup",
+      run: async () =>
+        sourceFromChunks(
+          [
+            { type: "start", messageId: "a-tool" },
+            { type: "text-start", id: "t1" },
+            { type: "text-delta", id: "t1", delta: "thinking" },
+            { type: "text-end", id: "t1" },
+            { type: "tool-input-start", toolCallId: "tc1", toolName: "search" },
+            {
+              type: "tool-input-available",
+              toolCallId: "tc1",
+              toolName: "search",
+              input: { q: "x" },
+            },
+          ],
+          "UND_ERR_BODY_TIMEOUT"
+        ) as never,
+      onTurnComplete: async (event) => {
+        turnCompletes.push(event);
+      },
+    });
+
+    const harness = mockChatAgent(agent, { chatId: "cae-partial-cleanup" });
+    try {
+      await harness.sendMessage(userMessage("hi", "u-1"));
+      await waitFor(() => turnCompletes.length >= 1);
+
+      const evt = turnCompletes[0]!;
+      expect(evt.responseMessage).toBeDefined();
+      const parts = evt.responseMessage!.parts as Array<{ type: string }>;
+      expect(extractText(evt.responseMessage)).toBe("thinking");
+      expect(parts.some((p) => p.type.startsWith("tool-"))).toBe(false);
+    } finally {
+      await harness.close();
+    }
+  });
 });
 
 describe("chat.createSession turn.complete() — source-stream failure", () => {