feat(webhooks): channel egress stream mode (live debounced edits)

Implement delivery: "stream" for channel egress. Instead of one edit at turn complete ("final"), the framework tees the reply stream and debounce-edits the ack message as tokens arrive, so the reply grows live in the thread. Trailing-edge, one edit per interval (Slack chat.update is ~1/s); the turn-complete final edit remains the authoritative last write. Streaming needs an ack (a message ref to edit); without one it degrades to a single final edit. Best-effort: an edit failure is logged, not fatal.

SDK-only (the run loop's run() output pipe gains a channel tee, gated on delivery === "stream"); non-stream and non-channel turns pipe unchanged.

Unreleased feature, so no released-API impact.

Author: ericallam

.changeset/webhook-channels.md

@@ -5,7 +5,7 @@
 "trigger.dev": patch
 ---
 
-Add channels: chat frontends for agents. A `chat.agent` can now take `channels: [...]` alongside `events: [...]`. A channel is a bidirectional chat surface (Slack, etc.): a verified inbound event is routed to a durable per-key session and run as a turn (via the connector's `inbound()` mapper), and the agent's reply is posted back to the surface (`outbound()` + `send()`), so a thread is a real conversation with the agent. Egress runs in-run; "final" mode posts an ack placeholder at turn start and edits it to the answer at turn complete.
+Add channels: chat frontends for agents. A `chat.agent` can now take `channels: [...]` alongside `events: [...]`. A channel is a bidirectional chat surface (Slack, etc.): a verified inbound event is routed to a durable per-key session and run as a turn (via the connector's `inbound()` mapper), and the agent's reply is posted back to the surface (`outbound()` + `send()`), so a thread is a real conversation with the agent. Egress runs in-run with two delivery modes: "final" (default) posts an ack placeholder at turn start and edits it to the answer at turn complete; "stream" debounce-edits the message live as the reply streams.
 
 `chat.channels.custom({ source, key, inbound, outbound, send, ack, filter, delivery })` is the generic connector (bring your own source and egress). The new `@trigger.dev/slack` package ships `slack({ token, ... })`: the Slack Events API verifier, the `url_verification` handshake, a per-thread session key (with a `thread_ts || ts` fallback for thread starts), the real `chat.postMessage`/`chat.update` egress, and a mandatory self-message filter so the bot never replies to its own posts.
 

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

@@ -4708,6 +4708,70 @@ function defaultChannelOutbound(reply: ChannelReply): ChannelMessage | null {
   return reply.text ? { text: reply.text } : null;
 }
 
+// "stream" egress: as the reply streams, debounce-edit the ack message (previousRef) with the growing
+// text. Trailing-edge, one edit per interval (Slack chat.update ~1/s); the turn-complete final edit is
+// the authoritative last write. Best-effort: an edit failure is logged, never fatal.
+const CHANNEL_STREAM_EDIT_INTERVAL_MS = 1000;
+function makeChannelStreamEditor<TEvent>(
+  connector: ChannelConnector<TEvent>,
+  channelEvent: { event: unknown; deliveryId: string },
+  ackRef: string
+) {
+  const outbound = connector.outbound ?? defaultChannelOutbound;
+  let latest = "";
+  let timer: ReturnType<typeof setTimeout> | undefined;
+  let inFlight = false;
+  let stopped = false;
+
+  const edit = async () => {
+    if (stopped || inFlight) return;
+    const text = latest;
+    const message = outbound({
+      text,
+      message: { id: ackRef, role: "assistant", parts: [{ type: "text", text }] } as UIMessage,
+      final: false,
+      stopped: false,
+    });
+    if (!message) return;
+    inFlight = true;
+    try {
+      await connector.send!(message, {
+        event: channelEvent.event as TEvent,
+        deliveryId: channelEvent.deliveryId,
+        previousRef: ackRef,
+        mode: "stream",
+        final: false,
+      });
+    } catch (error) {
+      logger.warn("chat.agent: channel stream edit failed", { error });
+    } finally {
+      inFlight = false;
+    }
+  };
+
+  return {
+    observe(chunk: unknown) {
+      if (stopped) return;
+      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);
+      }
+    },
+    stop() {
+      stopped = true;
+      if (timer) {
+        clearTimeout(timer);
+        timer = undefined;
+      }
+    },
+  };
+}
+
 // 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.
@@ -7358,7 +7422,31 @@ function chatAgent<
                           resolveOnFinish!();
                         },
                       });
-                      await pipeChat(uiStream, { signal: combinedSignal, spanName: "stream response" });
+                      // Channel egress "stream" mode: tee the output and debounce-edit the ack message
+                      // as the reply streams. Needs the ack ref to edit; the final edit at turn-complete
+                      // is the authoritative last write. Non-stream / non-channel turns pipe unchanged.
+                      let streamForPipe: typeof uiStream = uiStream;
+                      if (
+                        wireChannelEvent &&
+                        channelConn?.send &&
+                        channelConn.delivery === "stream" &&
+                        channelAckRef &&
+                        uiStream instanceof ReadableStream
+                      ) {
+                        const editor = makeChannelStreamEditor(channelConn, wireChannelEvent, channelAckRef);
+                        streamForPipe = uiStream.pipeThrough(
+                          new TransformStream({
+                            transform(chunk, controller) {
+                              editor.observe(chunk);
+                              controller.enqueue(chunk);
+                            },
+                            flush() {
+                              editor.stop();
+                            },
+                          })
+                        );
+                      }
+                      await pipeChat(streamForPipe, { signal: combinedSignal, spanName: "stream response" });
                     }
                   } catch (error) {
                     // Handle AbortError from streamText gracefully