Add Web Channel for arbitrary Durable Objects

.changeset/bright-webs-chat.md

@@ -0,0 +1,12 @@
+---
+"@cloudflare/channels": minor
+---
+
+Add a Web Channel for arbitrary Durable Objects. It speaks the AIChatAgent
+browser protocol using an owned WebSockets capability, dispatches incoming
+turns through `ChannelHost.onMessage`, and streams `ChannelChunk`s back as AI SDK
+UI message chunks.
+
+Push-based Channels can now bind live ingress to the Host's normal routing and
+dispatch path. `consumeChunks` also accepts an `AbortSignal` so transports can
+cancel active generation safely.

.changeset/pink-poems-brake.md

@@ -0,0 +1,30 @@
+---
+"@cloudflare/channels": minor
+---
+
+Stream outbound messages.
+
+`host.stream(surface, chunks, options?)` takes a
+`ReadableStream<ChannelChunk>` and resolves one `DeliveryResult`. Channels that
+can stream consume the stream themselves; Channels that cannot never learn it
+was a stream, because the Host collects the answer and calls `deliver` once.
+
+- New `ChannelChunk` union covering `text`, `reasoning`, `tool`, and `source`,
+  plus an optional `stream` method on `Channel` and `OutboundResolver`.
+- New `consumeChunks` helper for Channel authors, which reads a stream to
+  completion and then finalizes exactly once, whether it closed or errored.
+- Slack streams through `chat.startStream` / `appendStream` / `stopStream`,
+  collecting into an ordinary message for top-level channels where Slack does
+  not support native streaming. Telegram previews with `sendMessageDraft`
+  before persisting the answer with `sendMessage`.
+- `fanout` tees the stream per destination; `fallback` buffers consumed chunks
+  and replays them to the next destination after a failure.
+- `toChannelChunks` maps an AI SDK `fullStream` onto `ChannelChunk`.
+- `DeliveryResult`'s `uncertain` arm gains an optional `reference`, and the
+  three statuses are now defined by what the reader received rather than by
+  what the transport accepted. A stream that ends before its answer is complete
+  is `uncertain`.
+
+Slack reply surfaces derived from ingress now carry `recipientUserId` and
+`recipientTeamId` for non-direct messages, which Slack requires to stream into
+a channel.

.oxfmtrc.json

@@ -5,6 +5,7 @@
   "experimentalSortPackageJson": false,
   "ignorePatterns": [
     "packages/agents/CHANGELOG.md",
+    "packages/channels/live-tests/snapshots",
     "site/agents/.astro",
     "**/routeTree.gen.ts",
     "**/think.d.ts"

packages/channels/README.md

@@ -147,6 +147,47 @@ export default {
 Each Channel authenticates its own input and declines what isn't its business,
 so the Host asks them in configuration order and the first to claim it wins.
 
+### Web chat from any Durable Object
+
+The Web Channel speaks the same browser protocol as `AIChatAgent` without
+requiring the Durable Object to extend `Agent`. It is an ordinary Channel that
+uses an owned WebSockets capability; install that capability into Lifecycle and
+handle its normalized messages through the Host:
+
+```typescript
+import { DurableObject } from "cloudflare:workers";
+import { Lifecycle } from "agents/lifecycle";
+import { ChannelHost, type ChannelInboundMessage } from "@cloudflare/channels";
+import { toChannelChunks } from "@cloudflare/channels/ai-sdk";
+import { web } from "@cloudflare/channels/web";
+import { streamText } from "ai";
+
+export class Chat extends DurableObject<Env> {
+  readonly web = web();
+  readonly channels = new ChannelHost({
+    channels: { web: this.web },
+    onMessage: ({ message }) => this.onMessage(message)
+  });
+  readonly lifecycle = Lifecycle.install(this).use(this.web.webSockets);
+
+  async onMessage(message: ChannelInboundMessage) {
+    const result = streamText({
+      model: this.env.MODEL,
+      prompt: message.message.text
+    });
+    await this.channels.stream(
+      message.replySurface!,
+      toChannelChunks(result.fullStream)
+    );
+  }
+}
+```
+
+The current compatibility layer converts neutral `ChannelChunk`s into AI SDK
+UI message chunks. Live cancellation is supported, while durable stream replay
+is intentionally not yet implemented; reconnecting clients receive an idle
+resume response.
+
 ### Routing
 
 A Channel's `route` turns one normalized event into an opaque application
@@ -297,16 +338,18 @@ Durability is a property of how your application uses it.
 | ----------------------------------------------------------------------- | --------------------------------------------------------------------------- |
 | A `dispatchId` stable across redelivery and unaffected by routing       | Deduplicate on it before starting any side effect                           |
 | The Host awaits your callback before the provider is acknowledged       | Hand off durably before returning — a DO RPC, queue send, or workflow start |
-| One provider attempt per `deliver()`, reported honestly                 | Decide whether to retry; `uncertain` may duplicate a real delivery          |
+| One outbound attempt per `deliver()` or `stream()`, reported honestly   | Decide whether to retry; `uncertain` may duplicate a real delivery          |
 | Surfaces are plain JSON you can persist                                 | Keep configured channel keys stable                                         |
 | Decisions arrive as normalized events carrying your own `interactionId` | Own settlement; an interaction id is not an authorization credential        |
 
 ## Future work
 
 - [ ] Approval-link ingress: signing, verification, and a confirmation page, so
       link approvals return through the same normalized path as Slack buttons
-- [ ] Streaming output delivery — see
-      [`design/rfc-channel-streaming.md`](../../design/rfc-channel-streaming.md)
+- [ ] Reader-initiated stream cancellation: Slack's `message_stream_stopped`
+      and Telegram's `stopped_message_generation` should reach the running
+      generation as ordinary ingress, so aborting it errors the stream and
+      each Channel finalizes on the path it already has
 - [ ] More built-in channels
 - [ ] Rendering templates (pretty emails)
 - [ ] Automatic webhook registration

packages/channels/live-tests/README.md

@@ -1,8 +1,9 @@
 # Live delivery tests
 
-This local-only test calls the real `telegram()`, `slack()`, and `email()`
-adapters, then reads each destination through an independent provider API. It
-is not part of normal package tests, Nx affected tests, or CI.
+These local-only tests call the real `telegram()`, `slack()`, `email()`, and `web()`
+adapters through `ChannelHost`, then read each destination through an independent
+provider API. They are not part of normal package tests, Nx affected tests, or
+CI.
 
 The configured destinations must be disposable. The test deletes their messages
 before and after delivery. Telegram's immutable chat/channel creation service
@@ -22,12 +23,30 @@ Set these variables in an uncommitted environment file or the shell:
   `CHANNELS_LIVE_FASTMAIL_API_TOKEN`,
   `CHANNELS_LIVE_CLOUDFLARE_ACCOUNT_ID`,
   `CHANNELS_LIVE_CLOUDFLARE_API_TOKEN`
+- Web: `CHANNELS_LIVE_WEB_URL` and, for a deployed fixture,
+  `CHANNELS_LIVE_WEB_TOKEN`
 
 Telegram observation uses a non-bot Teleproto `StringSession`. Slack needs
 `chat:write`, `channels:history`, and membership in the configured channel. The
 email sender needs Cloudflare Email Service access; Fastmail supplies independent
 JMAP observation and deletion.
 
+Slack live streaming posts a disposable anchor and streams its thread reply. It
+also needs the reader's user and team ids; the binding derives them from
+`auth.test` and the one channel member that is not this bot, so no extra
+configuration or scope is needed.
+
+Telegram only shows drafts in private chats. Point
+`CHANNELS_LIVE_TELEGRAM_CHAT_ID` at a private chat with the bot to exercise the
+draft path. A group still receives the terminal message, but cannot prove that
+streaming previews reached a reader.
+
+The Web binding observes the destination through a real
+`WebSocketChatTransport`. A separate non-hibernating Cap'n Web session drives
+the fixture Durable Object's `ChannelHost`, keeping each streamed delivery on
+one live object instance. The fixture persists only the reply surface; each test
+uses a fresh object name and clears it afterward.
+
 ## Run
 
 ```sh
@@ -40,6 +59,26 @@ Run one provider with Vitest's name filter:
 pnpm --filter @cloudflare/channels test:live -t telegram
 ```
 
-Each test sends `Cloudflare Channels live delivery smoke test.`, polls once per
-second for up to two minutes, waits five more seconds for duplicates, and
-snapshots the exact observed `[{ text }]` array.
+For a local Web run, start the fixture and test it from separate terminals:
+
+```sh
+pnpm --filter @cloudflare/channels dev:live:web
+```
+
+```sh
+CHANNELS_LIVE_WEB_URL=http://127.0.0.1:8799 \
+  pnpm --filter @cloudflare/channels test:live -t web
+```
+
+To use a deployed fixture, configure its token and deploy the worker, then set
+the matching URL and token in the test environment:
+
+```sh
+pnpm exec wrangler secret put LIVE_TEST_TOKEN \
+  --config packages/channels/live-tests/web/wrangler.jsonc
+pnpm exec wrangler deploy \
+  --config packages/channels/live-tests/web/wrangler.jsonc
+```
+
+The worker accepts tokenless requests only on localhost. Do not commit the
+token or an environment file containing it.

packages/channels/live-tests/binding.ts

@@ -1,11 +1,32 @@
-export type ObservedMessage = { text: string };
+import type { ChannelChunk, DeliveryResult } from "../src/channel";
+import type { ChannelHost } from "../src/host";
+import type { ChannelMessageSurface } from "../src/surface";
+
+export type ObservedMessage = {
+  text: string;
+  [key: string]: unknown;
+};
+
+/** The outbound Host surface exercised by every live delivery scenario. */
+export type LiveDeliveryHost = Pick<ChannelHost, "deliver" | "stream">;
+
+export type LiveStreamSession = {
+  push(chunk: ChannelChunk): Promise<void>;
+  finish(): Promise<DeliveryResult>;
+  /** End the stream the way a failed generation does. */
+  fail(reason: string): Promise<DeliveryResult>;
+};
 
 export type LiveDeliveryBinding = {
-  name: "telegram" | "slack" | "email";
+  name: "telegram" | "slack" | "slack-thread" | "email" | "web";
   destination: string;
-  open?(): Promise<void>;
+  host: LiveDeliveryHost;
+  surface: ChannelMessageSurface;
+  /** Initialize the observer and start from an empty destination. */
+  open(): Promise<void>;
   clear(): Promise<void>;
-  deliver(text: string): Promise<unknown>;
+  /** Provider-side evidence that an ephemeral preview reached the reader. */
+  previews?(): number;
   read(): Promise<ObservedMessage[]>;
   close?(): Promise<void>;
 };

packages/channels/live-tests/bindings/email.ts

@@ -1,9 +1,11 @@
+import { execFileSync } from "node:child_process";
 import {
   email,
   type ChannelEmailMessage,
   type EmailSendBinding
 } from "../../src/adapters/email/email";
 import type { ChannelMessageSurface } from "../../src/surface";
+import { ChannelHost } from "../../src/host";
 import {
   requiredEnv,
   type LiveDeliveryBinding,
@@ -13,6 +15,26 @@ import {
 const CORE = "urn:ietf:params:jmap:core";
 const MAIL = "urn:ietf:params:jmap:mail";
 
+function wranglerAuthToken(): string {
+  try {
+    const output = execFileSync("pnpm", ["exec", "wrangler", "auth", "token"], {
+      encoding: "utf8",
+      stdio: ["ignore", "pipe", "ignore"]
+    });
+    const token = output
+      .split("\n")
+      .map((line) => line.trim())
+      .filter(Boolean)
+      .at(-1);
+    if (!token) throw new Error("Wrangler returned an empty token");
+    return token;
+  } catch {
+    throw new Error(
+      "Could not obtain a Cloudflare API token from Wrangler. Run `pnpm exec wrangler login` before the email live tests."
+    );
+  }
+}
+
 type FastmailSession = { apiUrl: string; accountId: string };
 type FastmailEmail = {
   id: string;
@@ -94,9 +116,10 @@ export function emailBinding(): LiveDeliveryBinding {
     from,
     binding: cloudflareBinding(
       requiredEnv("CHANNELS_LIVE_CLOUDFLARE_ACCOUNT_ID"),
-      requiredEnv("CHANNELS_LIVE_CLOUDFLARE_API_TOKEN")
+      wranglerAuthToken()
     )
   });
+  const host = new ChannelHost({ channels: { email: channel } });
   const surface: ChannelMessageSurface = {
     channelKey: "email",
     version: 1,
@@ -115,6 +138,16 @@ export function emailBinding(): LiveDeliveryBinding {
     ).ids;
   }
 
+  async function clear(): Promise<void> {
+    const emailIds = await ids();
+    if (emailIds.length > 0) {
+      await jmap(fastmailToken, session, "Email/set", {
+        accountId: session.accountId,
+        destroy: emailIds
+      });
+    }
+  }
+
   async function messages(): Promise<FastmailEmail[]> {
     const emailIds = await ids();
     if (emailIds.length === 0) return [];
@@ -135,22 +168,15 @@ export function emailBinding(): LiveDeliveryBinding {
 
   return {
     name: "email",
+    host,
+    surface,
     destination: `email inbox ${to}`,
     async open() {
       session = await openFastmail(fastmailToken);
+      await clear();
     },
-    async clear() {
-      const emailIds = await ids();
-      if (emailIds.length > 0) {
-        await jmap(fastmailToken, session, "Email/set", {
-          accountId: session.accountId,
-          destroy: emailIds
-        });
-      }
-    },
-    async deliver(text) {
-      return await channel.deliver!(surface, { markdown: text });
-    },
+    clear,
+
     async read(): Promise<ObservedMessage[]> {
       return (await messages()).map((message) => {
         const partId = message.textBody[0].partId;