Repository navigation
feat(channels): ChannelGateway routes webhooks to the conversation's agent - #2432
Conversation
🦋 Changeset detectedLatest commit: 1362902 The changes in this PR will be included in the next version bump. This PR includes changesets to release 2 packages
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 |
| return { | ||
| type: "message", | ||
| eventId: dispatchId, | ||
| message: { id: dispatchId, role: "user", parts } |
There was a problem hiding this comment.
🔴 Telegram edits duplicate transcript messages
When Telegram edits a message, toInboundEvent gives it a new transcript ID instead of the original message ID. normalizedMessage retains the original ID, but each edit has a new event ID. The conversation displays both versions instead of replacing the original.
Learn more
A transcript message ID identifies the message to replace, while an event ID identifies one delivery. Telegram edits keep the same provider message ID but have a fresh update ID in normalizedMessage. The gateway hashes the event ID and uses it for both fields. Channels.receive sends the message onward, so edits become separate transcript messages instead of replacing their original.
Example: A Telegram message with ID 42 arrives in update 100 and is edited in update 101. Both events refer to message 42, but the gateway emits two distinct transcript IDs, so the original and edit coexist.
Recommended fix: Derive the transcript message ID from the configured channel key and message.id, while keeping dispatchId derived from event.eventId for deduplication. Add an edited-message regression test.
Was this helpful? React with 👍 or 👎 to provide feedback.
| for (const envelope of result.events) { | ||
| await this.#dispatch(channelKey, channel, envelope); | ||
| } | ||
| return result.response; | ||
| } catch { | ||
| return new Response("Failed to handle Channel event", { status: 500 }); |
There was a problem hiding this comment.
🟡 Retried batches can duplicate conversations
When a later event fails, #webhook returns 500 after dispatching earlier events. A provider retry reroutes the entire batch; changed routes deliver earlier events to another conversation despite their stable dispatchId.
Learn more
The gateway dispatches each envelope before it acknowledges a webhook. If an envelope in a multi-event result fails, the catch returns HTTP 500, inviting a provider retry of the entire webhook. The Channels deduplication key includes conversation ID, so it only suppresses repeats sent to the same conversation. A changed route sends a previously accepted envelope to a second conversation.
Example: An ingress yields A and B; A reaches route one, then B fails. Before the provider retries the batch, the route for A changes to route two. A is accepted in both conversations.
Recommended fix: Record successful event IDs and their routing destinations durably before acknowledging or retrying batches, or define batch-level failure handling that does not re-dispatch successful envelopes into changed routes.
Was this helpful? React with 👍 or 👎 to provide feedback.
| const { message } = event; | ||
| const parts: MessagePart[] = [ | ||
| { type: "text", text: message.markdown ?? message.text } | ||
| ]; | ||
| for (const attachment of message.attachments ?? []) { | ||
| if (attachment.url && attachment.mediaType) { | ||
| parts.push({ | ||
| type: "file", | ||
| url: attachment.url, | ||
| mediaType: attachment.mediaType, | ||
| ...(attachment.name !== undefined && { filename: attachment.name }) | ||
| }); | ||
| } |
There was a problem hiding this comment.
🔴 Email subjects and attachments vanish
When inboundEmail supplies a subject and attachments, toInboundEvent drops the subject and every attachment without a URL. inboundEmail produces attachments without URLs, so the agent cannot inspect them.
Learn more
The email adapter extracts the subject as message.title and converts attachments into records containing media type, name, size and sometimes text, but no URL. The gateway projects inbound messages into text plus URL-backed file parts, dropping the subject and all email attachments. The agent receives only the body, even when the sender supplied files or a useful subject.
Example: An email titled "Invoice" with a text attachment invoice.txt arrives. The adapter emits the title and text attachment, but the dispatched agent sees only the body text.
Recommended fix: Preserve the title and available attachment facts in a serializable transcript representation. For binary attachments, expose a retrievable URL or another supported transport before creating file parts; do not silently discard them.
Was this helpful? React with 👍 or 👎 to provide feedback.
🟢 agents import sizes: 1 entry point changed, no growth
Changed exports (1)
How this worksEach runtime export is bundled on its own, minified, and gzipped. Changes smaller than 100 B, or smaller than 1% and 1 KiB, are ignored. Growth over 10% or 5 KiB is marked 🔴. This report is informational and does not fail CI. The workflow artifact contains every measurement. Compared |
ae78aad to
ac0986c
Compare
ac0986c to
e2d9799
Compare
agents
@cloudflare/ai-chat
@cloudflare/codemode
hono-agents
@cloudflare/shell
@cloudflare/think
@cloudflare/voice
@cloudflare/worker-bundler
commit: |
e2d9799 to
0e8de89
Compare
| try { | ||
| channel = this.#configuredChannel(surface.channelKey); | ||
| } catch (error) { | ||
| await chunks.cancel().catch(() => {}); |
There was a problem hiding this comment.
🔴 Unknown channel leaves stream delivery pending
When an unknown channel receives a teed stream, stream() waits for chunks.cancel() before rejecting. A tee branch's cancellation waits for its sibling, so an open sibling leaves delivery pending indefinitely.
Learn more
A ReadableStream.tee() branch's cancellation promise can remain pending while the sibling branch stays open. The gateway attempts to cancel an undeliverable source, but awaiting that promise delays the original unknown-channel rejection indefinitely. consumeChunks starts cancellation before finalization for the same reason.
Example: A caller tees a generation into a display branch and a delivery branch, then calls gateway.stream() with an unknown channel key. The display branch remains open; delivery never rejects, although the channel lookup already failed.
Recommended fix: Start chunks.cancel() without awaiting it in this error branch, and retain the rejection handler for cleanup failures. Verify that an open tee sibling does not delay the unknown-channel rejection.
| await chunks.cancel().catch(() => {}); | |
| void chunks.cancel().catch(() => {}); |
Was this helpful? React with 👍 or 👎 to provide feedback.
0e8de89 to
7ed4bca
Compare
7ed4bca to
cc492a7
Compare
| ); | ||
| } | ||
| await this.#agent(route).receive(toInboundEvent(event, dispatchId), { | ||
| route, |
There was a problem hiding this comment.
🔴 Distinct routes share one conversation
When two routes select the same agent, receive sends both events without a conversation ID. Channels.receive ignores route and puts both events in the default conversation.
Learn more
A route selects the agent object, while a conversation ID selects a session inside that object. The gateway sends only route, and Channels.receive falls back to defaultConversation when conversationId is absent. Consequently, two routes selecting one agent both use its default session, even though the gateway assigned them different routes.
Example: An application maps thread:a and thread:b to one agent object. Messages from both threads enter that object's default conversation; separate transcripts were expected.
Recommended fix: Have Channels.receive use from.route as the conversation ID when no explicit conversationId is supplied, or have the gateway supply the intended conversation ID explicitly. Preserve any deliberate harness-default mapping through a clearly defined separate option.
Was this helpful? React with 👍 or 👎 to provide feedback.
| @@ -1,5 +1,6 @@ | |||
| export * from "./channel"; | |||
| export * from "./conversations"; | |||
| export * from "./gateway"; | |||
There was a problem hiding this comment.
…agent This PR adds `ChannelGateway`, the Worker entry point that authenticates channel webhooks and routes each event to the agent that holds its conversation. ## Why - Webhooks arrive at the Worker, not the agent. The Worker has to verify them, normalize them, pick a route, and call the right Durable Object. - Keeping routing in the gateway lets the agent stay unaware of providers. The agent only sees `GatewayEvent`s and a `GatewayOrigin`. - A route picks the agent object through `agent(route)`, so one Worker can serve several agent classes by naming routes after them. Routes default to the event's thread, with `defaultRoute` and `onRoute` for applications that group threads differently. A webhook names no conversation, so it lands in the agent's default one. ## Public API Surface Additive, exported from `agents/experimental/channels`: | Symbol | Kind | Notes | | --- | --- | --- | | `ChannelGateway` | class | `fetch` for webhooks, plus outbound `deliver` | | `ChannelGatewayOptions` | type | `channels`, `agent(route)`, `defaultRoute`, `findUser`, `onRoute` | | `GatewayAgent`, `ChannelRouteEvent` | types | The agent stub shape and the route observer event | ## Code Changes - `gateway.ts` verifies, normalizes, routes and dispatches, with a stable `dispatchId` per Channel and event. - `stream.ts` and `channel.ts` gain the stream collection the gateway uses for outbound delivery. - `gateway.test.ts` covers routing, identity and dispatch. - Minor changeset for `agents`.
cc492a7 to
1362902
Compare
This PR adds
ChannelGateway, the Worker entry point that authenticates channel webhooks and routes each event to the agent that holds its conversation.Why
GatewayEvents and aGatewayOrigin.agent(route), so one Worker can serve several agent classes by naming routes after them. Routes default to the event's thread, withdefaultRouteandonRoutefor applications that group threads differently. A webhook names no conversation, so it lands in the agent's default one.Public API Surface
Additive, exported from
agents/experimental/channels:ChannelGatewayfetchfor webhooks, plus outbounddeliverChannelGatewayOptionschannels,agent(route),defaultRoute,findUser,onRouteGatewayAgent,ChannelRouteEventCode Changes
gateway.tsverifies, normalizes, routes and dispatches, with a stabledispatchIdper Channel and event.stream.tsandchannel.tsgain the stream collection the gateway uses for outbound delivery.gateway.test.tscovers routing, identity and dispatch.agents.