Stop provider tool-call replays from regressing tool part state (#1404) (#1412)

Author: threepointone

.changeset/agents-isreplay-chunk-helper.md

@@ -0,0 +1,15 @@
+---
+"agents": patch
+---
+
+Make `applyChunkToParts` idempotent against an existing tool part with the same `toolCallId`, and add `isReplayChunk(parts, chunk)` for stream broadcasters that want to drop provider replay chunks ([#1404](https://github.com/cloudflare/agents/issues/1404)).
+
+Some providers (notably the OpenAI Responses API) re-emit a prior tool call in continuation streams. The previous `tool-input-start` handler unconditionally pushed a fresh tool part, which produced duplicate parts in the message; `tool-input-delta` and `tool-input-available` overwrote a fully resolved input/state if a chunk happened to arrive for an already-known toolCallId. The new behavior:
+
+- `tool-input-start` for a `toolCallId` that already exists in `parts` is a no-op (it does not push a duplicate or regress state).
+- `tool-input-delta` only mutates input while the existing part is still `input-streaming`.
+- `tool-input-available` only advances from `input-streaming` to `input-available`; replays against parts that have already moved past `input-streaming` (including `approval-requested`/`approval-responded` and any terminal state) are no-ops.
+
+`isReplayChunk(parts, chunk)` is exported from `agents/chat` for stream broadcasters (e.g. `AIChatAgent._streamSSEReply`) that want to detect "this chunk is a replay of an already-known tool call" and skip re-broadcasting it. AI SDK v6's `updateToolPart` on the client mutates an existing tool part in place when the toolCallId matches, so re-broadcasting these replay chunks would visibly regress an `output-available` part to `input-streaming` on connected clients. `tool-output-available` is _not_ treated as a replay because its in-place update is safe when the output already matches.
+
+Tool calls that the model genuinely wants to re-issue always carry a new toolCallId, so an existing match is never a legitimate "start over".

.changeset/ai-chat-tool-replay-regression.md

@@ -0,0 +1,14 @@
+---
+"@cloudflare/ai-chat": patch
+---
+
+Stop provider tool-call replays from regressing tool part state during continuation streams ([#1404](https://github.com/cloudflare/agents/issues/1404)).
+
+Some providers (notably the OpenAI Responses API) re-emit prior tool calls in continuation streams as a `tool-input-start` → `tool-input-delta` → `tool-input-available` → `tool-output-available` sequence carrying the _same_ `toolCallId` and the _same_ `output` the part already holds. The AI SDK's `updateToolPart` mutates an existing tool part in place when the toolCallId matches, so a replayed `tool-input-start` was clobbering an `output-available` part back to `input-streaming` on the client and producing the worker warn `_applyToolResult: Tool part with toolCallId X not in expected state`.
+
+Two fixes:
+
+- `_streamSSEReply` now drops replay tool-input chunks before broadcasting them to clients or storing them for resume, using the new shared `isReplayChunk` helper. The cloned server-side streaming message is never corrupted because `applyChunkToParts` is idempotent against existing toolCallIds for these chunk types (also fixed below).
+- `_applyToolResult` accepts `output-available` and `output-error` as valid starting states for _idempotent_ re-application. A duplicate `cf_agent_tool_result` (cross-tab re-run, redelivered WS frame, provider replay round-trip) is now a silent no-op rather than a warn + skipped update. The cross-message `tool-output-available`/`tool-output-error` fallback in `_streamSSEReply` gets the same tolerance.
+
+`_findAndUpdateToolPart` skips the SQLite write and `MESSAGE_UPDATED` broadcast when the apply produced no semantic change, so idempotent re-applies don't churn UI on connected tabs.

packages/agents/src/chat/index.ts

@@ -1,5 +1,6 @@
 export {
   applyChunkToParts,
+  isReplayChunk,
   type MessageParts,
   type MessagePart,
   type StreamChunkData

packages/agents/src/chat/message-builder.ts

@@ -182,6 +182,22 @@ export function applyChunkToParts(
     }
 
     case "tool-input-start": {
+      // Idempotent against an existing tool part with the same toolCallId.
+      // Some providers (notably the OpenAI Responses API) replay prior
+      // tool calls in continuation streams as a fresh `tool-input-start`
+      // → `tool-input-delta` → `tool-input-available` →
+      // `tool-output-available` sequence carrying the original toolCallId
+      // and original output. Without this guard a replay would push a
+      // duplicate part into the streaming message *and* clobber the
+      // original part's state when the AI SDK's mutate-in-place
+      // `updateToolPart` processes the replay on the client (issue #1404).
+      // A model that genuinely wants a fresh tool call always emits a
+      // new toolCallId, so an existing match is never a legitimate
+      // "start over".
+      const existing = findToolPartByCallId(parts, chunk.toolCallId);
+      if (existing) {
+        return true;
+      }
       parts.push({
         type: `tool-${chunk.toolName}`,
         toolCallId: chunk.toolCallId,
@@ -200,8 +216,15 @@ export function applyChunkToParts(
     }
 
     case "tool-input-delta": {
+      // Only mutate input while the tool is still actively input-streaming.
+      // Deltas arriving after the tool has already advanced (input-available
+      // or any terminal state) are provider replay and must not regress
+      // a fully-formed input back to a partial one.
       const toolPart = findToolPartByCallId(parts, chunk.toolCallId);
-      if (toolPart) {
+      if (
+        toolPart &&
+        (toolPart as Record<string, unknown>).state === "input-streaming"
+      ) {
         (toolPart as Record<string, unknown>).input = chunk.input;
       }
       return true;
@@ -211,40 +234,59 @@ export function applyChunkToParts(
       const existing = findToolPartByCallId(parts, chunk.toolCallId);
       if (existing) {
         const p = existing as Record<string, unknown>;
-        p.state = "input-available";
-        p.input = chunk.input;
-        if (chunk.providerExecuted != null) {
-          p.providerExecuted = chunk.providerExecuted;
-        }
-        if (chunk.providerMetadata != null) {
-          p.callProviderMetadata = chunk.providerMetadata;
-        }
-        if (chunk.title != null) {
-          p.title = chunk.title;
+        // Only advance from the streaming-input phase. Once the tool is
+        // already at input-available or any terminal state
+        // (output-available, output-error, output-denied,
+        // approval-requested, approval-responded), this chunk is a
+        // provider replay and must not regress state or overwrite a
+        // resolved input/output. See the comment on tool-input-start.
+        if (p.state === "input-streaming") {
+          p.state = "input-available";
+          p.input = chunk.input;
+          if (chunk.providerExecuted != null) {
+            p.providerExecuted = chunk.providerExecuted;
+          }
+          if (chunk.providerMetadata != null) {
+            p.callProviderMetadata = chunk.providerMetadata;
+          }
+          if (chunk.title != null) {
+            p.title = chunk.title;
+          }
         }
-      } else {
-        parts.push({
-          type: `tool-${chunk.toolName}`,
-          toolCallId: chunk.toolCallId,
-          toolName: chunk.toolName,
-          state: "input-available",
-          input: chunk.input,
-          ...(chunk.providerExecuted != null
-            ? { providerExecuted: chunk.providerExecuted }
-            : {}),
-          ...(chunk.providerMetadata != null
-            ? { callProviderMetadata: chunk.providerMetadata }
-            : {}),
-          ...(chunk.title != null ? { title: chunk.title } : {})
-        } as MessagePart);
+        return true;
       }
+      parts.push({
+        type: `tool-${chunk.toolName}`,
+        toolCallId: chunk.toolCallId,
+        toolName: chunk.toolName,
+        state: "input-available",
+        input: chunk.input,
+        ...(chunk.providerExecuted != null
+          ? { providerExecuted: chunk.providerExecuted }
+          : {}),
+        ...(chunk.providerMetadata != null
+          ? { callProviderMetadata: chunk.providerMetadata }
+          : {}),
+        ...(chunk.title != null ? { title: chunk.title } : {})
+      } as MessagePart);
       return true;
     }
 
     case "tool-input-error": {
       const existing = findToolPartByCallId(parts, chunk.toolCallId);
       if (existing) {
         const p = existing as Record<string, unknown>;
+        // First-write-wins: a tool that's already terminal must not be
+        // regressed (or re-decided as an error) by a later chunk. A
+        // tool-input-error here is either provider replay or a confused
+        // upstream — preserve the existing terminal state.
+        if (
+          p.state === "output-available" ||
+          p.state === "output-error" ||
+          p.state === "output-denied"
+        ) {
+          return true;
+        }
         p.state = "output-error";
         p.errorText = chunk.errorText;
         p.input = chunk.input;
@@ -362,6 +404,48 @@ export function applyChunkToParts(
   }
 }
 
+/**
+ * Returns true if `chunk` would be a no-op replay against the already-known
+ * `parts` — i.e. some upstream is re-emitting events for a tool call that
+ * the message has already advanced past.
+ *
+ * Used by stream broadcasters to suppress re-broadcasting these chunks to
+ * connected clients. AI SDK v6's `updateToolPart` mutates an existing tool
+ * part in place when a chunk arrives with a matching `toolCallId`, so a
+ * replayed `tool-input-start` would clobber an `output-available` part back
+ * to `input-streaming` on the client (issue #1404).
+ *
+ * Only returns true when re-broadcasting would *visibly regress* state on
+ * a v6 client. Safe-by-construction chunk types (e.g. `tool-output-available`
+ * carrying the same output the part already has) return false.
+ *
+ * Conditions:
+ * - `tool-input-start` for a `toolCallId` that already exists in `parts`.
+ * - `tool-input-delta` for a `toolCallId` whose existing part is no longer
+ *   `input-streaming`.
+ * - `tool-input-available` for a `toolCallId` whose existing part is no
+ *   longer `input-streaming` (i.e. has already advanced to `input-available`
+ *   or any terminal state).
+ */
+export function isReplayChunk(
+  parts: MessagePart[],
+  chunk: StreamChunkData
+): boolean {
+  if (
+    chunk.type !== "tool-input-start" &&
+    chunk.type !== "tool-input-delta" &&
+    chunk.type !== "tool-input-available"
+  ) {
+    return false;
+  }
+  if (!chunk.toolCallId) return false;
+  const existing = findToolPartByCallId(parts, chunk.toolCallId);
+  if (!existing) return false;
+  if (chunk.type === "tool-input-start") return true;
+  const state = (existing as Record<string, unknown>).state;
+  return state !== "input-streaming";
+}
+
 /**
  * Finds the last part in the array matching the given type.
  * Searches from the end for efficiency (the part we want is usually recent).