feat(webhooks): channel egress (reply round-trip, final mode)

Close the channel loop: an agent turn triggered by a channel delivery now posts its reply back to the channel. In "final" mode the framework posts an ack placeholder at turn start, then edits it to the answer at turn complete.

The connector gains the egress half: outbound (map the turn's reply to a channel message, or null to post nothing) and send (post/edit, provider-specific). The framework owns the delivery mode, the message-ref threading, and (for the future stream mode) the debounce. A new generic chat.channels.custom lets you supply your own send for non-Slack surfaces; chat.channels.slack's real send is the @trigger.dev/slack package.

SDK-only: egress runs in-run (it needs the run's reply and channel context), driven from the turn boundaries. A manual-pipe turn has no captured response, so it opts out. delivery defaults to "final"; token-level "stream" (debounced live edits) is a fast-follow.

Unreleased feature, so no released-API impact.

Author: ericallam

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

@@ -57,6 +57,8 @@ import type {
   WebhookVerifierArtifact,
   WebhookSecretProvisioning,
   WebhookHmacConfig,
+  AnyWebhookSource,
+  InferWebhookEvent,
 } from "@trigger.dev/core/v3";
 import { chatEvent, normalizeKeyString } from "./webhooks.js";
 // Runtime VALUES go through the ESM/CJS shim so the CJS build can `require`
@@ -4596,8 +4598,35 @@ export type ChatResumeEvent<TClientData = unknown, TUIM extends UIMessage = UIMe
 // A connector registers an agent-scoped endpoint that delivers each verified event as a channel
 // message; the run applies inbound() to the raw event to get the turn's message. The loop guard is the
 // server-side `filter` (an ignored event never becomes a delivery), so inbound is a pure mapper.
+// The reply round-trips (egress) in-run: outbound() maps the turn's reply to a channel message, send()
+// posts/edits it. delivery "final" posts an ack at turn start then edits it to the answer at turn end.
 export type ChannelMessageInput = string | UIMessage;
 
+// The channel message a provider posts. Framework requires `text`; providers extend (Slack adds blocks).
+export type ChannelMessage = { text: string; [key: string]: unknown };
+
+// What outbound() receives: enough to decide what (or whether, null) to post. For "stream", `text` is
+// the accumulated text so far.
+export type ChannelReply = {
+  text: string;
+  message: UIMessage;
+  final: boolean;
+  stopped: boolean;
+  error?: unknown;
+};
+
+export type ChannelSendCtx<TEvent = unknown> = {
+  event: TEvent;
+  deliveryId: string;
+  previousRef?: string;
+  mode: "final" | "stream";
+  final: boolean;
+};
+
+// Reserved for 2c: resolve a provider credential keyed by the incoming event's installation (team_id),
+// not the connector (one connector serves many Slack workspaces). The in-run send() calls it at post time.
+export type ResolveChannelToken<TEvent = unknown> = (event: TEvent) => Promise<string>;
+
 declare const channelEventPhantom: unique symbol;
 
 export interface ChannelConnector<TEvent = unknown> {
@@ -4608,7 +4637,11 @@ export interface ChannelConnector<TEvent = unknown> {
   secretProvisioning?: WebhookSecretProvisioning;
   filter?: string;
   inbound: (event: TEvent) => ChannelMessageInput;
-  delivery: "stream" | "final"; // egress mode (used by the reply round-trip; a later phase)
+  // Egress (optional; a channel can be inbound-only). Fires only when `send` is set.
+  outbound?: (reply: ChannelReply) => ChannelMessage | null;
+  ack?: (event: TEvent) => ChannelMessage | null; // placeholder posted at turn start ("final" mode)
+  send?: (message: ChannelMessage, ctx: ChannelSendCtx<TEvent>) => Promise<{ ref?: string }>;
+  delivery: "final" | "stream"; // "final" (ack + edit) is v1; "stream" (debounced edits) is a fast-follow
   readonly [channelEventPhantom]?: TEvent;
 }
 export type AnyChannelConnector = ChannelConnector<any>;
@@ -4656,11 +4689,14 @@ export function slackChannel<
   id: string;
   key: ValidatedWebhookKey<TEvent, TKey>;
   inbound: (event: TEvent) => ChannelMessageInput;
+  outbound?: (reply: ChannelReply) => ChannelMessage | null;
   filter?: TFilter & ValidateWebhookFilter<TEvent, TFilter>;
-  delivery?: "stream" | "final";
+  delivery?: "final" | "stream";
 }): ChannelConnector<TEvent> {
-  const { id, key, inbound, filter, delivery } = options;
+  const { id, key, inbound, outbound, filter, delivery } = options;
   resourceCatalog.registerDeclaredSessionWebhook(id);
+  // `send` is intentionally absent here: the real Slack sender (bot token + chat.postMessage/update)
+  // is the @trigger.dev/slack provider package. Until then a Slack channel is inbound-only.
   return {
     id,
     source: "slack",
@@ -4669,13 +4705,56 @@ export function slackChannel<
     secretProvisioning: "integrator",
     filter,
     inbound,
-    delivery: delivery ?? "stream",
+    outbound,
+    delivery: delivery ?? "final",
   } as ChannelConnector<TEvent>;
 }
 
+/**
+ * A generic chat-frontend channel over any verified source. Unlike `slack()`, you supply the egress
+ * `send` yourself (post/edit the reply back to your surface), so the whole round-trip is under your
+ * control. Use for providers without a preset, or to test the channel round-trip end to end.
+ */
+export function chatChannelCustom<
+  TSource extends AnyWebhookSource,
+  const TKey extends string = string,
+  const TFilter extends string = string,
+>(options: {
+  id: string;
+  source: TSource;
+  key: ValidatedWebhookKey<InferWebhookEvent<TSource>, TKey>;
+  inbound: (event: InferWebhookEvent<TSource>) => ChannelMessageInput;
+  outbound?: (reply: ChannelReply) => ChannelMessage | null;
+  ack?: (event: InferWebhookEvent<TSource>) => ChannelMessage | null;
+  send?: (
+    message: ChannelMessage,
+    ctx: ChannelSendCtx<InferWebhookEvent<TSource>>
+  ) => Promise<{ ref?: string }>;
+  filter?: TFilter & ValidateWebhookFilter<InferWebhookEvent<TSource>, TFilter>;
+  delivery?: "final" | "stream";
+}): ChannelConnector<InferWebhookEvent<TSource>> {
+  const { id, source, key, inbound, outbound, ack, send, filter, delivery } = options;
+  resourceCatalog.registerDeclaredSessionWebhook(id);
+  return {
+    id,
+    source: source.provider,
+    key: normalizeKeyString(key as string),
+    verifierArtifact: source.verifier,
+    secretProvisioning: source.secretProvisioning,
+    filter,
+    inbound,
+    outbound,
+    ack,
+    send,
+    delivery: delivery ?? "final",
+  } as ChannelConnector<InferWebhookEvent<TSource>>;
+}
+
 const chatChannels = {
   /** Slack as a chat frontend. See {@link slackChannel}. */
   slack: slackChannel,
+  /** A generic chat-frontend channel over any source, with your own egress. See {@link chatChannelCustom}. */
+  custom: chatChannelCustom,
 };
 
 // Turn a channel connector's inbound() result into a user UIMessage for the turn.
@@ -4684,6 +4763,18 @@ function toUserUIMessage(input: ChannelMessageInput, messageId: string): UIMessa
   return { id: messageId, role: "user", parts: [{ type: "text", text: input }] } as UIMessage;
 }
 
+// Sum the text parts of a UIMessage (the reply text egress posts).
+function uiMessageText(message: UIMessage): string {
+  return (message.parts ?? [])
+    .map((p) => (p && typeof p === "object" && "text" in p ? String((p as { text: unknown }).text) : ""))
+    .join("");
+}
+
+// Default outbound: post the reply text, or nothing when empty (a tool-only / empty turn).
+function defaultChannelOutbound(reply: ChannelReply): ChannelMessage | null {
+  return reply.text ? { text: reply.text } : null;
+}
+
 // 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.
@@ -6511,15 +6602,37 @@ function chatAgent<
               void _hsm;
               // Channel-delivered turn: map the raw provider event to the turn's message via the
               // connector's inbound(), resolved by connectorId. The loop guard is the connector's
-              // server-side filter, so inbound is a pure mapper here.
+              // server-side filter, so inbound is a pure mapper here. `channelConn` + `channelAckRef`
+              // are read again at turn-complete to egress the reply.
               let effectiveIncomingMessage = incomingMessage;
+              let channelConn: AnyChannelConnector | undefined;
+              let channelAckRef: string | undefined;
               if (wireChannelEvent) {
-                const connector = channels?.find((c) => c.id === wireChannelEvent.connectorId);
-                if (connector) {
+                channelConn = channels?.find((c) => c.id === wireChannelEvent.connectorId);
+                if (channelConn) {
                   effectiveIncomingMessage = toUserUIMessage(
-                    connector.inbound(wireChannelEvent.event),
+                    channelConn.inbound(wireChannelEvent.event),
                     currentWirePayload.messageId ?? wireChannelEvent.deliveryId
                   ) as typeof incomingMessage;
+                  // "final" egress: post an ack placeholder now so a message ref exists to edit at turn end.
+                  if (channelConn.send && channelConn.ack) {
+                    const ackMessage = channelConn.ack(wireChannelEvent.event);
+                    if (ackMessage) {
+                      try {
+                        const ackResult = await channelConn.send(ackMessage, {
+                          event: wireChannelEvent.event,
+                          deliveryId: wireChannelEvent.deliveryId,
+                          mode: channelConn.delivery,
+                          final: false,
+                        });
+                        channelAckRef = ackResult?.ref;
+                      } catch (ackError) {
+                        logger.warn("chat.agent: channel ack post failed; continuing", {
+                          error: ackError,
+                        });
+                      }
+                    }
+                  }
                 }
               }
               const incomingMessages: TUIMessage[] = effectiveIncomingMessage
@@ -7744,6 +7857,32 @@ function chatAgent<
                     turnAccessToken
                   );
 
+                  // Channel egress ("final"): map the turn's reply via outbound() and send it, editing
+                  // the ack placeholder posted at turn start (previousRef). Best-effort: an egress failure
+                  // is logged, not fatal. A manual-pipe turn has no responseMessage, so it opts out.
+                  if (wireChannelEvent && channelConn?.send && turnCompleteEvent.responseMessage) {
+                    const outbound = channelConn.outbound ?? defaultChannelOutbound;
+                    const channelMessage = outbound({
+                      text: uiMessageText(turnCompleteEvent.responseMessage),
+                      message: turnCompleteEvent.responseMessage,
+                      final: true,
+                      stopped: turnCompleteEvent.stopped,
+                    });
+                    if (channelMessage) {
+                      try {
+                        await channelConn.send(channelMessage, {
+                          event: wireChannelEvent.event,
+                          deliveryId: wireChannelEvent.deliveryId,
+                          previousRef: channelAckRef,
+                          mode: channelConn.delivery,
+                          final: true,
+                        });
+                      } catch (egressError) {
+                        logger.warn("chat.agent: channel egress send failed", { error: egressError });
+                      }
+                    }
+                  }
+
                   // Fire onTurnComplete — stream is closed, use for persistence.
                   if (onTurnComplete) {
                     await tracer.startActiveSpan(