diff --git a/src/anthropic/lib/__init__.py b/src/anthropic/lib/__init__.py index 076480d7..f78b48cc 100644 --- a/src/anthropic/lib/__init__.py +++ b/src/anthropic/lib/__init__.py @@ -1 +1,2 @@ from ._files import files_from_dir as files_from_dir, async_files_from_dir as async_files_from_dir +from ._thinking import strip_thinking_blocks as strip_thinking_blocks diff --git a/src/anthropic/lib/_thinking.py b/src/anthropic/lib/_thinking.py new file mode 100644 index 00000000..f6b32c7d --- /dev/null +++ b/src/anthropic/lib/_thinking.py @@ -0,0 +1,82 @@ +"""Utilities for managing extended thinking blocks in conversation history. + +Extended thinking blocks carry a ``signature`` field that is valid only for +the specific conversation context they were created in. Long multi-turn +sessions or any operation that reconstructs history from stored messages can +hit a 400 "Invalid ``signature`` in ``thinking`` block" error because the +server-side context that produced the signature no longer matches. +``strip_thinking_blocks`` removes those blocks before the history is replayed, +providing the only documented recovery path. +""" + +from __future__ import annotations + +from typing import Iterable + +from ..types.message_param import MessageParam + +__all__ = ["strip_thinking_blocks"] + +_THINKING_TYPES = frozenset({"thinking", "redacted_thinking"}) + + +def _block_type(block: object) -> str: + """Return the ``type`` field of a content block regardless of representation. + + Accepts plain dicts (TypedDict values, JSON-decoded payloads) and Pydantic + ``BaseModel`` instances (``ThinkingBlock``, ``RedactedThinkingBlock``, …). + Returns ``""`` for anything that carries no ``type``. + """ + if isinstance(block, dict): + return block.get("type", "") + return getattr(block, "type", "") + + +def strip_thinking_blocks( + messages: Iterable[MessageParam], +) -> list[MessageParam]: + """Return a copy of *messages* with every thinking and redacted-thinking block removed. + + Use this when replaying conversation history fails with + ``"Invalid `signature` in `thinking` block"``: the ``signature`` stored in + a ``ThinkingBlock`` is bound to the exact server-side conversation context + it was created in and is not guaranteed to remain valid after session + expiry, model upgrades, or history reconstruction from storage. + + Only the ``content`` list is modified; string content and all non-thinking + blocks are preserved verbatim. The ``role`` and any other keys on each + message dict are copied through unchanged. + + Example:: + + try: + response = client.messages.create(model=model, messages=history, ...) + except anthropic.BadRequestError as exc: + if "Invalid `signature` in `thinking` block" in str(exc): + history = strip_thinking_blocks(history) + response = client.messages.create(model=model, messages=history, ...) + else: + raise + + Args: + messages: The conversation history to clean. Each entry must be a + ``MessageParam``-compatible mapping with at least ``role`` and + ``content`` keys. + + Returns: + A new ``list[MessageParam]`` in the same order, with thinking and + redacted-thinking content blocks omitted from every message. + """ + result: list[MessageParam] = [] + for msg in messages: + content = msg.get("content") # type: ignore[attr-defined] + if isinstance(content, str): + # Plain-text content carries no thinking blocks. + result.append(dict(msg)) # type: ignore[arg-type] + continue + filtered: list[object] = [ + block for block in content # type: ignore[union-attr] + if _block_type(block) not in _THINKING_TYPES + ] + result.append({**msg, "content": filtered}) # type: ignore[typeddict-item] + return result diff --git a/tests/lib/test_thinking.py b/tests/lib/test_thinking.py new file mode 100644 index 00000000..425aca42 --- /dev/null +++ b/tests/lib/test_thinking.py @@ -0,0 +1,194 @@ +"""Tests for strip_thinking_blocks (:mod:`anthropic.lib._thinking`). + +Extended thinking block signatures are tied to the conversation context they +were created in and may become invalid when replaying stored history. +``strip_thinking_blocks`` removes those blocks so the history can be re-sent +without a 400 error. +""" + +from __future__ import annotations + +from anthropic.lib._thinking import strip_thinking_blocks + +# --------------------------------------------------------------------------- +# Helpers — build content blocks as plain dicts (the most common representation +# callers have) and as minimal object stubs (mimicking BaseModel responses). +# --------------------------------------------------------------------------- + + +def _thinking(text: str = "thinking…") -> dict: + return {"type": "thinking", "thinking": text, "signature": "sig_abc"} + + +def _redacted() -> dict: + return {"type": "redacted_thinking", "data": "opaque"} + + +def _text(text: str = "hello") -> dict: + return {"type": "text", "text": text} + + +def _tool_use() -> dict: + return {"type": "tool_use", "id": "tu_1", "name": "my_tool", "input": {}} + + +class _FakeThinkingBlock: + """Minimal stub that looks like a ThinkingBlock BaseModel instance.""" + + type = "thinking" + thinking = "deep thought" + signature = "sig_xyz" + + +class _FakeRedactedBlock: + type = "redacted_thinking" + data = "bytes" + + +class _FakeTextBlock: + type = "text" + text = "plain text" + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +def test_string_content_passes_through_unchanged() -> None: + messages = [{"role": "user", "content": "hello"}] + result = strip_thinking_blocks(messages) + assert result == [{"role": "user", "content": "hello"}] + + +def test_thinking_block_removed_from_assistant() -> None: + messages = [ + { + "role": "assistant", + "content": [_thinking(), _text("answer")], + } + ] + result = strip_thinking_blocks(messages) + assert len(result[0]["content"]) == 1 # type: ignore[arg-type] + assert result[0]["content"][0]["type"] == "text" # type: ignore[index] + + +def test_redacted_thinking_block_removed() -> None: + messages = [ + { + "role": "assistant", + "content": [_redacted(), _text("answer")], + } + ] + result = strip_thinking_blocks(messages) + assert len(result[0]["content"]) == 1 # type: ignore[arg-type] + assert result[0]["content"][0]["type"] == "text" # type: ignore[index] + + +def test_mixed_content_only_removes_thinking_types() -> None: + messages = [ + { + "role": "assistant", + "content": [_thinking(), _text("A"), _tool_use(), _redacted(), _text("B")], + } + ] + result = strip_thinking_blocks(messages) + content = result[0]["content"] # type: ignore[index] + assert len(content) == 3 # type: ignore[arg-type] + types = [b["type"] for b in content] # type: ignore[index] + assert types == ["text", "tool_use", "text"] + + +def test_user_messages_without_thinking_unchanged() -> None: + messages = [ + {"role": "user", "content": [_text("what?")]}, + ] + result = strip_thinking_blocks(messages) + assert result[0]["content"] == [_text("what?")] # type: ignore[index] + + +def test_non_thinking_blocks_preserved_verbatim() -> None: + blocks = [_text("one"), _tool_use(), _text("two")] + messages = [{"role": "assistant", "content": blocks}] + result = strip_thinking_blocks(messages) + assert result[0]["content"] == blocks # type: ignore[index] + + +def test_all_thinking_produces_empty_content_list() -> None: + # Edge case: a message whose entire content is thinking blocks. + messages = [{"role": "assistant", "content": [_thinking(), _redacted()]}] + result = strip_thinking_blocks(messages) + assert result[0]["content"] == [] # type: ignore[index] + + +def test_empty_messages_list() -> None: + assert strip_thinking_blocks([]) == [] + + +def test_multiple_messages_processed_independently() -> None: + messages = [ + {"role": "user", "content": "first"}, + {"role": "assistant", "content": [_thinking(), _text("first answer")]}, + {"role": "user", "content": "second"}, + {"role": "assistant", "content": [_thinking(), _text("second answer")]}, + ] + result = strip_thinking_blocks(messages) + assert result[0]["content"] == "first" + assert result[1]["content"] == [_text("first answer")] # type: ignore[index] + assert result[2]["content"] == "second" + assert result[3]["content"] == [_text("second answer")] # type: ignore[index] + + +def test_basemodel_instances_stripped() -> None: + # When blocks are BaseModel instances (as returned by the SDK), the type + # attribute is accessed via getattr rather than dict lookup. + messages = [ + { + "role": "assistant", + "content": [_FakeThinkingBlock(), _FakeTextBlock()], + } + ] + result = strip_thinking_blocks(messages) + content = result[0]["content"] # type: ignore[index] + assert len(content) == 1 # type: ignore[arg-type] + assert content[0].type == "text" # type: ignore[index] + + +def test_redacted_basemodel_instance_stripped() -> None: + messages = [ + { + "role": "assistant", + "content": [_FakeRedactedBlock(), _FakeTextBlock()], + } + ] + result = strip_thinking_blocks(messages) + assert len(result[0]["content"]) == 1 # type: ignore[arg-type] + + +def test_does_not_mutate_original_messages() -> None: + original_content = [_thinking(), _text("answer")] + messages = [{"role": "assistant", "content": original_content}] + strip_thinking_blocks(messages) + # The original list must be unchanged. + assert len(original_content) == 2 + assert original_content[0]["type"] == "thinking" + + +def test_role_and_other_keys_preserved() -> None: + messages = [ + { + "role": "assistant", + "content": [_thinking(), _text("hi")], + } + ] + result = strip_thinking_blocks(messages) + assert result[0]["role"] == "assistant" + + +def test_idempotent() -> None: + messages = [ + {"role": "assistant", "content": [_thinking(), _text("hi")]}, + ] + once = strip_thinking_blocks(messages) + twice = strip_thinking_blocks(once) + assert once == twice