Repository navigation
feat(channels): Channels, the capability that serves a conversation's surfaces - #2431
Conversation
🦋 Changeset detectedLatest commit: e734256 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 |
| const streaming = await this.#options.streams.list({ | ||
| state: "streaming", | ||
| // Streams lists 100 by default. A loop until none are left would drop | ||
| // the cap, but `list()` matches tags exactly, not by prefix, so streams | ||
| // other capabilities left streaming would come back on every pass. | ||
| limit: 1000 | ||
| }); |
There was a problem hiding this comment.
🔴 Older responses stay live after restart
If Streams.list returns 1,000 newer live streams, onStart never visits older responses. Their readers tail indefinitely because no producer survives restart.
Learn more
Streams.list orders all streams newest first and enforces the requested limit before any tag filtering here. Startup marks only the returned Channels streams interrupted. Any older Channels stream remains streaming, and readBatches waits for another append or settlement even though its producer was lost on restart.
Example: A Durable Object has 1,000 newer streaming streams from another capability and one older response r1. Startup skips r1; subscribing to it replays stored chunks and then waits forever.
Recommended fix: Add paginated/prefix-filtered enumeration to Streams or otherwise iterate all streaming rows without repeatedly returning unrelated streams, and settle every Channels response before startup finishes.
Was this helpful? React with 👍 or 👎 to provide feedback.
| await this.#dispatch( | ||
| event.type !== "approval-response" || "turnId" in event | ||
| ? event | ||
| : await this.#withTurn(event, origin), | ||
| origin | ||
| ); |
There was a problem hiding this comment.
🟡 Concurrent approvals fail before earlier events finish
When an approval arrives during an earlier event, #withTurn reads the snapshot before that event completes. It rejects instead of waiting for #queues to expose the pending approval.
Learn more
Events are serialized by #dispatch, but a webhook approval with no turn ID resolves its turn before entering that queue. When the prior event is still creating the approval request, its message is missing from the snapshot read by #withTurn. The approval rejects rather than taking effect after the earlier event.
Example: Message e1 starts a turn and is still writing an approval-requested tool part when webhook event e2 approves it. e2 reads the old snapshot and fails, although e1 then finishes saving the approval.
Recommended fix: Enqueue turn resolution and the subsequent onEvent call as one serialized operation for the conversation, retaining the event-ID deduplication behavior.
Was this helpful? React with 👍 or 👎 to provide feedback.
| /** Recent events by key, each with the outcome repeats receive. */ | ||
| readonly #seen = new Map<string, Promise<void>>(); |
| const ending = await streams.status(responseId); | ||
| return ending?.state === "errored" ? "interrupted" : "ended"; |
🟢 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 |
717ffe7 to
34b446d
Compare
agents
@cloudflare/ai-chat
@cloudflare/codemode
hono-agents
@cloudflare/shell
@cloudflare/think
@cloudflare/voice
@cloudflare/worker-bundler
commit: |
e1d0658 to
c0fa1ec
Compare
| async #reset(conversationId: string): Promise<void> { | ||
| const streams = this.streams; | ||
| const tag = TAG_PREFIX + conversationId; | ||
| for (;;) { | ||
| const batch = await streams.list({ tag }); | ||
| if (batch.length === 0) break; | ||
| for (const { streamId, state } of batch) { | ||
| if (state === "streaming") (await streams.open(streamId)).close(); | ||
| await streams.delete(streamId); | ||
| } | ||
| } | ||
| // Events accepted before the reset may be sent again to rebuild it. | ||
| const prefix = JSON.stringify([conversationId]).slice(0, -1) + ","; | ||
| for (const key of this.#seen.keys()) { | ||
| if (key.startsWith(prefix)) this.#seen.delete(key); | ||
| } | ||
| await this.#publish(conversationId, { type: "reset" }); |
There was a problem hiding this comment.
🔴 Reset retains turns from old context
After a session reset, #reset leaves durable turn and alias records intact. Snapshots show old awaiting-input turns, and reused operation IDs can inherit stale statuses.
Learn more
A reset starts a new session context, but Channels stores turn and alias records separately from response streams. HarnessConversations reads those records for snapshots and new operations. Deleting streams and the in-memory deduplication entries leaves the prior context's turn records available. A later operation with a reused ID can inherit a settled record, so the new run is not reported correctly.
Example: A session has turn op-1 awaiting a tool result. Resetting clears the harness transcript, but a fresh snapshot still includes op-1; resending op-1 in the new context does not create a new queued turn.
Recommended fix: Clear both channels:harness:turn: and channels:harness:alias: records scoped to this session when processing the reset event, before publishing the reset update. Keep the deletion scoped to one conversation.
Was this helpful? React with 👍 or 👎 to provide feedback.
| await this.#save(conversationId, { | ||
| turnId: operationId, | ||
| startedBy: operationId, | ||
| messageId: operationId, | ||
| status: "queued" | ||
| }); |
There was a problem hiding this comment.
🔴 Harness message IDs lose turn outcomes
When a harness submission uses a distinct messageId, #operation records its operation ID as the message ID. #settle then misses its assistant messages and marks unanswered tool calls completed.
Learn more
The shared harness accepts a caller-specified messageId independent of its operation ID. Its queued status contains only the operation ID, so the turn record created here cannot identify the submitted user's message. #settle locates the user message using turn.messageId and collects subsequent assistant messages. With a different ID it finds none, so a pending tool call is reported as completed instead of awaiting input.
Example: A harness client submits operation op-7 with messageId: "user-7"; the model emits an assistant message requesting approval. The resulting turn has no messageIds and reports completed, hiding the approval from the snapshot.
Recommended fix: Provide the real message ID with the harness operation status or associate the operation with the user message event, then persist that ID in the turn before settlement. Check direct harness submissions using custom message IDs in tests.
Was this helpful? React with 👍 or 👎 to provide feedback.
| case "cancel": { | ||
| const turn = this.#turnRecord(conversationId, event.turnId); | ||
| await link.session.abort(turn?.operation ?? event.turnId); | ||
| return; |
There was a problem hiding this comment.
🟡 Awaiting-input turns survive cancellation
When a participant cancels an awaiting-input turn, onEvent aborts its already completed operation. No status transition follows, so the turn remains awaiting input.
Learn more
An operation can finish while its assistant message still requests a tool answer. #settle then saves the turn as awaiting-input, while the operation is already done. Cancel passes the saved operation ID to the harness's abort method, which stops operations, but no later event is guaranteed for a completed operation. The durable turn therefore stays open in snapshots.
Example: Operation op-1 finishes after requesting approval and its turn is awaiting input. A cancel of op-1 calls abort("op-1") after it has finished; the snapshot still lists the approval turn.
Recommended fix: For an awaiting-input turn, explicitly persist and publish an aborted turn after canceling the relevant work. Preserve normal harness-driven abort handling for queued and running operations.
Was this helpful? React with 👍 or 👎 to provide feedback.
| @@ -1,4 +1,5 @@ | |||
| export * from "./channel"; | |||
| export * from "./conversations"; | |||
There was a problem hiding this comment.
… surfaces
This PR adds `Channels`, a Lifecycle capability that serves an agent harness's sessions as conversations to every surface that joins them.
## Why
- The harness owns the transcript and turns. Something has to carry inbound events to it, publish turn and message updates, and stream responses to every surface showing the conversation.
- Serving any harness through one interface means an agent connects with one line, `Channels.forHarness(harness, { channels })`, and no agent-specific bridge.
- Responses go through `Streams`, so they are durable and resumable without a channel-specific stream store. Resume is state-based: on start, Channels watches each known session again, and a run in progress continues its partial message.
## Public API Surface
Additive, exported from `agents/experimental/channels`:
| Symbol | Kind | Notes |
| --- | --- | --- |
| `Channels` | class | Lifecycle capability; `forHarness`, `receive`, `streams`, `websockets` |
| `ChannelsOptions`, `ForHarnessOptions` | types | A harness, its channels, and optional shared `Streams` and `WebSockets` |
| `ChannelsHost`, `ChannelState` | types | What a mounted channel is given, and its per-channel state |
| `ConversationChannel` | interface | What a surface implementation provides |
| `GatewayEvent`, `ConversationUpdate`, `ResponseEnding`, `ResponseReadOptions` | types | Events and updates passed between layers |
## Code Changes
- `conversations.ts` implements `Channels`. It mounts each `ConversationChannel` with a host that offers shared WebSockets, the conversation operations, listing and dispatch, and fans updates out to every channel.
- `harness-conversations.ts` maps the harness to conversations. A conversation is a session with the same id, and a turn is one operation. A run is one response, which steers share. Tool results and approval responses reach the harness as tool answers on the turn they answer, and a refused answer leaves the turn waiting.
- `response.ts` (internal) validates chunk order, so a malformed response fails at the writer, not on a surface.
- An event that names no conversation joins the harness's default session.
- Minor changeset for `agents`.
## pnpm-workspace.yaml (default) ## Dependency Updates | Package | From | To | Type | | --- | --- | --- | --- | | `agents` | 0.26.0 | 0.27.0 | minor | ## Release Notes <details> <summary><b>agents</b> (0.26.0 → 0.27.0)</summary> This release adds four new experimental harnesses and a `web_search` tool, and reorganizes the experimental Channels API. - **New experimental harnesses:** `AiSdkHarness`, `ThinkHarness`, `ContainerHarness` (which runs Claude Code or Codex in a Container) and `OpenCodeHarness`, all with the same shape as `PiHarness`. - **New `agents/websearch`:** a `web_search` tool over Cloudflare's Web Search API for pi, the AI SDK and TanStack AI. - **Channels moved to `agents/experimental/channels`:** Channels is being rebuilt around conversations and turns, and the old `agents/channels` entry point has been removed. The new API includes `ChannelGateway`, the Web Channel and its client, an AI SDK chat transport, and a new `npx agents tui` terminal client. - **Fixes:** Streams, Tasks, `useAgentChat` reconnects, `WebChannelClient`, MCP credential cleanup on `removeServer()`, and x402 with MCP SDK v2. ### Minor Changes - [#2434](cloudflare/agents#2434) [`0ebeae5`](cloudflare/agents@0ebeae5) Thanks [@<!---->cjol](https://github.com/cjol)! - Add `AiSdkHarness` from the new `agents/harness/ai-sdk` entry point, which runs each message with `streamText` and keeps sessions and transcripts in the Durable Object, so `Channels.forHarness` can serve an AI SDK model. The same entry point exports the AI SDK message conversions it is built on and `createSendMessageTool`, an AI SDK tool that sends a message to a surface through the gateway. - [#2431](cloudflare/agents#2431) [`6393265`](cloudflare/agents@6393265) Thanks [@<!---->cjol](https://github.com/cjol)! - Add `Channels` to `agents/experimental/channels`: a Lifecycle capability that serves an agent harness's sessions as conversations to every surface that joins them, with durable, resumable responses on `Streams`. Connect a harness with `Channels.forHarness(harness, { channel …[full notes](https://github.com/cloudflare/agents/releases/tag/agents%400.27.0) </details> --- *This PR was auto-generated by [catalog-update-action](https://github.com/brandhaug/catalog-update-action).* Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
This PR adds
Channels, a Lifecycle capability that serves an agent harness's sessions as conversations to every surface that joins them.Why
Channels.forHarness(harness, { channels }), and no agent-specific bridge.Streams, so they are durable and resumable without a channel-specific stream store. Resume is state-based: on start, Channels watches each known session again, and a run in progress continues its partial message.Public API Surface
Additive, exported from
agents/experimental/channels:ChannelsforHarness,receive,streams,websocketsChannelsOptions,ForHarnessOptionsStreamsandWebSocketsChannelsHost,ChannelStateConversationChannelGatewayEvent,ConversationUpdate,ResponseEnding,ResponseReadOptionsCode Changes
conversations.tsimplementsChannels. It mounts eachConversationChannelwith a host that offers shared WebSockets, the conversation operations, listing and dispatch, and fans updates out to every channel.harness-conversations.tsmaps the harness to conversations. A conversation is a session with the same id, and a turn is one operation. A run is one response, which steers share. Tool results and approval responses reach the harness as tool answers on the turn they answer, and a refused answer leaves the turn waiting.response.ts(internal) validates chunk order, so a malformed response fails at the writer, not on a surface.agents.