feat(channels): stream outbound messages - #2180
Conversation
🦋 Changeset detectedLatest commit: b681fb9 The changes in this PR will be included in the next version bump. This PR includes changesets to release 1 package
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
There was a problem hiding this comment.
Note
This report is out of date. Scroll down for Devin Review's latest report on this PR.
🔍 Devin Review: 1 flag
Not posted on this PR by your GitHub settings — view it in Devin Review. (Configure)
agents
@cloudflare/ai-chat
@cloudflare/codemode
hono-agents
@cloudflare/shell
@cloudflare/think
@cloudflare/voice
@cloudflare/worker-bundler
commit: |
🔴 agents import sizesMeasured 295 runtime imports as minified bundles. The primary size is gzip; raw minified size is included for diagnosis. An existing import growing by more than 10% is marked red. This report is informational.
Compared Changed imports (237)
All 265 current runtime imports
Reported by agent-think[bot]. |
There was a problem hiding this comment.
Note
This report is out of date. Scroll down for Devin Review's latest report on this PR.
Devin Review found 2 new potential issues.
1 flag not posted on this PR by your GitHub settings — view it in Devin Review. (Configure)
| return invalidSurface(); | ||
| } | ||
|
|
||
| const channel = this.#configuredChannel(surface.channelKey); |
There was a problem hiding this comment.
🟡 Unknown destinations leave producers running
Unknown destinations make #configuredChannel throw before chunks is cancelled. Producers can retain timers or external resources after delivery has failed.
| const channel = this.#configuredChannel(surface.channelKey); | |
| let channel: Channel; | |
| try { | |
| channel = this.#configuredChannel(surface.channelKey); | |
| } catch (error) { | |
| await chunks.cancel().catch(() => {}); | |
| throw error; | |
| } |
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
Note
This report is out of date. Scroll down for Devin Review's latest report on this PR.
Devin Review found 2 new potential issues.
2 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
| if ( | ||
| apiResponse?.ok === true && | ||
| typeof apiResponse.channel === "string" && | ||
| typeof apiResponse.ts === "string" | ||
| ) { | ||
| return { channelId: apiResponse.channel, ts: apiResponse.ts }; |
There was a problem hiding this comment.
🔴 Successful Slack streams truncate
callSlack rejects successful append and stop responses unless they repeat the channel and timestamp. The first append cancels the remaining answer, and the completed stream reports uncertain.
Prompt for agents
Slack Web API success payloads differ by method. chat.startStream returns the channel and timestamp needed to create a reference, while chat.appendStream and chat.stopStream can return only ok: true. Refactor callSlack in packages/channels/src/adapters/slack.ts so each caller validates the success shape appropriate to its method. Keep requiring channel and timestamp for chat.startStream and chat.postMessage, but accept an ok-only success for append and stop. Add tests whose append and stop responses contain only ok: true, matching the provider contract.
Was this helpful? React with 👍 or 👎 to provide feedback.
| ...(!isDirectMessage && { | ||
| recipientUserId: actorId, | ||
| recipientTeamId: teamId | ||
| }) |
There was a problem hiding this comment.
🟡 Direct-message approval replies fail
Direct-message approval surfaces omit the recipient IDs required by chat.startStream. Streaming a reply to those approval actions is rejected by Slack.
| ...(!isDirectMessage && { | |
| recipientUserId: actorId, | |
| recipientTeamId: teamId | |
| }) | |
| recipientUserId: actorId, | |
| recipientTeamId: teamId |
Was this helpful? React with 👍 or 👎 to provide feedback.
This PR adds outbound message streaming to
@cloudflare/channels, including native Slack and Telegram streaming, composite routing, and an AI SDK adapter.Why
Public API Surface
ChannelChunktext,reasoning,tool, andsourcestream chunksChannelChunkSourceReadableStream<ChannelChunk>inputChannelDeliveryOptionsChannelStreamOptionsChannel.streamOutboundResolver.streamChannelHost.streamconsumeChunksChunkConsumerStreamOutcometoChannelChunksfullStreamto Channel chunksDeliveryResultnow permits areferenceon itsuncertainarm.delivernow takesChannelDeliveryOptionsas its third argument instead of a bareChannelDeliveryContext.Architectural Changes
Composite Channels preserve the same contract:
fanouttees the source so each destination receives its own stream.fallbackretains consumed chunks and replays them after a confirmed failure.uncertainstops fallback to avoid potentially duplicating partial delivery.Code Changes
chat.startStream,chat.appendStream, andchat.stopStreamfor threads and direct messages. Top-level channel streams collect intochat.postMessage.sendMessageDraftfor private-chat previews and always finishes withsendMessage, including after an interrupted generation.TextSegmentJoinerto preserve semantic spacing around tool and metadata boundaries.fullStreamparts into neutral chunks and turns abort or error parts into abnormal stream completion.Compatibility
The third argument to
deliverchanged from:ChannelDeliveryContextto:
ChannelDeliveryOptionsCallers passing a delivery ID must wrap it: