From a84c351ee93954081f47044f467222605d5ed1ad Mon Sep 17 00:00:00 2001 From: "Michael J. Jabbour" Date: Tue, 14 Jul 2026 22:27:21 -0400 Subject: [PATCH] =?UTF-8?q?fix:=20Close=20#207=20=E2=80=94=20sanitize=20th?= =?UTF-8?q?inking=20blocks=20in=20cross-provider=20sessions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cross-provider sessions can contain thinking-shaped blocks that the Anthropic API rejects on resume, causing 400 errors and bricked sessions. Added _sanitize_thinking_blocks() in the shared request path (_complete_chat_request, before cache-control application, inherited by fable subclass): strips thinking blocks lacking valid non-empty string signatures, covers both real-world shapes (signature:null from chat-completions; missing signature key from OpenAI Responses producer). - Inserts placeholder when message content empties (API rejects empty arrays) - Preserves valid thinking and redacted_thinking fields - Never raises, logs WARNING on strip, emits provider:thinking_blocks_sanitized event - 19 new tests covering both signature shapes, placeholder insertion, mixed content, no-mutation defensive suite Fixes: microsoft-amplifier/amplifier-support#207 Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- .../__init__.py | 128 ++++++ tests/test_thinking_block_sanitization.py | 374 ++++++++++++++++++ 2 files changed, 502 insertions(+) create mode 100644 tests/test_thinking_block_sanitization.py diff --git a/amplifier_module_provider_anthropic/__init__.py b/amplifier_module_provider_anthropic/__init__.py index a12eb55..5357ecb 100644 --- a/amplifier_module_provider_anthropic/__init__.py +++ b/amplifier_module_provider_anthropic/__init__.py @@ -230,8 +230,18 @@ async def _get_process_semaphore(max_concurrent: int) -> asyncio.Semaphore | Non BETA_HEADER_FAST_MODE = "fast-mode-2026-02-01" PROVIDER_FALLBACK_OPEN = "provider:fallback_open" PROVIDER_FALLBACK_ACTIVE = "provider:fallback_active" +PROVIDER_THINKING_BLOCKS_SANITIZED = "provider:thinking_blocks_sanitized" FALLBACK_STATE_VERSION = 1 +# Placeholder inserted when sanitizing thinking blocks leaves an assistant +# message with an empty content array. Anthropic rejects empty content +# arrays just as it rejects invalid-signature thinking blocks, so a message +# that becomes empty after stripping needs a minimal well-formed stand-in. +_THINKING_SANITIZE_PLACEHOLDER: dict[str, Any] = { + "type": "text", + "text": "[thinking content omitted]", +} + # --------------------------------------------------------------------------- # Deprecated model retirement dates — warn once per process per model # --------------------------------------------------------------------------- @@ -1375,6 +1385,95 @@ def _strip_thinking_blocks(request: ChatRequest) -> ChatRequest: ] return stripped + @staticmethod + def _sanitize_thinking_blocks( + messages: list[dict[str, Any]], + ) -> tuple[list[dict[str, Any]], int]: + """Strip thinking blocks with invalid signatures from assistant messages. + + Anthropic's API strict-validates `thinking.signature` as a non-empty + string and rejects the ENTIRE request with a 400 if any block fails + that check. On session resume, assistant turns produced by other + providers (or persisted before this provider's normal serialization + runs) can carry thinking-shaped blocks that don't meet this bar. + Two real-world shapes have been observed: + + - chat-completions producer: {"type": "thinking", ..., + "signature": None} + - OpenAI Responses producer: {"type": "thinking", ...} with the + "signature" key ABSENT entirely, and "content" holding + provider-internal strings (encrypted blob + response item id). + + Forwarding either shape gets the whole request rejected, bricking + the session on every subsequent resume attempt. Strip them here, + in the shared request-construction path, so every subclass + (including the Fable provider) inherits the protection. + + `redacted_thinking` blocks are never touched: they legitimately + carry no `signature` field (they carry `data` instead), and Anthropic + accepts them as-is. + + If stripping empties an assistant message's content array, a + minimal placeholder block is inserted -- Anthropic also rejects + messages with an empty content array. + + Defensive: this must never raise. A message that can't be + interpreted is passed through unchanged rather than aborting the + request; a malformed history block is a data-quality problem to + clean up, not a crash to propagate. + + Args: + messages: Anthropic-formatted message list (post-_convert_messages). + + Returns: + (sanitized_messages, stripped_count) -- a new list; the input + list and its unaffected message dicts are not mutated. + """ + sanitized: list[dict[str, Any]] = [] + stripped_count = 0 + + for msg in messages: + try: + if not isinstance(msg, dict) or msg.get("role") != "assistant": + sanitized.append(msg) + continue + + content = msg.get("content") + if not isinstance(content, list): + sanitized.append(msg) + continue + + kept_blocks = [] + message_stripped = 0 + for block in content: + if isinstance(block, dict) and block.get("type") == "thinking": + signature = block.get("signature") + if not isinstance(signature, str) or not signature.strip(): + message_stripped += 1 + continue + kept_blocks.append(block) + + if message_stripped == 0: + sanitized.append(msg) + continue + + stripped_count += message_stripped + new_msg = dict(msg) + if not kept_blocks: + # Emptying the array entirely would itself trigger a + # 400 -- keep the message well-formed. + kept_blocks = [dict(_THINKING_SANITIZE_PLACEHOLDER)] + new_msg["content"] = kept_blocks + sanitized.append(new_msg) + except Exception: + logger.exception( + "[PROVIDER] Error sanitizing thinking blocks for a message; " + "passing it through unmodified" + ) + sanitized.append(msg) + + return sanitized, stripped_count + @staticmethod def _is_overload_fallback_error(error: KernelLLMError) -> bool: """Return True when the error indicates model overload, not generic throttling.""" @@ -2205,6 +2304,16 @@ async def _complete_chat_request( # Combine: context THEN conversation all_messages = context_user_msgs + conversation_msgs + + # Sanitize thinking blocks BEFORE cache control so cache_control lands + # on the correct (post-sanitization) final block. This runs for every + # request in the shared base path, so the Fable subclass inherits it. + # See _sanitize_thinking_blocks() for the malformed shapes this guards + # against (fixes amplifier-support#207). + all_messages, _thinking_blocks_stripped = self._sanitize_thinking_blocks( + all_messages + ) + # Apply cache control to last message for incremental context caching all_messages = self._apply_message_cache_control(all_messages) logger.info(f"[PROVIDER] Final message count for API: {len(all_messages)}") @@ -2215,6 +2324,25 @@ async def _complete_chat_request( request_caps = await self._get_request_capabilities(effective_model) model_ceiling = request_caps.max_output_tokens + if _thinking_blocks_stripped: + logger.warning( + "[PROVIDER] Stripped %d thinking block(s) with invalid or " + "missing signature from assistant message history before " + "sending to the Anthropic API (model=%s). This happens when " + "a session resumes across a provider that doesn't produce " + "Anthropic-signed thinking blocks.", + _thinking_blocks_stripped, + effective_model, + ) + await self._emit_provider_event( + PROVIDER_THINKING_BLOCKS_SANITIZED, + { + "provider": "anthropic", + "model": effective_model, + "stripped_count": _thinking_blocks_stripped, + }, + ) + # Emit once-per-process deprecation warning for models nearing retirement if ( effective_model in _DEPRECATED_MODELS diff --git a/tests/test_thinking_block_sanitization.py b/tests/test_thinking_block_sanitization.py new file mode 100644 index 0000000..a21681e --- /dev/null +++ b/tests/test_thinking_block_sanitization.py @@ -0,0 +1,374 @@ +"""Tests for thinking-block sanitization (fixes amplifier-support#207). + +Cross-provider session resume can persist thinking-shaped content blocks +that the Anthropic API rejects outright, bricking the session. Two +malformed shapes are seen in the wild: + + - Shape A (chat-completions producer): thinking block with + ``signature: null``. + - Shape B (OpenAI Responses producer): thinking block with the + ``signature`` key ENTIRELY ABSENT, and ``content`` holding + provider-internal strings (an encrypted blob + a response item id, + e.g. ``["gAAAAAB...", "rs_abc123"]``). This is the same shape found in + the real corrupted transcript referenced in the issue (25 occurrences). + +``AnthropicProvider._sanitize_thinking_blocks`` runs in the shared +request-construction path (``_complete_chat_request``) for every request, +so subclasses (e.g. the Fable provider) inherit the protection. + +Covers: + (a) Shape A stripped. + (b) Shape B stripped. + (c) Valid thinking block (non-empty string signature) passes through. + (d) redacted_thinking passes through untouched (no signature expected). + (e) Mixed-content message keeps non-thinking blocks. + (f) Emptied assistant message gets a minimal placeholder block. + (g) Non-assistant / non-list-content messages are left untouched. + (h) Stripped counts accumulate correctly across blocks and messages. + (i) The sanitizer never mutates its input and never raises, even on + malformed/unexpected shapes. + (j) Integration through complete(): warning logged with count, event + emitted on the hook bus, and the sanitized (not raw) messages are + what actually get sent to the Anthropic API. +""" + +import asyncio +import logging +from typing import Any, cast +from unittest.mock import AsyncMock, MagicMock + +import pytest +from amplifier_core import ModuleCoordinator +from amplifier_core.message_models import ChatRequest, Message + +from amplifier_module_provider_anthropic import ( + PROVIDER_THINKING_BLOCKS_SANITIZED, + AnthropicProvider, +) +from tests._helpers import DummyResponse, FakeCoordinator + +# --------------------------------------------------------------------------- +# Fixtures: the two malformed shapes from the issue, plus valid comparisons +# --------------------------------------------------------------------------- + + +def _shape_a_null_signature() -> dict[str, Any]: + """chat-completions producer: signature explicitly null.""" + return { + "type": "thinking", + "thinking": "internal reasoning from a chat-completions turn", + "signature": None, + "visibility": "internal", + } + + +def _shape_b_missing_signature() -> dict[str, Any]: + """OpenAI Responses producer: signature key entirely absent. + + Mirrors the real corrupted transcript pattern: encrypted-blob + + response-item-id strings under "content", no "signature" key at all. + """ + return { + "type": "thinking", + "thinking": "internal reasoning from an OpenAI Responses turn", + "content": ["gAAAAAB_synthetic_encrypted_blob", "rs_synthetic_item_id"], + "visibility": "internal", + } + + +def _valid_thinking_block( + signature: str = "valid-anthropic-signature", +) -> dict[str, Any]: + return { + "type": "thinking", + "thinking": "internal reasoning from a real Anthropic turn", + "signature": signature, + } + + +def _redacted_thinking_block() -> dict[str, Any]: + return {"type": "redacted_thinking", "data": "opaque-redacted-blob"} + + +def _text_block(text: str = "hello") -> dict[str, Any]: + return {"type": "text", "text": text} + + +def _assistant_msg(content: list[dict[str, Any]]) -> dict[str, Any]: + return {"role": "assistant", "content": content} + + +# --------------------------------------------------------------------------- +# Unit tests: AnthropicProvider._sanitize_thinking_blocks (dict-level) +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + "malformed_block,label", + [ + (_shape_a_null_signature(), "shape-a-null-signature"), + (_shape_b_missing_signature(), "shape-b-missing-signature"), + ], +) +def test_malformed_thinking_block_is_stripped(malformed_block, label): + messages = [ + {"role": "user", "content": "hi"}, + _assistant_msg([malformed_block, _text_block("visible reply")]), + ] + + sanitized, stripped_count = AnthropicProvider._sanitize_thinking_blocks(messages) + + assert stripped_count == 1, label + assistant_out = sanitized[1] + types_out = [b["type"] for b in assistant_out["content"]] + assert "thinking" not in types_out, label + assert types_out == ["text"], label + assert assistant_out["content"][0]["text"] == "visible reply" + + +def test_valid_thinking_block_passes_through_unchanged(): + valid_block = _valid_thinking_block() + messages = [_assistant_msg([valid_block, _text_block("reply")])] + + sanitized, stripped_count = AnthropicProvider._sanitize_thinking_blocks(messages) + + assert stripped_count == 0 + assert sanitized[0]["content"][0] == valid_block + assert sanitized[0]["content"][0]["signature"] == "valid-anthropic-signature" + + +def test_redacted_thinking_passes_through_unchanged(): + redacted = _redacted_thinking_block() + messages = [_assistant_msg([redacted, _text_block("reply")])] + + sanitized, stripped_count = AnthropicProvider._sanitize_thinking_blocks(messages) + + assert stripped_count == 0 + types_out = [b["type"] for b in sanitized[0]["content"]] + assert "redacted_thinking" in types_out + assert sanitized[0]["content"][0] == redacted + + +def test_mixed_content_message_keeps_non_thinking_blocks(): + messages = [ + _assistant_msg( + [ + _shape_a_null_signature(), + _text_block("first"), + _redacted_thinking_block(), + _valid_thinking_block(), + _text_block("second"), + ] + ) + ] + + sanitized, stripped_count = AnthropicProvider._sanitize_thinking_blocks(messages) + + assert stripped_count == 1 + types_out = [b["type"] for b in sanitized[0]["content"]] + assert types_out == ["text", "redacted_thinking", "thinking", "text"] + + +def test_emptied_message_gets_placeholder_block(): + """Stripping the only block would leave an empty content array -- + Anthropic rejects that too, so a placeholder must be inserted.""" + messages = [_assistant_msg([_shape_b_missing_signature()])] + + sanitized, stripped_count = AnthropicProvider._sanitize_thinking_blocks(messages) + + assert stripped_count == 1 + content = sanitized[0]["content"] + assert len(content) == 1 + assert content != [] + assert content[0]["type"] == "text" + assert content[0].get("text") + + +def test_multiple_malformed_blocks_in_one_message_all_stripped(): + """Mirrors the real corrupted transcript: many shape-B blocks in a row.""" + many_malformed = [_shape_b_missing_signature() for _ in range(5)] + messages = [_assistant_msg([*many_malformed, _text_block("final reply")])] + + sanitized, stripped_count = AnthropicProvider._sanitize_thinking_blocks(messages) + + assert stripped_count == 5 + assert [b["type"] for b in sanitized[0]["content"]] == ["text"] + + +def test_stripped_count_accumulates_across_messages(): + messages = [ + _assistant_msg([_shape_a_null_signature(), _text_block("a")]), + {"role": "user", "content": "continue"}, + _assistant_msg([_shape_b_missing_signature(), _valid_thinking_block()]), + ] + + _, stripped_count = AnthropicProvider._sanitize_thinking_blocks(messages) + + assert stripped_count == 2 + + +def test_non_assistant_messages_untouched(): + user_msg = {"role": "user", "content": [_text_block("hi")]} + tool_msg = {"role": "tool", "content": [{"type": "tool_result", "content": "ok"}]} + messages = [user_msg, tool_msg] + + sanitized, stripped_count = AnthropicProvider._sanitize_thinking_blocks(messages) + + assert stripped_count == 0 + assert sanitized == [user_msg, tool_msg] + + +def test_assistant_message_with_string_content_untouched(): + messages = [{"role": "assistant", "content": "plain string reply"}] + + sanitized, stripped_count = AnthropicProvider._sanitize_thinking_blocks(messages) + + assert stripped_count == 0 + assert sanitized[0]["content"] == "plain string reply" + + +def test_does_not_mutate_input_messages(): + original_block = _shape_a_null_signature() + original_msg = _assistant_msg([original_block, _text_block("reply")]) + messages = [original_msg] + + sanitized, stripped_count = AnthropicProvider._sanitize_thinking_blocks(messages) + + assert stripped_count == 1 + # Original list/message/blocks are untouched. + assert original_msg["content"] == [original_block, _text_block("reply")] + assert len(original_msg["content"]) == 2 + # The returned message is a different object when content changed. + assert sanitized[0] is not original_msg + + +def test_unaffected_message_returned_as_is_not_copied(): + """Messages with no malformed thinking blocks aren't needlessly copied.""" + valid_msg = _assistant_msg([_valid_thinking_block(), _text_block("reply")]) + messages = [valid_msg] + + sanitized, stripped_count = AnthropicProvider._sanitize_thinking_blocks(messages) + + assert stripped_count == 0 + assert sanitized[0] is valid_msg + + +@pytest.mark.parametrize( + "weird_messages", + [ + [{"role": "assistant", "content": [None, "not-a-dict", 42]}], + [{"role": "assistant", "content": [{"type": "thinking"}]}], # no signature key + [{"role": "assistant"}], # no content key at all + ["not-a-dict-message"], + [{"role": "assistant", "content": [{"signature": "sig-no-type"}]}], + ], +) +def test_sanitizer_never_raises_on_malformed_shapes(weird_messages): + """Defensive: malformed/unexpected input is cleaned up, not a crash.""" + sanitized, stripped_count = AnthropicProvider._sanitize_thinking_blocks( + weird_messages + ) + assert isinstance(sanitized, list) + assert isinstance(stripped_count, int) + + +# --------------------------------------------------------------------------- +# Integration tests: through complete() -- observability + outgoing payload +# --------------------------------------------------------------------------- + + +def _make_provider(**config_overrides) -> AnthropicProvider: + provider = AnthropicProvider( + api_key="test-key", + config={ + "default_model": "claude-sonnet-4-5-20250929", + "use_streaming": False, + "max_retries": 0, + **config_overrides, + }, + ) + provider.coordinator = cast(ModuleCoordinator, FakeCoordinator()) + return provider + + +def _make_raw_mock() -> MagicMock: + raw = MagicMock() + raw.parse.return_value = DummyResponse() + raw.headers = {} + return raw + + +def _get_api_params(mock_create: AsyncMock) -> dict[str, Any]: + assert mock_create.await_count == 1 + _, kwargs = mock_create.call_args + return kwargs + + +def test_complete_strips_malformed_thinking_before_sending_to_api(caplog): + """End-to-end: a resumed session with a shape-A block in history must + not forward it to the Anthropic API, must log a warning with the + count, and must emit the sanitization event.""" + provider = _make_provider() + provider.client.messages.with_raw_response.create = AsyncMock( + return_value=_make_raw_mock() + ) + + # Simulate the post-_convert_messages output containing a malformed + # thinking block, as would occur on resume of a cross-provider session. + # (ChatRequest/Message models always normalize a missing "signature" + # key to signature=None, so we exercise the wire-level shapes directly + # against the method that runs after that conversion.) + original_convert = provider._convert_messages + + def _convert_with_malformed_history(messages): + converted = original_convert(messages) + converted.insert( + 0, + _assistant_msg( + [_shape_b_missing_signature(), _text_block("earlier reply")] + ), + ) + return converted + + provider._convert_messages = _convert_with_malformed_history + + request = ChatRequest(messages=[Message(role="user", content="continue")]) + + with caplog.at_level(logging.WARNING, logger="amplifier_module_provider_anthropic"): + asyncio.run(provider.complete(request)) + + params = _get_api_params(provider.client.messages.with_raw_response.create) + + # No malformed thinking block reached the API payload. + for msg in params["messages"]: + content = msg.get("content") + if isinstance(content, list): + for block in content: + if isinstance(block, dict) and block.get("type") == "thinking": + assert isinstance(block.get("signature"), str) + assert block["signature"].strip() + + # Warning logged with the stripped count. + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert any("Stripped" in r.message and "1" in r.message for r in warnings) + + # Event emitted on the hook bus with the correct payload shape. + hooks = cast(FakeCoordinator, provider.coordinator).hooks + assert PROVIDER_THINKING_BLOCKS_SANITIZED in hooks.emitted_names() + payload = hooks.payload_for(PROVIDER_THINKING_BLOCKS_SANITIZED) + assert payload is not None + assert payload["stripped_count"] == 1 + assert payload["provider"] == "anthropic" + + +def test_complete_emits_no_sanitization_event_when_nothing_stripped(): + provider = _make_provider() + provider.client.messages.with_raw_response.create = AsyncMock( + return_value=_make_raw_mock() + ) + + request = ChatRequest(messages=[Message(role="user", content="hello")]) + asyncio.run(provider.complete(request)) + + hooks = cast(FakeCoordinator, provider.coordinator).hooks + assert PROVIDER_THINKING_BLOCKS_SANITIZED not in hooks.emitted_names()