feat(think): configurable messenger concurrency strategy (#2313)

.changeset/think-messenger-concurrency.md

@@ -0,0 +1,5 @@
+---
+"@cloudflare/think": minor
+---
+
+Add `messengerConcurrency` to choose the Chat SDK concurrency strategy for messenger replies (#2313). It defaults to the existing 600 ms `burst` strategy, exported as `DEFAULT_MESSENGER_CONCURRENCY`, and accepts any `ConcurrencyStrategy` or `ConcurrencyConfig`, such as `"queue"` or `{ strategy: "burst", debounceMs: 1500 }`.

docs/think/index.md

@@ -403,6 +403,7 @@ path.
 | `getMessengers()`          | `{}`                             | Messenger ingress and delivery declarations — see [Messengers](./messengers.md)                                                                                                                                                                                                                                                                                                                                                                                                                                             |
 | `getActions()`             | `{}`                             | Server actions (idempotency, approvals, authorization) compiled into tools — see [Actions](./actions.md)                                                                                                                                                                                                                                                                                                                                                                                                                    |
 | `configureChannels()`      | `{}`                             | Per-channel policy and surfaces beyond the implicit `web` channel — see [Channels](./channels.md)                                                                                                                                                                                                                                                                                                                                                                                                                           |
+| `messengerConcurrency`     | Burst, 600 ms                    | Chat SDK concurrency strategy for every messenger on the agent. Set it as a class field — see [Messengers](./messengers.md#change-the-concurrency-strategy)                                                                                                                                                                                                                                                                                                                                                                 |
 | `maxSteps`                 | `10`                             | Max tool-call rounds per turn                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
 | `sendReasoning`            | `true`                           | Send reasoning chunks to chat clients                                                                                                                                                                                                                                                                                                                                                                                                                                                                                       |
 | `configureSession()`       | identity                         | Configure the default session handle: compaction and search — see [Sessions](https://github.com/cloudflare/agents/blob/main/docs/agents/sessions.md)                                                                                                                                                                                                                                                                                                                                                                        |

docs/think/messengers.md

@@ -162,6 +162,30 @@ In a subscribed thread, a burst counts as a mention when any of its messages
 mentions the bot. With the default `respondTo`, a mention followed by an
 ordinary line is still answered.
 
+### Change the Concurrency Strategy
+
+Set `messengerConcurrency` to use another Chat SDK concurrency strategy. It
+defaults to `DEFAULT_MESSENGER_CONCURRENCY`, which is
+`{ strategy: "burst", debounceMs: 600 }`. One setting covers every messenger on
+the agent, and it must be a class field, because the runtime reads it before
+`onStart` runs.
+
+```ts
+import { Think } from "@cloudflare/think";
+
+export class SupportAgent extends Think<Env> {
+  // Answer the first message right away. Messages that arrive during the
+  // reply are answered together once it finishes.
+  messengerConcurrency = "queue" as const;
+}
+```
+
+Use `{ strategy: "burst", debounceMs: 1500 }` to wait longer for a burst. The
+`"drop"` strategy discards a message that arrives while a reply is running.
+`"concurrent"` hands every message to Think without a thread lock, so none are
+batched or dropped. Think still runs the turns of one conversation one at a
+time, so the replies arrive in order rather than in parallel.
+
 ## Conversation Targets
 
 The default conversation mode is one Think sub-agent per Chat SDK thread. This

packages/think/src/messengers/chat-sdk.ts

@@ -5,6 +5,8 @@ import type {
   Attachment as ChatAttachment,
   Author as ChatAuthor,
   ChatConfig,
+  ConcurrencyConfig,
+  ConcurrencyStrategy,
   Message as ChatMessage,
   SerializedThread,
   Thread as ChatThread
@@ -111,6 +113,14 @@ export interface MessengerDefinition {
 
 export type ThinkMessengers = Record<string, MessengerDefinition>;
 
+/** The Chat SDK `concurrency` setting for a Think agent's messengers. */
+export type MessengerConcurrency = ConcurrencyStrategy | ConcurrencyConfig;
+
+export const DEFAULT_MESSENGER_CONCURRENCY: MessengerConcurrency = {
+  debounceMs: 600,
+  strategy: "burst"
+};
+
 export interface NormalizedMessengerDefinition extends MessengerDefinition {
   id: string;
   path: string;
@@ -229,10 +239,14 @@ export class ThinkMessengerRuntime {
   >();
   private readonly definitions: NormalizedMessengerDefinition[];
 
+  private readonly concurrency: MessengerConcurrency;
+
   constructor(
     definitions: ThinkMessengers,
-    private readonly host: MessengerThinkHost
+    private readonly host: MessengerThinkHost,
+    options?: { concurrency?: MessengerConcurrency }
   ) {
+    this.concurrency = options?.concurrency ?? DEFAULT_MESSENGER_CONCURRENCY;
     this.definitions = normalizeMessengers(definitions);
     for (const definition of this.definitions) {
       this.definitionsByAdapterName.set(definition.adapterName, definition);
@@ -368,7 +382,7 @@ export class ThinkMessengerRuntime {
     ) as Record<string, Adapter>;
     const chat = new Chat({
       adapters,
-      concurrency: { debounceMs: 600, strategy: "burst" },
+      concurrency: this.concurrency,
       fallbackStreamingPlaceholderText: FALLBACK_STREAMING_PLACEHOLDER_TEXT,
       state: createChatSdkState({
         agent: ThinkMessengerStateAgent,

packages/think/src/messengers/index.ts

@@ -1,5 +1,6 @@
 export {
   chatSdkMessenger,
+  DEFAULT_MESSENGER_CONCURRENCY,
   defaultChatSdkEvent,
   defaultConversationName,
   idempotencyKeyForEvent,
@@ -16,6 +17,7 @@ export {
 export type {
   ChatSdkMessengerEventInput,
   ChatSdkMessengerOptions,
+  MessengerConcurrency,
   MessengerConversationMode,
   MessengerConversationResolver,
   MessengerConversationTarget,

packages/think/src/tests/agents/messengers.ts

@@ -219,6 +219,8 @@ export class ThinkMessengerDeliveryTestAgent extends Think {
   }
 
   override getMessengers(): ThinkMessengers {
+    // #2313: an agent named `queue-…` answers each message as it arrives.
+    if (this.name.startsWith("queue-")) this.messengerConcurrency = "queue";
     return {
       fake: chatSdkMessenger({
         adapter: this._recordingAdapter(),

packages/think/src/tests/messengers.test.ts

@@ -1312,6 +1312,23 @@ describe("think messengers core", () => {
       expect(await agent.getRecorded("post")).toEqual(["Got"]);
     });
 
+    it("answers the first message at once under a configured queue strategy (#2313)", async () => {
+      const agent = await sendBurst("queue-dm", [
+        { id: "q1", text: "summarize the thread", threadId: "fake:dm-queue" },
+        { id: "q2", text: "for me", threadId: "fake:dm-queue" },
+        { id: "q3", text: "and keep it short", threadId: "fake:dm-queue" }
+      ]);
+
+      const prompts = await agent.getRecorded("prompt");
+      expect(prompts[0]).toBe("summarize the thread");
+      expect(prompts.flatMap((prompt) => prompt.split("\n")).sort()).toEqual([
+        "and keep it short",
+        "for me",
+        "summarize the thread"
+      ]);
+      expect(await agent.getRecorded("post")).toHaveLength(prompts.length);
+    });
+
     it("keeps each sender's label when a group burst mixes senders", async () => {
       const bob = { fullName: "Bob", userId: "user-bob" };
       const ada = { fullName: "Ada", userId: "user-ada" };

packages/think/src/think.ts

@@ -479,9 +479,13 @@ import type { CreateFetchToolsOptions, FetchToolEvent } from "./tools/fetch";
 import { truncatePausedExecutionOutput } from "./tools/execute";
 import { ExtensionManager, sanitizeName } from "./extensions/manager";
 import { ThinkMessengerRuntime } from "./messengers/chat-sdk";
-import { MESSENGER_REPLY_FIBER_NAME } from "./messengers";
+import {
+  DEFAULT_MESSENGER_CONCURRENCY,
+  MESSENGER_REPLY_FIBER_NAME
+} from "./messengers";
 import type {
   DeliveryKind,
+  MessengerConcurrency,
   MessengerContext,
   MessengerDeliverySurface,
   ThinkMessengers,
@@ -3289,6 +3293,17 @@ export class Think<
    */
   messageConcurrency: MessageConcurrency = "queue";
 
+  /**
+   * How the Chat SDK handles messages that arrive in a messenger thread while
+   * the agent is still answering, or before it starts. Accepts any Chat SDK
+   * `concurrency` value: `"queue"`, `"debounce"`, `"burst"`, `"concurrent"`,
+   * `"drop"`, or a `ConcurrencyConfig`. One setting covers every messenger on
+   * the agent. Set it as a class field: it is read before `onStart` runs.
+   *
+   * @default { strategy: "burst", debounceMs: 600 }
+   */
+  messengerConcurrency: MessengerConcurrency = DEFAULT_MESSENGER_CONCURRENCY;
+
   /**
    * Byte budget for hydrating the persisted transcript into the in-memory
    * message cache (`this.messages`).
@@ -5298,7 +5313,8 @@ export class Think<
 
     this._messengerRuntime = new ThinkMessengerRuntime(
       messengerDefs,
-      this as unknown as MessengerThinkHost
+      this as unknown as MessengerThinkHost,
+      { concurrency: this.messengerConcurrency }
     );
     this._messengerRuntime.initialize();
   }