fix(think): run WebSocket chat turns on the implicit web channel (#2255)

.changeset/think-ws-web-channel.md

@@ -0,0 +1,11 @@
+---
+"@cloudflare/think": patch
+---
+
+WebSocket chat turns now run on the implicit `web` channel (#2255).
+
+A `web` entry in `configureChannels()` (its `instructions`, `tools`, and `maxTurns`) now applies to turns sent from `useAgentChat`, and `this.activeChannel` is set during those turns. Previously these turns ran without a channel context, so the `web` policy was silently ignored.
+
+The channel is stamped on each new user message and kept when the client re-sends the transcript, so continuations after client tool results and recovered turns keep the `web` channel. A continuation without a `channel` extends the channel of the turn before it, so regenerating on web continues on web and a continuation run with an explicit `channel` is extended on that channel. A channel in client-sent message metadata is still ignored. `runTurn()` and `chat()` calls without a `channel` are unchanged and run without a channel context.
+
+If you configured a `web` channel, its policy now takes effect for browser chat; check that its `tools` narrowing and `maxTurns` suit that traffic.

docs/think/channels.md

@@ -100,8 +100,21 @@ await this.runTurn({ input: "Read this out loud", channel: "voice" });
 
 Inside a turn, the active channel is available as `this.activeChannel` (a
 `ChannelContext` with `channelId`, `kind`, and messenger details when relevant).
-A turn with no `channel` runs without a channel context and applies no channel
-policy.
+
+Turns from browser clients over WebSocket chat (`useAgentChat`) always run on the
+implicit `web` channel, so a `web` entry in `configureChannels()` applies its
+`instructions`, `tools`, and `maxTurns` to them. The channel is stamped on each
+new user message, so continuations after tool results and recovered turns keep
+it. Clients cannot choose or forge a channel. Regenerating a message another
+channel sent runs on `web` and leaves that message's channel unchanged; a
+continuation after a tool result keeps `web`, but one that runs after the agent
+was evicted takes the stored message's channel. More generally, a continuation
+without a `channel` extends the channel of the turn before it, including a
+channel passed explicitly to `runTurn({ continuation: true, channel })`.
+
+A `runTurn()` or `chat()` call with no `channel` runs without a channel context
+and applies no channel policy. Pass `channel: "web"` to apply the `web` policy to
+such a turn.
 
 ## Deliver out of band
 

packages/think/src/tests/agents/think-session.ts

@@ -847,6 +847,11 @@ export class ThinkTestAgent extends Think {
 
   override configureChannels() {
     return {
+      web: {
+        kind: "web" as const,
+        ingress: { transport: "websocket" as const },
+        instructions: "WEB MODE"
+      },
       voice: {
         kind: "voice" as const,
         ingress: { transport: "voice" as const },
@@ -975,6 +980,40 @@ export class ThinkTestAgent extends Think {
     return this.getMessages();
   }
 
+  /**
+   * Queue continuations, in order, behind a held turn so none of them starts
+   * before the last is admitted.
+   */
+  async runQueuedContinuationsForTest(
+    channels: Array<string | undefined>
+  ): Promise<void> {
+    let release = () => {};
+    const gate = new Promise<void>((resolve) => {
+      release = resolve;
+    });
+    const queue = (
+      this as unknown as {
+        _turnQueue: {
+          enqueue(id: string, fn: () => Promise<void>): Promise<unknown>;
+        };
+      }
+    )._turnQueue;
+    const blocker = queue.enqueue(crypto.randomUUID(), () => gate);
+    const runs: Promise<unknown>[] = [];
+    for (const channel of channels) {
+      runs.push(this.continueLastTurn(undefined, { channel }));
+      await new Promise((resolve) => setTimeout(resolve, 10));
+    }
+    release();
+    await Promise.all([blocker, ...runs]);
+  }
+
+  async getAutoContinuationChannelForTest(): Promise<string | undefined> {
+    return (
+      this as unknown as { _channelForAutoContinuation(): string | undefined }
+    )._channelForAutoContinuation();
+  }
+
   async resetCapturedTurnChannelsForTest(): Promise<void> {
     this._capturedTurnChannels = [];
   }

packages/think/src/tests/ws-web-channel.test.ts

@@ -0,0 +1,214 @@
+/**
+ * Regression coverage for #2255: WebSocket chat turns run on the implicit
+ * `web` channel, so a `web` entry in `configureChannels()` applies to them.
+ */
+
+import { env, exports } from "cloudflare:workers";
+import { describe, expect, it } from "vitest";
+import { getAgentByName } from "agents";
+import type { UIMessage } from "ai";
+import type { ThinkTestAgent } from "./agents/think-session";
+
+const MSG_CHAT_REQUEST = "cf_agent_use_chat_request";
+const MSG_CHAT_RESPONSE = "cf_agent_use_chat_response";
+
+async function freshAgent() {
+  const room = crypto.randomUUID();
+  const agent = await getAgentByName(
+    env.ThinkTestAgent as unknown as DurableObjectNamespace<ThinkTestAgent>,
+    room
+  );
+  const res = await exports.default.fetch(
+    `http://example.com/agents/think-test-agent/${room}`,
+    { headers: { Upgrade: "websocket" } }
+  );
+  expect(res.status).toBe(101);
+  const ws = res.webSocket as WebSocket;
+  ws.accept();
+  return { agent, ws };
+}
+
+function userMessage(
+  text: string,
+  metadata?: Record<string, unknown>
+): UIMessage {
+  return {
+    id: crypto.randomUUID(),
+    role: "user",
+    parts: [{ type: "text", text }],
+    ...(metadata ? { metadata } : {})
+  };
+}
+
+/** Sends a chat request and resolves once its terminal `done` frame arrives. */
+function sendChat(
+  ws: WebSocket,
+  messages: UIMessage[],
+  trigger?: string
+): Promise<void> {
+  const id = crypto.randomUUID();
+  const done = new Promise<void>((resolve, reject) => {
+    const timer = setTimeout(
+      () => reject(new Error("Timeout waiting for done")),
+      10_000
+    );
+    const handler = (e: MessageEvent) => {
+      const msg = JSON.parse(e.data as string) as Record<string, unknown>;
+      if (msg.type === MSG_CHAT_RESPONSE && msg.id === id && msg.done) {
+        clearTimeout(timer);
+        ws.removeEventListener("message", handler);
+        resolve();
+      }
+    };
+    ws.addEventListener("message", handler);
+  });
+  ws.send(
+    JSON.stringify({
+      type: MSG_CHAT_REQUEST,
+      id,
+      init: { method: "POST", body: JSON.stringify({ messages, trigger }) }
+    })
+  );
+  return done;
+}
+
+function channelOf(message: UIMessage | undefined): unknown {
+  return (message?.metadata as { channel?: unknown } | undefined)?.channel;
+}
+
+describe("Think — WebSocket chat runs on the web channel (#2255)", () => {
+  it("applies the web channel policy and exposes activeChannel", async () => {
+    const { agent, ws } = await freshAgent();
+
+    await sendChat(ws, [userMessage("hello")]);
+
+    expect(await agent.getCapturedTurnChannelsForTest()).toEqual(["web"]);
+    const log = await agent.getBeforeTurnLog();
+    expect(log.at(-1)?.system).toContain("WEB MODE");
+    expect(log.at(-1)?.system).not.toContain("VOICE MODE");
+    ws.close();
+  });
+
+  it("stamps the web channel on the user message", async () => {
+    const { agent, ws } = await freshAgent();
+
+    await sendChat(ws, [userMessage("hello")]);
+
+    const messages = (await agent.getStoredMessages()) as UIMessage[];
+    expect(channelOf(messages.find((m) => m.role === "user"))).toBe("web");
+    ws.close();
+  });
+
+  it("keeps the stamp when the client re-sends the transcript", async () => {
+    const { agent, ws } = await freshAgent();
+    const first = userMessage("first");
+
+    await sendChat(ws, [first]);
+    const afterFirst = (await agent.getStoredMessages()) as UIMessage[];
+    const second = userMessage("second");
+    // A browser client echoes messages back without server-owned metadata.
+    await sendChat(ws, [
+      first,
+      ...afterFirst.filter((m) => m.role === "assistant"),
+      second
+    ]);
+
+    const messages = (await agent.getStoredMessages()) as UIMessage[];
+    const users = messages.filter((m) => m.role === "user");
+    expect(users.map((m) => m.id)).toEqual([first.id, second.id]);
+    expect(users.map(channelOf)).toEqual(["web", "web"]);
+    expect(await agent.getCapturedTurnChannelsForTest()).toEqual([
+      "web",
+      "web"
+    ]);
+    ws.close();
+  });
+
+  it("ignores a channel forged in client-sent metadata", async () => {
+    const { agent, ws } = await freshAgent();
+
+    await sendChat(ws, [userMessage("hello", { channel: "voice" })]);
+
+    expect(await agent.getCapturedTurnChannelsForTest()).toEqual(["web"]);
+    const log = await agent.getBeforeTurnLog();
+    expect(log.at(-1)?.system).not.toContain("VOICE MODE");
+    const messages = (await agent.getStoredMessages()) as UIMessage[];
+    expect(channelOf(messages.find((m) => m.role === "user"))).toBe("web");
+    ws.close();
+  });
+
+  it("re-resolves the web channel when the turn is continued", async () => {
+    const { agent, ws } = await freshAgent();
+
+    await sendChat(ws, [userMessage("hello")]);
+    await agent.resetCapturedTurnChannelsForTest();
+    await agent.runChannelTurnForTest({ continuation: true });
+
+    expect(await agent.getCapturedTurnChannelsForTest()).toEqual(["web"]);
+    ws.close();
+  });
+
+  it("continues a regeneration of another channel's message on web", async () => {
+    const { agent, ws } = await freshAgent();
+    await agent.runChannelTurnForTest({ input: "hello", channel: "voice" });
+    const stored = (await agent.getStoredMessages()) as UIMessage[];
+    const voiceUser = stored.find((m) => m.role === "user");
+    expect(channelOf(voiceUser)).toBe("voice");
+    await agent.resetCapturedTurnChannelsForTest();
+
+    await sendChat(
+      ws,
+      [{ ...voiceUser!, metadata: undefined }],
+      "regenerate-message"
+    );
+
+    expect(await agent.getCapturedTurnChannelsForTest()).toEqual(["web"]);
+    // The stored message keeps the channel it was sent on.
+    const messages = (await agent.getStoredMessages()) as UIMessage[];
+    expect(channelOf(messages.find((m) => m.id === voiceUser!.id))).toBe(
+      "voice"
+    );
+    // A tool-result continuation extends the web turn, not the voice one.
+    expect(await agent.getAutoContinuationChannelForTest()).toBe("web");
+
+    await agent.resetCapturedTurnChannelsForTest();
+    await agent.runChannelTurnForTest({ continuation: true });
+    expect(await agent.getCapturedTurnChannelsForTest()).toEqual(["web"]);
+    ws.close();
+  });
+
+  it("extends the channel an explicit-channel continuation ran on", async () => {
+    const { agent, ws } = await freshAgent();
+    await sendChat(ws, [userMessage("hello")]);
+
+    await agent.runChannelTurnForTest({ continuation: true, channel: "voice" });
+
+    expect(await agent.getAutoContinuationChannelForTest()).toBe("voice");
+    ws.close();
+  });
+
+  it("resolves a queued continuation's channel when it runs, not when it is admitted", async () => {
+    const { agent, ws } = await freshAgent();
+    await sendChat(ws, [userMessage("hello")]);
+    await agent.resetCapturedTurnChannelsForTest();
+
+    await agent.runQueuedContinuationsForTest(["voice", undefined]);
+
+    expect(await agent.getCapturedTurnChannelsForTest()).toEqual([
+      "voice",
+      "voice"
+    ]);
+    ws.close();
+  });
+
+  it("still runs server-driven turns without a channel by default", async () => {
+    const { agent, ws } = await freshAgent();
+
+    await agent.runChannelTurnForTest({ input: "hello" });
+
+    expect(await agent.getCapturedTurnChannelsForTest()).toEqual([""]);
+    const log = await agent.getBeforeTurnLog();
+    expect(log.at(-1)?.system).not.toContain("WEB MODE");
+    ws.close();
+  });
+});