[Bug]: AnthropicChatModel streaming merges all `thinking` blocks of one response into a single block, keeping only the last `signature`

[BlueX888]

Prerequisites

• I have searched the existing issues and discussions, and this is not a duplicate.
• This is a bug, not a usage question. (For questions, please use Discussions instead.)

Background / Description

AnthropicChatModel._parse_anthropic_stream_completion_response allocates one thinking_id and one text_id for the whole response (src/agentscope/model/_anthropic/_model.py:415-416) and routes every thinking_delta / signature_delta / text_delta to that single block id regardless of event.index (lines 487, 493, 501). Only tool_use blocks are tracked per index.

When a response contains more than one thinking block, streaming therefore:

1. concatenates all thinking text into one ThinkingBlock,
2. overwrites signature on every signature_delta, so only the last block's signature survives (the first one is lost),
3. loses the block order relative to text / tool_use blocks.

With stream=False the same response is parsed correctly (_parse_anthropic_completion_response keeps one ThinkingBlock per content block), and the Anthropic SDK's own accumulator (anthropic.lib.streaming._messages.accumulate_event) keys blocks by index and keeps them separate. So stream and non-stream produce different message histories for the same model output.

This is not an exotic response shape. Anthropic's docs describe multiple thinking blocks in a single response as the normal case with interleaved / adaptive thinking (thinking_mode="adaptive" in AnthropicChatModel.Parameters enables it automatically), and give a concrete example of a response whose content is thinking(sig A) → thinking(sig B) → tool_use (progress updates: "Each one comes back as its own thinking block with its own signature"). The docs also say thinking blocks "must be passed back complete and unmodified" and that "modified thinking blocks are rejected with a 400 error" (preserving thinking blocks).

Consequence in a ReAct loop: the merged block (thinking = A + B, signature = sig_B) is stored in memory and sent back with the tool result on the next call. Since sig_B was issued for block B alone and block A's signature is gone, the request no longer matches what the API returned. #2135 reported exactly this symptom (thinking or redacted_thinking blocks in the latest assistant message cannot be modified, at content index 10, intermittent); #2139 fixed the redacted_thinking part of it, but the single-thinking_id merging is still present on main.

The text_id has the same problem (e.g. text → tool_use → text collapses to one text block placed before the tool call), but the thinking case is the one that breaks the next request.

Error Messages

# Verbatim output of the repro script below (main @ 61cdeae4, no network).
# The SDK accumulator and stream=False keep 3 blocks; stream=True keeps 2
# and sig_A is gone.

== anthropic SDK accumulate_event ==
[
  {
    "signature": "sig_A",
    "thinking": "REASONING ",
    "type": "thinking"
  },
  {
    "signature": "sig_B",
    "thinking": "Editing auth.py to add the refresh call.",
    "type": "thinking"
  },
  {
    "input": {
      "path": "auth.py"
    },
    "name": "edit_file",
    "type": "tool_use"
  }
]

== agentscope stream=False ==
[
  {
    "type": "thinking",
    "thinking": "REASONING ",
    "signature": "sig_A"
  },
  {
    "type": "thinking",
    "thinking": "Editing auth.py to add the refresh call.",
    "signature": "sig_B"
  },
  {
    "type": "tool_call",
    "name": "edit_file",
    "input": "{\"path\": \"auth.py\"}"
  }
]

== agentscope stream=True (final is_last chunk) ==
[
  {
    "type": "thinking",
    "thinking": "REASONING Editing auth.py to add the refresh call.",
    "signature": "sig_B"
  },
  {
    "type": "tool_call",
    "name": "edit_file",
    "input": "{\"path\": \"auth.py\"}"
  }
]

== replayed assistant message (stream=True history) ==
[
  {
    "role": "assistant",
    "content": [
      {
        "type": "thinking",
        "thinking": "REASONING Editing auth.py to add the refresh call.",
        "signature": "sig_B"
      },
      {
        "type": "tool_use",
        "id": "toolu_1",
        "name": "edit_file",
        "input": {
          "path": "auth.py"
        }
      }
    ]
  }
]

Expected: the stream=True result equals the stream=False result (three blocks, sig_A and sig_B each on their own ThinkingBlock, in the original order).

Steps to Reproduce

1. Code (repro_anthropic_stream_thinking.py). It feeds the same event sequence — built from the real anthropic.types event classes, in the shape documented for progress updates — to (a) the Anthropic SDK's accumulator, (b) AnthropicChatModel(stream=False), (c) AnthropicChatModel(stream=True), then formats (c) back with AnthropicChatFormatter. No API key or network needed.

import asyncio
import json
from unittest.mock import AsyncMock, MagicMock

from anthropic.types import (
    InputJSONDelta, Message, MessageDeltaUsage, RawContentBlockDeltaEvent,
    RawContentBlockStartEvent, RawContentBlockStopEvent, RawMessageDeltaEvent,
    RawMessageStartEvent, RawMessageStopEvent, SignatureDelta, ThinkingBlock,
    ThinkingDelta, ToolUseBlock, Usage,
)
from anthropic.types.raw_message_delta_event import Delta
from anthropic.lib.streaming._messages import accumulate_event

from agentscope.credential import AnthropicCredential
from agentscope.formatter import AnthropicChatFormatter
from agentscope.message import UserMsg, AssistantMsg
from agentscope.model import AnthropicChatModel

USAGE = Usage(input_tokens=10, output_tokens=0)

# thinking(sig_A) -> thinking(sig_B) -> tool_use, as documented for progress updates
EVENTS = [
    RawMessageStartEvent(type="message_start", message=Message(
        id="msg_1", type="message", role="assistant", model="claude-opus-4-6",
        content=[], stop_reason=None, stop_sequence=None, usage=USAGE)),
    RawContentBlockStartEvent(type="content_block_start", index=0,
        content_block=ThinkingBlock(type="thinking", thinking="", signature="")),
    RawContentBlockDeltaEvent(type="content_block_delta", index=0,
        delta=ThinkingDelta(type="thinking_delta", thinking="REASONING ")),
    RawContentBlockDeltaEvent(type="content_block_delta", index=0,
        delta=SignatureDelta(type="signature_delta", signature="sig_A")),
    RawContentBlockStopEvent(type="content_block_stop", index=0),
    RawContentBlockStartEvent(type="content_block_start", index=1,
        content_block=ThinkingBlock(type="thinking", thinking="", signature="")),
    RawContentBlockDeltaEvent(type="content_block_delta", index=1,
        delta=ThinkingDelta(type="thinking_delta",
                            thinking="Editing auth.py to add the refresh call.")),
    RawContentBlockDeltaEvent(type="content_block_delta", index=1,
        delta=SignatureDelta(type="signature_delta", signature="sig_B")),
    RawContentBlockStopEvent(type="content_block_stop", index=1),
    RawContentBlockStartEvent(type="content_block_start", index=2,
        content_block=ToolUseBlock(type="tool_use", id="toolu_1", name="edit_file", input={})),
    RawContentBlockDeltaEvent(type="content_block_delta", index=2,
        delta=InputJSONDelta(type="input_json_delta", partial_json='{"path": "auth.py"}')),
    RawContentBlockStopEvent(type="content_block_stop", index=2),
    RawMessageDeltaEvent(type="message_delta",
        delta=Delta(stop_reason="tool_use", stop_sequence=None),
        usage=MessageDeltaUsage(output_tokens=30)),
    RawMessageStopEvent(type="message_stop"),
]


class _FakeStream:
    def __init__(self, events):
        self._events = events

    async def __aenter__(self):
        return self

    async def __aexit__(self, *exc):
        return False

    async def __aiter__(self):
        for e in self._events:
            yield e


def _dump(blocks):
    out = []
    for b in blocks:
        d = {"type": b.type}
        if b.type == "thinking":
            d["thinking"], d["signature"] = b.thinking, getattr(b, "signature", None)
        elif b.type == "text":
            d["text"] = b.text
        elif b.type == "tool_call":
            d["name"], d["input"] = b.name, b.input
        out.append(d)
    return out


async def main():
    snapshot = None
    for e in EVENTS:
        snapshot = accumulate_event(event=e, current_snapshot=snapshot)
    print("== anthropic SDK accumulate_event ==")
    print(json.dumps([
        {k: v for k, v in b.model_dump().items()
         if k in ("type", "thinking", "signature", "name", "input")}
        for b in snapshot.content], indent=2))

    model_ns = AnthropicChatModel(
        credential=AnthropicCredential(api_key="fake"), model="claude-opus-4-6", stream=False)
    model_ns.client = MagicMock()
    model_ns.client.messages.create = AsyncMock(return_value=Message(
        id="msg_1", type="message", role="assistant", model="claude-opus-4-6",
        content=snapshot.content, stop_reason="tool_use", stop_sequence=None, usage=USAGE))
    res_ns = await model_ns([UserMsg(name="user", content="fix the login test")])
    print("\n== agentscope stream=False ==")
    print(json.dumps(_dump(res_ns.content), indent=2))

    model_s = AnthropicChatModel(
        credential=AnthropicCredential(api_key="fake"), model="claude-opus-4-6", stream=True)
    model_s.client = MagicMock()
    model_s.client.messages.create = AsyncMock(return_value=_FakeStream(EVENTS))
    final = None
    async for chunk in await model_s([UserMsg(name="user", content="fix the login test")]):
        if chunk.is_last:
            final = chunk
    print("\n== agentscope stream=True (final is_last chunk) ==")
    print(json.dumps(_dump(final.content), indent=2))

    replay = await AnthropicChatFormatter().format(
        [AssistantMsg(name="assistant", content=final.content)])
    print("\n== replayed assistant message (stream=True history) ==")
    print(json.dumps(replay, indent=2))


asyncio.run(main())

2. Run: python repro_anthropic_stream_thinking.py
3. Compare the stream=True section with the other two (see Error Messages above).

Root cause: src/agentscope/model/_anthropic/_model.py:415-416 (text_id / thinking_id created once per response) and :487, :493, :501 (deltas keyed by those ids instead of event.index). content_block_start at :455-468 already does the right thing for tool_use via tool_call_mapping[event.index].

I'm happy to send a PR: allocate a fresh block id on every content_block_start (keyed by event.index, same as tool_call_mapping) and look it up for text_delta / thinking_delta / signature_delta. I have this working locally; the repro then matches stream=False, and tests/model_anthropic_test.py + tests/model_base_test.py pass unchanged. Would add a streaming test with two thinking blocks.

Environment

• AgentScope Version: 2.0.7.post1 (main @ 61cdeae)
• Python Version: 3.12.14
• OS: macOS 26.2 (arm64)
• anthropic SDK: 1.3.0