fix(sdk): channel stream editor tail, abort cleanup, and error egress

Three fixes to chat.agent channel egress:

The debounced stream editor now re-arms after an edit it skipped because a
previous edit was still in flight, so text buffered during that window still
reaches the channel instead of stalling until the next delta.

The stream editor is stopped when the reply stream is aborted or cancelled,
not just on normal completion, so a late timer can no longer edit the channel
message after the turn has ended.

A turn that throws now edits the placeholder to show the error, so a channel
user sees the failure instead of a message stuck on the loading placeholder.

Author: ericallam

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

@@ -5057,13 +5057,23 @@ function makeChannelStreamEditor<TEvent>(
 ) {
   const outbound = connector.outbound ?? defaultChannelOutbound;
   let latest = "";
+  let lastSent = "";
   let timer: ReturnType<typeof setTimeout> | undefined;
   let inFlight = false;
   let stopped = false;
 
+  const arm = () => {
+    if (stopped || timer) return;
+    timer = setTimeout(() => {
+      timer = undefined;
+      void edit();
+    }, CHANNEL_STREAM_EDIT_INTERVAL_MS);
+  };
+
   const edit = async () => {
     if (stopped || inFlight) return;
     const text = latest;
+    if (text === lastSent) return;
     const message = outbound({
       text,
       message: { id: ackRef, role: "assistant", parts: [{ type: "text", text }] } as UIMessage,
@@ -5080,10 +5090,12 @@ function makeChannelStreamEditor<TEvent>(
         mode: "stream",
         final: false,
       });
+      lastSent = text;
     } catch (error) {
       logger.warn("chat.agent: channel stream edit failed", { error });
     } finally {
       inFlight = false;
+      if (!stopped && latest !== lastSent) arm();
     }
   };
 
@@ -5093,11 +5105,7 @@ function makeChannelStreamEditor<TEvent>(
       const c = chunk as { type?: string; delta?: unknown };
       if (c?.type === "text-delta" && typeof c.delta === "string") {
         latest += c.delta;
-        if (!timer)
-          timer = setTimeout(() => {
-            timer = undefined;
-            void edit();
-          }, CHANNEL_STREAM_EDIT_INTERVAL_MS);
+        arm();
       }
     },
     stop() {
@@ -5110,6 +5118,39 @@ function makeChannelStreamEditor<TEvent>(
   };
 }
 
+type ChannelStreamEditor = { observe(chunk: unknown): void; stop(): void };
+
+/**
+ * Wrap the reply stream so it debounce-edits the channel message as it streams.
+ * Both `flush` (the stream completed) and `cancel` (the stream was aborted or
+ * cancelled downstream) stop the editor, so a pending debounce timer can never
+ * fire an edit after the turn ended, onto a message the next turn already owns.
+ */
+function makeChannelStreamTap(editor: ChannelStreamEditor): TransformStream<unknown, unknown> {
+  const transformer: {
+    transform(chunk: unknown, controller: TransformStreamDefaultController<unknown>): void;
+    flush(): void;
+    cancel?: (reason?: unknown) => void;
+  } = {
+    transform(chunk, controller) {
+      editor.observe(chunk);
+      controller.enqueue(chunk);
+    },
+    flush() {
+      editor.stop();
+    },
+    cancel() {
+      editor.stop();
+    },
+  };
+  return new TransformStream<unknown, unknown>(transformer);
+}
+
+export {
+  makeChannelStreamEditor as __makeChannelStreamEditorForTests,
+  makeChannelStreamTap as __makeChannelStreamTapForTests,
+};
+
 // The `action` type onAction receives: the actionSchema output (when set) unioned with the action
 // envelopes of any listed chat.event descriptors. ChatEventActions<[]> is `never`, so it collapses
 // cleanly when no events are listed; `unknown` is kept only when neither source is present.
@@ -6925,6 +6966,8 @@ function chatAgent<
           let channelConn: AnyChannelConnector | undefined;
           let channelWorkingReaction: string | undefined;
           let channelWireEvent: { event: unknown; deliveryId: string } | undefined;
+          let channelAckRef: string | undefined;
+          let channelStreamEditor: ChannelStreamEditor | undefined;
           let droppedStaleInteraction = false;
           try {
             // Extract turn-level context before entering the span. Slim
@@ -6941,7 +6984,6 @@ function chatAgent<
             void _hsm;
             channelWireEvent = wireChannelEvent;
             let effectiveIncomingMessage = incomingMessage;
-            let channelAckRef: string | undefined;
             if (wireChannelEvent) {
               channelConn = channels?.find((c) => c.id === wireChannelEvent.connectorId);
               if (channelConn) {
@@ -7799,21 +7841,13 @@ function chatAgent<
                         channelAckRef &&
                         uiStream instanceof ReadableStream
                       ) {
-                        const editor = makeChannelStreamEditor(
+                        channelStreamEditor = makeChannelStreamEditor(
                           channelConn,
                           wireChannelEvent,
                           channelAckRef
                         );
                         streamForPipe = uiStream.pipeThrough(
-                          new TransformStream({
-                            transform(chunk, controller) {
-                              editor.observe(chunk);
-                              controller.enqueue(chunk);
-                            },
-                            flush() {
-                              editor.stop();
-                            },
-                          })
+                          makeChannelStreamTap(channelStreamEditor)
                         );
                       }
                       await pipeChat(tapUIMessageChunks(streamForPipe, turnBufferedChunks), {
@@ -7833,6 +7867,7 @@ function chatAgent<
                     }
                   } finally {
                     msgSub.off();
+                    channelStreamEditor?.stop();
                   }
 
                   // Wait for onFinish to fire — on abort this may resolve slightly
@@ -8593,6 +8628,37 @@ function chatAgent<
               }
             }
 
+            if (channelWireEvent && channelConn?.send && channelAckRef) {
+              const channelErrorText =
+                turnError instanceof Error ? turnError.message : "An unexpected error occurred";
+              const errorMessage = (channelConn.outbound ?? defaultChannelOutbound)({
+                text: channelErrorText,
+                message: {
+                  id: channelAckRef,
+                  role: "assistant",
+                  parts: [{ type: "text", text: channelErrorText }],
+                } as UIMessage,
+                final: true,
+                stopped: false,
+                error: turnError,
+              });
+              if (errorMessage) {
+                try {
+                  await channelConn.send(errorMessage, {
+                    event: channelWireEvent.event,
+                    deliveryId: channelWireEvent.deliveryId,
+                    previousRef: channelAckRef,
+                    mode: channelConn.delivery,
+                    final: true,
+                  });
+                } catch (egressError) {
+                  logger.warn("chat.agent: channel error egress send failed", {
+                    error: egressError,
+                  });
+                }
+              }
+            }
+
             let errorTurnCompleteResult:
               | Awaited<ReturnType<typeof writeTurnCompleteChunk>>
               | undefined;

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

@@ -1,7 +1,11 @@
 import { mockChatAgent, recordingChannelConnector } from "../src/v3/test/index.js";
 
-import { describe, expect, it } from "vitest";
-import { chat } from "../src/v3/ai.js";
+import { describe, expect, it, vi } from "vitest";
+import {
+  chat,
+  __makeChannelStreamEditorForTests,
+  __makeChannelStreamTapForTests,
+} from "../src/v3/ai.js";
 import { simulateReadableStream, streamText } from "ai";
 import { MockLanguageModelV3 } from "ai/test";
 import type { LanguageModelV3StreamPart } from "@ai-sdk/provider";
@@ -147,4 +151,127 @@ describe("chat.agent channels", () => {
       await harness.close();
     }
   });
+
+  it("edits the placeholder to the error when a channel turn throws", async () => {
+    const channel = recordingChannelConnector();
+
+    const agent = chat.agent({
+      id: "chatChannels.error-egress",
+      channels: [channel],
+      run: async () => {
+        throw new Error("turn exploded");
+      },
+    });
+
+    const harness = mockChatAgent(agent, { chatId: "chan-5" });
+    try {
+      await harness.sendChannelEvent({ event: { text: "cause an error", threadId: "chan-5" } });
+
+      expect(channel.acks).toHaveLength(1);
+      const finalSend = channel.sent[channel.sent.length - 1]!;
+      expect(finalSend.ctx.final).toBe(true);
+      expect(finalSend.ctx.previousRef).toBe(channel.acks[0]!.ref);
+      expect(finalSend.message.text).toBe("turn exploded");
+    } finally {
+      await harness.close();
+    }
+  });
+});
+
+describe("makeChannelStreamEditor", () => {
+  function streamConnector(send: (message: { text: string }) => Promise<{ ref?: string }>) {
+    return { delivery: "stream" as const, send } as never;
+  }
+
+  it("re-arms the debounce timer so text buffered during an in-flight edit still lands", async () => {
+    vi.useFakeTimers();
+    try {
+      const sends: string[] = [];
+      let releaseFirst!: () => void;
+      const firstGate = new Promise<void>((resolve) => {
+        releaseFirst = resolve;
+      });
+      let call = 0;
+      const editor = __makeChannelStreamEditorForTests(
+        streamConnector(async (message) => {
+          sends.push(message.text);
+          call += 1;
+          if (call === 1) await firstGate;
+          return { ref: "r1" };
+        }),
+        { event: {}, deliveryId: "d1" },
+        "r1"
+      );
+
+      editor.observe({ type: "text-delta", delta: "a" });
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(sends).toEqual(["a"]);
+
+      editor.observe({ type: "text-delta", delta: "b" });
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(sends).toEqual(["a"]);
+
+      releaseFirst();
+      await vi.advanceTimersByTimeAsync(0);
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(sends).toEqual(["a", "ab"]);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("does not fire a pending edit after stop()", async () => {
+    vi.useFakeTimers();
+    try {
+      const sends: string[] = [];
+      const editor = __makeChannelStreamEditorForTests(
+        streamConnector(async (message) => {
+          sends.push(message.text);
+          return { ref: "r1" };
+        }),
+        { event: {}, deliveryId: "d1" },
+        "r1"
+      );
+
+      editor.observe({ type: "text-delta", delta: "a" });
+      editor.stop();
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(sends).toEqual([]);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+});
+
+describe("makeChannelStreamTap", () => {
+  it("stops the editor when the stream completes (flush)", async () => {
+    let stops = 0;
+    const tap = __makeChannelStreamTapForTests({ observe: () => {}, stop: () => (stops += 1) });
+    const source = new ReadableStream({
+      start(controller) {
+        controller.enqueue({ type: "text-delta", delta: "x" });
+        controller.close();
+      },
+    });
+    const reader = source.pipeThrough(tap).getReader();
+    let done = false;
+    while (!done) {
+      done = (await reader.read()).done;
+    }
+    expect(stops).toBeGreaterThan(0);
+  });
+
+  it("stops the editor when the stream is cancelled mid-flight (abort)", async () => {
+    let stops = 0;
+    const tap = __makeChannelStreamTapForTests({ observe: () => {}, stop: () => (stops += 1) });
+    const source = new ReadableStream({
+      start(controller) {
+        controller.enqueue({ type: "text-delta", delta: "x" });
+      },
+    });
+    const reader = source.pipeThrough(tap).getReader();
+    await reader.read();
+    await reader.cancel("aborted");
+    expect(stops).toBeGreaterThan(0);
+  });
 });