feat(think): add per-turn chat metadata

.changeset/think-turn-metadata.md

@@ -0,0 +1,9 @@
+---
+"@cloudflare/think": minor
+---
+
+Add `ChatOptions.metadata` — a per-turn, recovery-safe metadata carrier.
+
+Server-side callers of `chat()` / `chatWithMessengerContext()` can now attach immutable per-turn metadata (e.g. "which authenticated principal initiated this turn"). It is stamped onto the turn's user message alongside `channel` (as `metadata.turnMetadata`) so a recovered/continued turn re-resolves it from durable history, and is readable turn-scoped via the new `Think.activeTurnMetadata` getter.
+
+The trust model mirrors the channel stamp: the reserved metadata keys `channel` and `turnMetadata` are now stripped from client-supplied messages at intake, so a client can never forge server-written turn context (this also closes the prior gap where a client message could carry a forged `metadata.channel`). This lets messenger/RPC entry points carry correct multi-user identity without either mutable agent-wide state (a last-writer-wins race) or the submission path (which loses incremental streaming).

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

@@ -698,6 +698,7 @@ export class ThinkTestAgent extends Think {
   }> = [];
   private _beforeTurnMessagesJson: string[] = [];
   private _capturedTurnChannels: string[] = [];
+  private _capturedTurnMetadata: (Record<string, unknown> | undefined)[] = [];
 
   override configureChannels() {
     return {
@@ -754,13 +755,48 @@ export class ThinkTestAgent extends Think {
     });
     this._beforeTurnMessagesJson.push(JSON.stringify(ctx.messages));
     this._capturedTurnChannels.push(this.activeChannel?.channelId ?? "");
+    this._capturedTurnMetadata.push(this.activeTurnMetadata);
     if (this._turnConfigOverride) return this._turnConfigOverride;
   }
 
   async getCapturedTurnChannelsForTest(): Promise<string[]> {
     return this._capturedTurnChannels;
   }
 
+  async getCapturedTurnMetadataForTest(): Promise<
+    (Record<string, unknown> | undefined)[]
+  > {
+    return this._capturedTurnMetadata;
+  }
+
+  async getActiveTurnMetadataForTest(): Promise<
+    Record<string, unknown> | undefined
+  > {
+    return this.activeTurnMetadata;
+  }
+
+  async runChatTurnForTest(options: {
+    input?: string;
+    channel?: string;
+    metadata?: Record<string, unknown>;
+  }): Promise<void> {
+    await this.chat(options.input ?? "hi", new TestCollectingCallback(), {
+      channel: options.channel,
+      metadata: options.metadata
+    });
+  }
+
+  async persistIncomingMessageForTest(msg: UIMessage): Promise<void> {
+    await (
+      this as unknown as {
+        _persistIncomingMessage(
+          m: UIMessage,
+          serverMessages: readonly UIMessage[]
+        ): Promise<void>;
+      }
+    )._persistIncomingMessage(msg, this.messages);
+  }
+
   async runChannelTurnForTest(options: {
     input?: string;
     channel?: string;

packages/think/src/tests/turn-metadata.test.ts

@@ -0,0 +1,104 @@
+import { env } from "cloudflare:workers";
+import { getServerByName } from "partyserver";
+import { describe, expect, it } from "vitest";
+import type { UIMessage } from "ai";
+import type { ThinkTestAgent } from "./agents/think-session";
+
+async function freshAgent(name: string) {
+  return getServerByName(
+    env.ThinkTestAgent as unknown as DurableObjectNamespace<ThinkTestAgent>,
+    name
+  );
+}
+
+/**
+ * `ChatOptions.metadata`: server-supplied per-turn metadata is stamped on the
+ * turn's user message (alongside `channel`), readable turn-scoped via
+ * `activeTurnMetadata`, re-resolvable from durable history (recovery), and
+ * NEVER forgeable through client-supplied message metadata.
+ */
+describe("turn metadata", () => {
+  it("stamps chat() metadata on the user message and exposes it during the turn", async () => {
+    const agent = await freshAgent(`turn-meta-chat-${crypto.randomUUID()}`);
+
+    await agent.runChatTurnForTest({
+      input: "hello",
+      channel: "voice",
+      metadata: { actingUserId: "user-123", origin: "slack" }
+    });
+
+    // Turn-scoped read: beforeTurn saw the metadata for exactly this turn.
+    const captured = (await agent.getCapturedTurnMetadataForTest()) as (
+      | Record<string, unknown>
+      | undefined
+    )[];
+    expect(captured.at(-1)).toEqual({
+      actingUserId: "user-123",
+      origin: "slack"
+    });
+
+    // Durably stamped alongside the channel on the user message.
+    const messages = (await agent.getMessages()) as UIMessage[];
+    const user = messages.find((m) => m.role === "user");
+    expect(user?.metadata).toMatchObject({
+      channel: "voice",
+      turnMetadata: { actingUserId: "user-123", origin: "slack" }
+    });
+  });
+
+  it("re-resolves turn metadata from durable history (recovery read path)", async () => {
+    const agent = await freshAgent(`turn-meta-recover-${crypto.randomUUID()}`);
+
+    // Simulate a recovered continuation: the stamped user message is all that
+    // survives — no in-memory turn state.
+    await agent.persistTestMessage({
+      id: "u-meta",
+      role: "user",
+      parts: [{ type: "text", text: "recovered turn" }],
+      metadata: {
+        channel: "voice",
+        turnMetadata: { actingUserId: "user-456" }
+      }
+    });
+
+    expect(await agent.getActiveTurnMetadataForTest()).toEqual({
+      actingUserId: "user-456"
+    });
+  });
+
+  it("returns undefined when the latest user message carries none", async () => {
+    const agent = await freshAgent(`turn-meta-none-${crypto.randomUUID()}`);
+    await agent.persistTestMessage({
+      id: "u-plain",
+      role: "user",
+      parts: [{ type: "text", text: "no metadata here" }]
+    });
+    expect(await agent.getActiveTurnMetadataForTest()).toBeUndefined();
+  });
+
+  it("strips reserved metadata keys from client-supplied messages at intake", async () => {
+    const agent = await freshAgent(`turn-meta-forge-${crypto.randomUUID()}`);
+
+    await agent.persistIncomingMessageForTest({
+      id: "u-forged",
+      role: "user",
+      parts: [{ type: "text", text: "I forged my identity" }],
+      metadata: {
+        channel: "voice",
+        turnMetadata: { actingUserId: "victim-user" },
+        harmless: "kept"
+      }
+    } as UIMessage);
+
+    const messages = (await agent.getMessages()) as UIMessage[];
+    const stored = messages.find((m) => m.id === "u-forged");
+    expect(stored).toBeDefined();
+    const metadata = stored?.metadata as Record<string, unknown> | undefined;
+    expect(metadata?.harmless).toBe("kept");
+    expect(metadata && "channel" in metadata).toBe(false);
+    expect(metadata && "turnMetadata" in metadata).toBe(false);
+
+    // And the forged identity is not readable as turn metadata.
+    expect(await agent.getActiveTurnMetadataForTest()).toBeUndefined();
+  });
+});

packages/think/src/think.ts

@@ -1087,6 +1087,20 @@ export interface ChatOptions {
   onClientToolCall?: ClientToolExecutor;
   /** Channel id this turn belongs to. See {@link RunTurnBase.channel}. */
   channel?: string;
+  /**
+   * Server-supplied metadata for this turn. Persisted on the turn's user
+   * message alongside {@link ChatOptions.channel} (as `metadata.turnMetadata`)
+   * so a recovered/continued turn re-resolves it from durable history, and
+   * readable during the turn via {@link Think.activeTurnMetadata}.
+   *
+   * Trust contract mirrors the channel stamp: only server-side callers can set
+   * it — reserved metadata keys on client-supplied messages are stripped at
+   * intake. This gives messenger/RPC entry points (e.g.
+   * {@link Think.chatWithMessengerContext}) a per-turn, recovery-safe carrier
+   * for facts like "which authenticated principal initiated this turn" without
+   * resorting to mutable agent-wide state.
+   */
+  metadata?: Record<string, unknown>;
 }
 
 /** Input accepted by {@link Think.runTurn}. */
@@ -1735,6 +1749,13 @@ type ThinkWorkflowPromptContext = {
 
 const THINK_WORKFLOW_PROMPT_METADATA_KEY = "__thinkWorkflowPrompt";
 
+/**
+ * Message-metadata keys that are server-written turn context (stamped by
+ * `_stampChannel`) and trusted by hooks and recovery. They are stripped from
+ * client-supplied messages at intake so a client can never forge them.
+ */
+const RESERVED_MESSAGE_METADATA_KEYS = ["channel", "turnMetadata"] as const;
+
 /**
  * Reserved name for the synthetic tool a workflow `step.prompt` turn uses to
  * deliver its structured final answer. The agent runs a full multi-step,
@@ -3861,6 +3882,17 @@ export class Think<
     return this._activeChannelContext;
   }
 
+  /**
+   * The server-supplied metadata stamped on the active turn's user message
+   * ({@link ChatOptions.metadata}). Like the channel stamp, it is persisted on
+   * the durable user message, so recovered/continued turns re-resolve the same
+   * value; client-supplied message metadata can never populate it (reserved
+   * keys are stripped at intake). `undefined` when the turn carried none.
+   */
+  get activeTurnMetadata(): Record<string, unknown> | undefined {
+    return this._turnMetadataFromMessages(this.messages);
+  }
+
   /** Resolve a channel id to a turn-scoped {@link ChannelContext}, if registered. */
   private _resolveChannelContext(
     channel: string | undefined
@@ -3923,14 +3955,16 @@ export class Think<
   }
 
   /**
-   * Stamp the channel id onto user messages so a recovered/continued turn can
-   * re-resolve the channel from durable history.
+   * Stamp the channel id — and, when supplied, the caller's turn metadata —
+   * onto user messages so a recovered/continued turn can re-resolve both from
+   * durable history.
    */
   private _stampChannel(
     messages: UIMessage[],
-    channel: string | undefined
+    channel: string | undefined,
+    turnMetadata?: Record<string, unknown>
   ): UIMessage[] {
-    if (!channel) {
+    if (!channel && !turnMetadata) {
       return messages;
     }
     return messages.map((message) =>
@@ -3939,26 +3973,46 @@ export class Think<
             ...message,
             metadata: {
               ...(message.metadata as Record<string, unknown> | undefined),
-              channel
+              ...(channel ? { channel } : {}),
+              ...(turnMetadata ? { turnMetadata } : {})
             }
           }
         : message
     );
   }
 
-  /** The channel stamped on the latest user message in the given list, if any. */
-  private _channelFromMessages(messages: UIMessage[]): string | undefined {
+  /** Metadata of the latest user message in the given list, if any. */
+  private _latestUserMessageMetadata(
+    messages: UIMessage[]
+  ): Record<string, unknown> | undefined {
     for (let i = messages.length - 1; i >= 0; i--) {
       const message = messages[i];
       if (message.role === "user") {
-        const channel = (message.metadata as { channel?: unknown } | undefined)
-          ?.channel;
-        return typeof channel === "string" ? channel : undefined;
+        return message.metadata as Record<string, unknown> | undefined;
       }
     }
     return undefined;
   }
 
+  /** The channel stamped on the latest user message in the given list, if any. */
+  private _channelFromMessages(messages: UIMessage[]): string | undefined {
+    const channel = this._latestUserMessageMetadata(messages)?.channel;
+    return typeof channel === "string" ? channel : undefined;
+  }
+
+  /** The turn metadata stamped on the latest user message, if any. */
+  private _turnMetadataFromMessages(
+    messages: UIMessage[]
+  ): Record<string, unknown> | undefined {
+    const turnMetadata =
+      this._latestUserMessageMetadata(messages)?.turnMetadata;
+    return turnMetadata &&
+      typeof turnMetadata === "object" &&
+      !Array.isArray(turnMetadata)
+      ? (turnMetadata as Record<string, unknown>)
+      : undefined;
+  }
+
   /** Re-resolve the channel for a continuation from the latest user message. */
   private _channelFromLatestUserMessage(): string | undefined {
     return this._channelFromMessages(this.messages);
@@ -6691,7 +6745,11 @@ export class Think<
               ? await userMessage(this.messages)
               : this._normalizeChatMessages(userMessage);
 
-          for (const msg of this._stampChannel(resolved, options?.channel)) {
+          for (const msg of this._stampChannel(
+            resolved,
+            options?.channel,
+            options?.metadata
+          )) {
             await this._appendMessageToHistory(msg);
           }
           this._broadcastMessages();
@@ -12189,11 +12247,35 @@ export class Think<
     msg: UIMessage,
     serverMessages: readonly UIMessage[]
   ): Promise<void> {
+    const sanitized = this._stripReservedMessageMetadata(msg);
     const resolved =
-      msg.role === "assistant" ? resolveToolMergeId(msg, serverMessages) : msg;
+      sanitized.role === "assistant"
+        ? resolveToolMergeId(sanitized, serverMessages)
+        : sanitized;
     await this._upsertMessageInHistory(resolved);
   }
 
+  /**
+   * Strip {@link RESERVED_MESSAGE_METADATA_KEYS} from a client-supplied message
+   * at intake. Those keys are server-written turn context (stamped by
+   * `_stampChannel`) that hooks and recovery re-resolve and trust, so a client
+   * must never be able to forge them.
+   */
+  private _stripReservedMessageMetadata(msg: UIMessage): UIMessage {
+    const metadata = msg.metadata as Record<string, unknown> | undefined;
+    if (
+      !metadata ||
+      !RESERVED_MESSAGE_METADATA_KEYS.some((key) => key in metadata)
+    ) {
+      return msg;
+    }
+    const rest = { ...metadata };
+    for (const key of RESERVED_MESSAGE_METADATA_KEYS) {
+      delete rest[key];
+    }
+    return { ...msg, metadata: rest };
+  }
+
   private _persistClientTools(): void {
     if (this._lastClientTools) {
       this._configSet("lastClientTools", JSON.stringify(this._lastClientTools));