fix(anthropic): stage thinking signatures until block stop (#104043)

src/agents/anthropic-transport-stream.test.ts

@@ -25,13 +25,55 @@ type AnthropicStreamOptions = Parameters<AnthropicStreamFn>[2];
 type RequestTransportConfig = Parameters<typeof attachModelProviderRequestTransport>[1];
 
 function createSseResponse(events: Record<string, unknown>[] = []): Response {
-  const body = events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join("");
+  const body = serializeSseEvents(events);
   return new Response(body, {
     status: 200,
     headers: { "content-type": "text/event-stream" },
   });
 }
 
+function serializeSseEvents(events: Record<string, unknown>[]): string {
+  return events.map((event) => `data: ${JSON.stringify(event)}\n\n`).join("");
+}
+
+function createFailingSseResponse(events: Record<string, unknown>[], error: Error): Response {
+  const encoder = new TextEncoder();
+  let sentEvents = false;
+  const body = new ReadableStream<Uint8Array>({
+    pull(controller) {
+      if (!sentEvents) {
+        sentEvents = true;
+        controller.enqueue(encoder.encode(serializeSseEvents(events)));
+        return;
+      }
+      controller.error(error);
+    },
+  });
+  return new Response(body, {
+    status: 200,
+    headers: { "content-type": "text/event-stream" },
+  });
+}
+
+function createInterruptedThinkingEvents(): Record<string, unknown>[] {
+  return [
+    {
+      type: "message_start",
+      message: { id: "msg_1", usage: { input_tokens: 6, output_tokens: 0 } },
+    },
+    {
+      type: "content_block_start",
+      index: 0,
+      content_block: { type: "thinking", thinking: "step by step", signature: "" },
+    },
+    {
+      type: "content_block_delta",
+      index: 0,
+      delta: { type: "signature_delta", signature: "partial-signature" },
+    },
+  ];
+}
+
 function createStalledSseResponse(params: { onCancel: (reason: unknown) => void }): Response {
   const encoder = new TextEncoder();
   const body = new ReadableStream<Uint8Array>({
@@ -1941,6 +1983,124 @@ describe("anthropic transport stream", () => {
     });
   });
 
+  it.each([
+    {
+      label: "the stream ends before content_block_stop",
+      response: () => createSseResponse(createInterruptedThinkingEvents()),
+      stopReason: "stop",
+    },
+    {
+      label: "the provider errors before content_block_stop",
+      response: () =>
+        createSseResponse([
+          ...createInterruptedThinkingEvents(),
+          { type: "error", error: { message: "provider failed" } },
+        ]),
+      stopReason: "error",
+    },
+    {
+      label: "the response body fails",
+      response: () =>
+        createFailingSseResponse(
+          createInterruptedThinkingEvents(),
+          new Error("response body failed"),
+        ),
+      stopReason: "error",
+    },
+  ])("does not persist signature deltas when $label", async ({ response, stopReason }) => {
+    guardedFetchMock.mockResolvedValueOnce(response());
+
+    const result = await runTransportStream(
+      makeAnthropicTransportModel(),
+      { messages: [{ role: "user", content: "think" }] } as AnthropicStreamContext,
+      { apiKey: "sk-ant-api" } as AnthropicStreamOptions,
+    );
+
+    expect(result.stopReason).toBe(stopReason);
+    expect(result.content[0]).toMatchObject({
+      type: "thinking",
+      thinking: "step by step",
+      thinkingSignature: "",
+    });
+  });
+
+  it("does not persist signature deltas when the request aborts", async () => {
+    const controller = new AbortController();
+    guardedFetchMock.mockResolvedValueOnce(
+      createOpenRawSseResponse({
+        body: serializeSseEvents(createInterruptedThinkingEvents()),
+        onCancel: () => undefined,
+      }),
+    );
+    setTimeout(() => controller.abort(new Error("request aborted")), 20);
+
+    const result = await runTransportStream(
+      makeAnthropicTransportModel(),
+      { messages: [{ role: "user", content: "think" }] } as AnthropicStreamContext,
+      {
+        apiKey: "sk-ant-api",
+        signal: controller.signal,
+      } as AnthropicStreamOptions,
+    );
+
+    expect(result.stopReason).toBe("aborted");
+    expect(result.content[0]).toMatchObject({
+      type: "thinking",
+      thinkingSignature: "",
+    });
+  });
+
+  it("commits only stopped signatures across interleaved thinking blocks", async () => {
+    guardedFetchMock.mockResolvedValueOnce(
+      createSseResponse([
+        {
+          type: "message_start",
+          message: { id: "msg_1", usage: { input_tokens: 6, output_tokens: 0 } },
+        },
+        {
+          type: "content_block_start",
+          index: 0,
+          content_block: { type: "thinking", thinking: "first", signature: "" },
+        },
+        {
+          type: "content_block_start",
+          index: 1,
+          content_block: { type: "thinking", thinking: "second", signature: "" },
+        },
+        {
+          type: "content_block_delta",
+          index: 1,
+          delta: { type: "signature_delta", signature: "complete-second" },
+        },
+        {
+          type: "content_block_delta",
+          index: 0,
+          delta: { type: "signature_delta", signature: "partial-first" },
+        },
+        { type: "content_block_stop", index: 1 },
+      ]),
+    );
+
+    const result = await runTransportStream(
+      makeAnthropicTransportModel(),
+      { messages: [{ role: "user", content: "think" }] } as AnthropicStreamContext,
+      { apiKey: "sk-ant-api" } as AnthropicStreamOptions,
+    );
+
+    expect(result.content).toEqual([
+      expect.objectContaining({
+        type: "thinking",
+        thinking: "first",
+        thinkingSignature: "",
+      }),
+      expect.objectContaining({
+        type: "thinking",
+        thinking: "second",
+        thinkingSignature: "complete-second",
+      }),
+    ]);
+  });
+
   it("captures OpenAI-style reasoning_content deltas from Anthropic-compatible streams", async () => {
     guardedFetchMock.mockResolvedValueOnce(
       createSseResponse([

src/agents/anthropic-transport-stream.ts

@@ -1256,7 +1256,9 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn {
         );
         const blocks = output.content;
         const blockIndexes = new Map<number, number>();
-        const signatureDeltaIndexes = new Set<number>();
+        // Signature deltas are opaque and only complete at content_block_stop.
+        // Keep partial bytes out of output so interrupted streams cannot poison replay.
+        const pendingThinkingSignatures = new Map<number, string>();
         const allowReasoningContentReplay = supportsReasoningContentReplay(model);
         const reasoningContentThinkingBlocks = new Map<number, number>();
         const reasoningContentTextBlocks = new Map<number, number>();
@@ -1453,6 +1455,7 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn {
               refusalBuffer?.discard();
               pendingTextEnds.length = 0;
               blockIndexes.clear();
+              pendingThinkingSignatures.clear();
               applyAnthropicFallbackBoundary({
                 output,
                 boundary: fallbackBoundary,
@@ -1492,6 +1495,7 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn {
               }
               continue;
             }
+            pendingThinkingSignatures.delete(index);
             if (contentBlock?.type === "text") {
               const text =
                 typeof contentBlock.text === "string"
@@ -1695,16 +1699,23 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn {
               typeof delta.signature === "string"
             ) {
               const signatureIndex = eventIndexKey(event.index);
-              if (!signatureDeltaIndexes.has(signatureIndex)) {
-                signatureDeltaIndexes.add(signatureIndex);
+              const pendingSignature = pendingThinkingSignatures.get(signatureIndex);
+              if (pendingSignature === undefined) {
                 block.thinkingSignature = "";
+                pendingThinkingSignatures.set(signatureIndex, delta.signature);
+              } else {
+                pendingThinkingSignatures.set(signatureIndex, pendingSignature + delta.signature);
               }
-              block.thinkingSignature = (block.thinkingSignature || "") + delta.signature;
             }
             continue;
           }
           if (event.type === "content_block_stop") {
             const eventIndex = typeof event.index === "number" ? event.index : undefined;
+            const pendingSignature =
+              eventIndex === undefined ? undefined : pendingThinkingSignatures.get(eventIndex);
+            if (eventIndex !== undefined) {
+              pendingThinkingSignatures.delete(eventIndex);
+            }
             const index = eventIndex === undefined ? undefined : blockIndexes.get(eventIndex);
             const block = index === undefined ? undefined : blocks[index];
             if (eventIndex === undefined || index === undefined || !block) {
@@ -1724,6 +1735,9 @@ export function createAnthropicMessagesTransportStreamFn(): StreamFn {
               continue;
             }
             if (block.type === "thinking") {
+              if (pendingSignature !== undefined) {
+                block.thinkingSignature = pendingSignature;
+              }
               eventSink.push({
                 type: "thinking_end",
                 contentIndex: index,