Repository navigation
feat(channels): the Web Channel, its browser client and WebSockets.use - #2433
Conversation
🦋 Changeset detectedLatest commit: 1b17acd 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 |
| for (const [responseId, response] of this.#responses) { | ||
| if (!running.has(responseId)) { | ||
| this.#responses.delete(responseId); | ||
| continue; | ||
| } |
There was a problem hiding this comment.
🔴 Unfinished response output lost on reconnect
If a response continues after its turn settles, #resume deletes its cached cursor on reconnect. Snapshots omit settled turns, so the client loses remaining chunks and the response ending.
Learn more
A response stream can continue delivering chunks after its turn settles. The client keeps such a response locally to emit activity until its end, but a reconnect snapshot includes only queued, running, or awaiting-input turns (ConversationSnapshot). The resume logic deletes every response absent from the running set, including one still being read. Its prior cursor and remaining activity are then lost.
Example: A turn settles while response R has sent five chunks but not ended. The socket closes and reconnects. The snapshot contains no running turn for R; the client deletes R instead of subscribing from cursor five.
Recommended fix: Retain and resubscribe responses whose reads have not ended, even if the snapshot no longer includes their turn. Reconcile their transcript visibility independently from the read lifecycle, and release each read only after its terminal frame.
Was this helpful? React with 👍 or 👎 to provide feedback.
| const defaultWeb = (request: Request): GatewayWebIdentity | undefined => { | ||
| const match = /^\/channels\/([^/]+)$/.exec(new URL(request.url).pathname); | ||
| return match | ||
| ? { route: decodeURIComponent(match[1]), participant: { id: "anonymous" } } | ||
| : undefined; |
There was a problem hiding this comment.
🟥 Default Web upgrades expose conversations anonymously
Without a custom web resolver, any caller can upgrade /channels/<route> as anonymous. The gateway forwards that identity to the conversation agent, allowing the caller to read its snapshot and submit events.
Was this helpful? React with 👍 or 👎 to provide feedback.
🔴 agents import sizes: 1 entry point over threshold
Changed exports (6)
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 |
agents
@cloudflare/ai-chat
@cloudflare/codemode
hono-agents
@cloudflare/shell
@cloudflare/think
@cloudflare/voice
@cloudflare/worker-bundler
commit: |
| for (const { event } of this.#outbox.values()) { | ||
| this.#send({ type: "channels:event", event }); | ||
| } |
There was a problem hiding this comment.
🔴 Create acknowledgments trigger repeated conversation creation
When channels:snapshot precedes a create acknowledgment, #outbox resends the create into the new conversation. Deduplication includes conversation ID, so each replay creates another conversation.
Learn more
The server's #message sends the new snapshot before acknowledging a successful create or fork. The client treats every snapshot as a reconnect and replays the entire outbox. #dispatch deduplicates on conversation ID and event ID together, so replaying that event after following a different conversation starts another create or fork. The next snapshot repeats the process.
Example: Alice sends conversation-create with event ID e1 in main. The server creates s1 and sends its snapshot. Before receiving the ack, Alice replays e1 in s1, which creates s2; the process continues.
Recommended fix: Do not replay pending events when a snapshot moves the client to a different conversation. Settle or reject old-conversation events explicitly, and only replay them after reconnecting to the same conversation.
Was this helpful? React with 👍 or 👎 to provide feedback.
| } | ||
| this.#lists.clear(); | ||
| this.#update(); | ||
| if (!this.#closed) setTimeout(() => this.#connect(), 1000); |
There was a problem hiding this comment.
🔴 Pending reconnect strands a followed socket
If follow() runs during a reconnect delay, the old close timer opens another socket and replaces #socket. The followed socket stays open, but its snapshots and acknowledgments are ignored.
Learn more
A close schedules a future #connect() unconditionally. follow opens a new socket immediately but cannot cancel that scheduled reconnect. When the old timer fires, #connect overwrites #socket; its message listener ignores all traffic on the followed socket because it is no longer the current socket. Both connections remain open.
Example: A socket closes at t=0 and schedules a reconnect for t=1000 ms. At t=500 ms the user follows f1, opening socket B. At t=1000 ms the timer opens socket C; frames from B are dropped even though B remains connected.
Recommended fix: Track and clear pending reconnect timers during follow() and close(), and guard the timer callback with the socket or connection generation it was created for.
Was this helpful? React with 👍 or 👎 to provide feedback.
| send(event: ClientEvent): Promise<void> { | ||
| const eventId = event.eventId ?? crypto.randomUUID(); | ||
| // SAFETY: the same event, with its id filled in. | ||
| const complete = { ...event, eventId } as InboundEvent; | ||
| return new Promise((resolve, reject) => { | ||
| const accepted = () => { | ||
| // Other connections get the message from Channels; this one adds its | ||
| // own once the agent has accepted it. | ||
| if (complete.type === "message") this.#upsert([complete.message]); | ||
| resolve(); | ||
| }; | ||
| this.#outbox.set(eventId, { event: complete, resolve: accepted, reject }); |
There was a problem hiding this comment.
🟡 Events sent after closing never settle
Calling send() after close() adds an event to #outbox without any future connection or acknowledgment. Its returned promise remains pending forever.
Learn more
close sets #closed and rejects pending events, but send does not check that flag when adding later events. A closed client will never reconnect to acknowledge those events.
Example: Call client.close() followed by await client.send({ type: "conversation-create" }). The await never finishes.
Recommended fix: Reject immediately at the start of send when #closed is true.
Was this helpful? React with 👍 or 👎 to provide feedback.
| "./experimental/channels/web": { | ||
| "types": "./dist/experimental/channels/web/index.d.ts", | ||
| "import": "./dist/experimental/channels/web/index.js", | ||
| "require": "./dist/experimental/channels/web/index.js" | ||
| }, | ||
| "./experimental/channels/web/client": { | ||
| "types": "./dist/experimental/channels/web/client.d.ts", | ||
| "import": "./dist/experimental/channels/web/client.js", | ||
| "require": "./dist/experimental/channels/web/client.js" |
There was a problem hiding this comment.
| if (request.headers.get("Upgrade")?.toLowerCase() === "websocket") { | ||
| const web = await this.#upgrade(request); | ||
| if (web) return web; | ||
| } | ||
| return this.#webhook(request); |
There was a problem hiding this comment.
🟥 Unmatched upgrades retain forged identity headers
When #upgrade returns undefined, fetch leaves the client's x-channels-identity header intact. A fallback forwarding that upgrade to a Channels agent lets the client choose its participant and conversation.
Was this helpful? React with 👍 or 👎 to provide feedback.
| return { | ||
| route: decodeURIComponent(match[1]), | ||
| ...(match[2] !== undefined && { | ||
| conversationId: decodeURIComponent(match[2]) | ||
| }), | ||
| participant: { id: "anonymous" } | ||
| }; |
| const { eventId } = frame.event; | ||
| let result; | ||
| try { | ||
| result = await this.#mounted().dispatch( | ||
| frame.event, | ||
| origin(identity, connection.id) | ||
| ); |
There was a problem hiding this comment.
This PR adds the Web Channel, which connects browsers and terminals to a conversation over WebSockets, and its client. ## Why - Clients need to join a conversation, receive the snapshot and live updates, and send messages, approvals and client tool results. - They also need to manage conversations: create, fork and reset one, list the agent's conversations and move between them. A create or fork moves only the connection that asked, and other connections stay where they are. - The agent already owns its WebSocket connections through the `WebSockets` capability. The Web Channel shares those connections instead of opening its own, so `WebSockets` gains `use` for adding protocol handlers that can claim a message. `new WebChannel()` takes the agent's `WebSockets` when Channels mounts it, or creates one that `channels.websockets` exposes. - The gateway resolves who is connecting from trusted request data and passes it in a header, so the Web Channel never trusts identity from the client. ## Public API Surface Additive: | Symbol | Entry point | Notes | | --- | --- | --- | | `WebChannel`, `WebChannelOptions` | `agents/experimental/channels/web` | Server side of the Web Channel; options are optional | | `ClientFrame`, `ServerFrame`, `parseClientFrame`, `WEB_IDENTITY_HEADER`, `WebIdentity` | `agents/experimental/channels/web` | Wire protocol. Every server frame names its conversation | | `WebChannelClient`, `WebChannelClientOptions`, `WebChannelClientState`, `WebChannelActivity` | `agents/experimental/channels/web/client` | Client, with `listConversations()` and `follow(id)` | | `ChannelGatewayOptions.web`, `GatewayWebIdentity` | `agents/experimental/channels` | Resolve a WebSocket upgrade to a route and, optionally, a conversation | | `WebSockets.use` | `agents/websockets` | Add handlers that run before the configured ones | ## Code Changes - `web/channel.ts` serves the protocol on shared connections. The snapshot lists the conversation operations a connection may send, and `channels:list-conversations` is answered from the harness's sessions. The list is pushed again after a create or fork. - `web/client.ts` keeps transcript and turn state with `apply-chunks.ts`. It follows a created or forked conversation and reconnects to the one it follows; `followUrl` overrides how that URL is built. - The gateway upgrades `/channels/<route>` and `/channels/<route>/<conversation>` by default. - A `ChannelsHarnessObject` in the worker tests plays a scripted harness and drives both sides end to end. `gateway-rpc.test.ts` sends a webhook through a `ChannelGateway` to the agent's `receive` over RPC. - Minor changeset for `agents`. # Conflicts: # packages/agents/AGENTS.md # Please enter the commit message for your changes. Lines starting # with '#' will be kept; you may remove them yourself if you want to. # An empty message aborts the commit. # # interactive rebase in progress; onto 74570a1 # Last commands done (11 commands done): # update-ref cc492a7 # pick 1c57b3e # feat(channels): the Web Channel, its browser client and WebSockets.use # Next commands to do (16 remaining commands): # update-ref 1c57b3e # pick 245a7b9 # feat(harness): add an AI SDK harness # You are currently rebasing branch 'channels/13-harness-bridge' on '74570a19a'. # # Changes to be committed: # new file: .changeset/channels-web.md # modified: packages/agents/AGENTS.md # modified: packages/agents/package.json # modified: packages/agents/scripts/build.ts # new file: packages/agents/src/experimental/channels/__tests__/apply-chunks.test.ts # new file: packages/agents/src/experimental/channels/__tests__/web-client.test.ts # new file: packages/agents/src/experimental/channels/apply-chunks.ts # modified: packages/agents/src/experimental/channels/gateway.ts # modified: packages/agents/src/experimental/channels/index.ts # new file: packages/agents/src/experimental/channels/web/channel.ts # new file: packages/agents/src/experimental/channels/web/client.ts # new file: packages/agents/src/experimental/channels/web/index.ts # new file: packages/agents/src/experimental/channels/web/protocol.ts # modified: packages/agents/src/tests-d/lifecycle-export.test-d.ts # new file: packages/agents/src/tests/capabilities/channels.ts # new file: packages/agents/src/tests/channels/conversations.test.ts # new file: packages/agents/src/tests/channels/gateway-rpc.test.ts # modified: packages/agents/src/tests/worker.ts # modified: packages/agents/src/tests/wrangler.jsonc # modified: packages/agents/src/websockets/options.ts # modified: packages/agents/src/websockets/websockets.ts #
This PR adds the Web Channel, which connects browsers and terminals to a conversation over WebSockets, and its client.
Why
WebSocketscapability. The Web Channel shares those connections instead of opening its own, soWebSocketsgainsusefor adding protocol handlers that can claim a message.new WebChannel()takes the agent'sWebSocketswhen Channels mounts it, or creates one thatchannels.websocketsexposes.Public API Surface
Additive:
WebChannel,WebChannelOptionsagents/experimental/channels/webClientFrame,ServerFrame,parseClientFrame,WEB_IDENTITY_HEADER,WebIdentityagents/experimental/channels/webWebChannelClient,WebChannelClientOptions,WebChannelClientState,WebChannelActivityagents/experimental/channels/web/clientlistConversations()andfollow(id)ChannelGatewayOptions.web,GatewayWebIdentityagents/experimental/channelsWebSockets.useagents/websocketsCode Changes
web/channel.tsserves the protocol on shared connections. The snapshot lists the conversation operations a connection may send, andchannels:list-conversationsis answered from the harness's sessions. The list is pushed again after a create or fork.web/client.tskeeps transcript and turn state withapply-chunks.ts. It follows a created or forked conversation and reconnects to the one it follows;followUrloverrides how that URL is built./channels/<route>and/channels/<route>/<conversation>by default.ChannelsHarnessObjectin the worker tests plays a scripted harness and drives both sides end to end.gateway-rpc.test.tssends a webhook through aChannelGatewayto the agent'sreceiveover RPC.agents.