fix(voice): honor pcm16 audio sample rate

.changeset/tidy-rethink-channels.md

@@ -0,0 +1,5 @@
+---
+"@cloudflare/rethink": minor
+---
+
+Add primitive host and channel primitives for composing WebSocket and Email transports on one Durable Object.

packages/rethink/CONTEXT.md

@@ -0,0 +1,49 @@
+# rethink
+
+Composition model for building Durable Object agents from narrow primitives
+instead of god base classes (`Agent`, `Think`).
+
+## Language
+
+**Primitive**:
+A plain `(ctx, deps)` object that may expose optional DO-shaped methods. The unit
+of composition on a Durable Object.
+_Avoid_: plugin, module, mixin
+
+**PrimitiveHost**:
+The thin, dispatch-only DO base class that fans shared entrypoints to the
+author's primitives. Carries no domain behavior or state.
+_Avoid_: Agent base, Think base, god class
+
+**ChannelIn**:
+A role a Primitive may implement: transport ingress. Accepts inbound events,
+claims them on shared entrypoints, and fans normalized messages to registered
+listeners.
+_Avoid_: messenger, ingress adapter (as the type name)
+
+**ChannelOut**:
+A role a Primitive may implement: transport egress. Opens a progressive stream
+to an explicit delivery target.
+_Avoid_: messenger, delivery surface (as the type name)
+
+**Channel**:
+Informal name for a Primitive that implements ChannelIn, ChannelOut, or both.
+Not a type by itself.
+_Avoid_: Channel as a single required dual-direction type
+
+**InboundMessage**:
+A generic transport-neutral envelope produced by a ChannelIn for listeners:
+identity, body, optional attachments, optional reply handle, and optional typed
+raw payload. Transport-specific delivery surface details live in the typed reply
+handle or raw payload.
+_Avoid_: event, payload (without saying which)
+
+**ChannelOut target**:
+A channel-owned, explicit delivery destination for ChannelOut. Often derived
+from an inbound reply handle, but never required to come from one.
+_Avoid_: central OutTarget union, reply context (implies inbound-only)
+
+**OutStream**:
+The handle returned by ChannelOut.openStream: write AI SDK UIMessageChunks,
+complete, interrupt, error.
+_Avoid_: send (as the primary egress API)

packages/rethink/README.md

@@ -0,0 +1,7 @@
+# @cloudflare/rethink
+
+Clean-room package for composable Think primitives.
+
+This package is intentionally small while the new API is designed. Add each
+piece of functionality as an independently usable primitive, then compose those
+primitives into higher-level agent behavior outside the primitive itself.

packages/rethink/package.json

@@ -0,0 +1,58 @@
+{
+  "name": "@cloudflare/rethink",
+  "version": "0.0.0",
+  "description": "Composable primitives for building Cloudflare Think-style agents",
+  "keywords": [
+    "cloudflare",
+    "agents",
+    "think",
+    "ai",
+    "sdk"
+  ],
+  "type": "module",
+  "repository": {
+    "directory": "packages/rethink",
+    "type": "git",
+    "url": "git+https://github.com/cloudflare/agents.git"
+  },
+  "bugs": {
+    "url": "https://github.com/cloudflare/agents/issues"
+  },
+  "exports": {
+    ".": {
+      "types": "./dist/index.d.ts",
+      "import": "./dist/index.js",
+      "require": "./dist/index.js"
+    }
+  },
+  "author": "Cloudflare Inc.",
+  "license": "MIT",
+  "files": [
+    "dist",
+    "README.md"
+  ],
+  "dependencies": {
+    "agents": "workspace:*",
+    "ai": "^6.0.202"
+  },
+  "scripts": {
+    "build": "tsx ./scripts/build.ts",
+    "types:test": "wrangler types tests/env.d.ts --config tests/wrangler.jsonc --include-runtime false && oxfmt --write tests/env.d.ts",
+    "test": "pnpm run test:unit && pnpm run test:workers",
+    "test:unit": "vitest --run --dir src",
+    "test:workers": "vitest --run -c tests/vitest.config.ts"
+  },
+  "nx": {
+    "targets": {
+      "build": {
+        "inputs": [
+          "production",
+          "^production"
+        ],
+        "outputs": [
+          "{projectRoot}/dist"
+        ]
+      }
+    }
+  }
+}

packages/rethink/scripts/build.ts

@@ -0,0 +1,26 @@
+import { build } from "tsdown";
+import { formatDeclarationFiles } from "../../../scripts/format-declarations";
+
+async function main() {
+  await build({
+    clean: true,
+    dts: true,
+    entry: ["src/index.ts"],
+    deps: {
+      skipNodeModulesBundle: true,
+      neverBundle: ["cloudflare:workers"]
+    },
+    format: "esm",
+    sourcemap: true,
+    fixedExtension: false
+  });
+
+  formatDeclarationFiles();
+
+  process.exit(0);
+}
+
+main().catch((err) => {
+  console.error(err);
+  process.exit(1);
+});

packages/rethink/src/channels/directory.ts

@@ -0,0 +1,57 @@
+import type { UIMessageChunk } from "ai";
+import type {
+  ChannelIn,
+  ChannelOut,
+  InboundMessage,
+  MessageHandler,
+  OutStream
+} from "./types";
+
+type Reply = (chunks: UIMessageChunk[]) => Promise<void>;
+
+type ListenAllHandler = (
+  msg: InboundMessage,
+  reply: Reply
+) => void | Promise<void>;
+
+/** Registry that connects ChannelIn listeners to ChannelOut reply streams. */
+export class ChannelDirectory {
+  #outputs = new Map<string, ChannelOut>();
+
+  constructor(outputs: ChannelOut[]) {
+    for (const output of outputs) this.#outputs.set(output.channelId, output);
+  }
+
+  listen<TRaw, TReplyTo>(
+    input: ChannelIn<TRaw, TReplyTo>,
+    handler: MessageHandler<TRaw, TReplyTo>
+  ): () => void {
+    return input.onMessage(handler);
+  }
+
+  listenAll(inputs: ChannelIn[], handler: ListenAllHandler): () => void {
+    const unsubscribers = inputs.map((input) =>
+      input.onMessage((msg) =>
+        handler(msg, (chunks) => this.reply(msg, chunks))
+      )
+    );
+    return () => {
+      for (const unsubscribe of unsubscribers) unsubscribe();
+    };
+  }
+
+  open(channelId: string, target: unknown): OutStream | undefined {
+    return this.#outputs.get(channelId)?.openStream(target);
+  }
+
+  async reply<TReplyTo>(
+    msg: InboundMessage<unknown, TReplyTo>,
+    chunks: UIMessageChunk[]
+  ): Promise<void> {
+    if (msg.replyTo === undefined) return;
+    const stream = this.open(msg.channelId, msg.replyTo);
+    if (!stream) return;
+    for (const chunk of chunks) await stream.write(chunk);
+    await stream.complete();
+  }
+}

packages/rethink/src/channels/email.ts

@@ -0,0 +1,110 @@
+import type { EmailSendBinding } from "agents";
+import type { UIMessageChunk } from "ai";
+import type { InboundEmail, Primitive } from "../primitives";
+import type {
+  ChannelIn,
+  ChannelOut,
+  InboundMessage,
+  MessageHandler,
+  OutStream
+} from "./types";
+import { textFromChunk } from "./utils";
+
+/** Raw email details preserved on inbound messages for email-specific consumers. */
+export interface EmailRaw {
+  from: string;
+  to: string;
+  subject: string;
+  messageId?: string;
+}
+
+/** Serializable target for replying over email. */
+export interface EmailTarget {
+  to: string;
+  subject?: string;
+  inReplyTo?: string;
+}
+
+export interface EmailChannelDeps {
+  binding: EmailSendBinding;
+  from: string;
+}
+
+/** Channel primitive that turns forwarded email into messages and replies. */
+export class EmailChannel
+  implements
+    Primitive,
+    ChannelIn<EmailRaw, EmailTarget>,
+    ChannelOut<EmailTarget>
+{
+  readonly channelId: string;
+  #handler?: MessageHandler<EmailRaw, EmailTarget>;
+
+  constructor(
+    private _ctx: DurableObjectState,
+    private deps: EmailChannelDeps,
+    options: { channelId?: string } = {}
+  ) {
+    this.channelId = options.channelId ?? "email";
+  }
+
+  onMessage(handler: MessageHandler<EmailRaw, EmailTarget>): () => void {
+    this.#handler = handler;
+    return () => {
+      if (this.#handler === handler) this.#handler = undefined;
+    };
+  }
+
+  async onEmail(msg: InboundEmail): Promise<boolean> {
+    await this.emit({
+      channelId: this.channelId,
+      from: msg.from,
+      body: msg.body,
+      replyTo: {
+        to: msg.from,
+        subject: `re: ${msg.subject}`,
+        inReplyTo: msg.messageId
+      },
+      raw: {
+        from: msg.from,
+        to: msg.to,
+        subject: msg.subject,
+        messageId: msg.messageId
+      }
+    });
+    return true;
+  }
+
+  openStream(target: EmailTarget): OutStream {
+    const chunks: UIMessageChunk[] = [];
+    const deps = this.deps;
+    return {
+      write(chunk) {
+        chunks.push(chunk);
+      },
+      async complete() {
+        await deps.binding.send({
+          from: deps.from,
+          to: target.to,
+          subject: target.subject ?? "message",
+          text: chunks.map(textFromChunk).join(""),
+          headers: target.inReplyTo
+            ? { "In-Reply-To": target.inReplyTo }
+            : undefined
+        });
+      },
+      interrupt() {
+        return;
+      },
+      error() {
+        return;
+      }
+    };
+  }
+
+  private async emit(
+    msg: InboundMessage<EmailRaw, EmailTarget>
+  ): Promise<void> {
+    await this.#handler?.(msg);
+  }
+}

packages/rethink/src/channels/index.ts

@@ -0,0 +1,18 @@
+export { ChannelDirectory } from "./directory";
+export { EmailChannel } from "./email";
+export { WebSocketChannel } from "./websocket";
+export type {
+  ChannelIn,
+  ChannelOut,
+  InboundMessage,
+  MessageHandler,
+  OutStream,
+  UIMessageChunk
+} from "./types";
+export type { EmailChannelDeps, EmailRaw, EmailTarget } from "./email";
+export type {
+  WebSocketChatRequestFrame,
+  WebSocketChatResponseFrame,
+  WebSocketRaw,
+  WebSocketTarget
+} from "./websocket";

packages/rethink/src/channels/types.ts

@@ -0,0 +1,37 @@
+import type { UIMessageChunk } from "ai";
+
+/** A transport-neutral inbound envelope emitted by ChannelIn primitives. */
+export interface InboundMessage<TRaw = unknown, TReplyTo = unknown> {
+  channelId: string;
+  from: string;
+  body: string;
+  replyTo?: TReplyTo;
+  raw?: TRaw;
+}
+
+/** Progressive outbound writer returned by ChannelOut primitives. */
+export interface OutStream {
+  write(chunk: UIMessageChunk): void | Promise<void>;
+  complete(): void | Promise<void>;
+  interrupt(): void | Promise<void>;
+  error(err: unknown): void | Promise<void>;
+}
+
+/** Listener registered by consumers that want inbound channel messages. */
+export type MessageHandler<TRaw = unknown, TReplyTo = unknown> = (
+  msg: InboundMessage<TRaw, TReplyTo>
+) => void | Promise<void>;
+
+/** Ingress role for primitives that emit transport-normalized messages. */
+export interface ChannelIn<TRaw = unknown, TReplyTo = unknown> {
+  readonly channelId: string;
+  onMessage(handler: MessageHandler<TRaw, TReplyTo>): () => void;
+}
+
+/** Egress role for primitives that deliver chunks to explicit targets. */
+export interface ChannelOut<TTarget = unknown> {
+  readonly channelId: string;
+  openStream(target: TTarget): OutStream;
+}
+
+export type { UIMessageChunk } from "ai";

packages/rethink/src/channels/utils.ts

@@ -0,0 +1,12 @@
+import type { UIMessageChunk } from "ai";
+
+export function textFromChunk(chunk: UIMessageChunk): string {
+  if (chunk.type === "text-delta" && typeof chunk.delta === "string") {
+    return chunk.delta;
+  }
+  return "";
+}
+
+export function errorText(err: unknown): string {
+  return err instanceof Error ? err.message : String(err);
+}