diff --git a/litellm/litellm_core_utils/prompt_templates/factory.py b/litellm/litellm_core_utils/prompt_templates/factory.py index c4e242fd360..746622aac29 100644 --- a/litellm/litellm_core_utils/prompt_templates/factory.py +++ b/litellm/litellm_core_utils/prompt_templates/factory.py @@ -6,6 +6,7 @@ import json import mimetypes import re import xml.etree.ElementTree as ET +from bisect import bisect_left from collections.abc import Iterator, Mapping, Sequence from enum import Enum from types import MappingProxyType @@ -2407,6 +2408,7 @@ def anthropic_messages_pt( ) = messages[msg_i] if user_message_types_block["role"] == "user": if isinstance(user_message_types_block["content"], list): + user_content_start: int = len(user_content) # pyright: ignore[reportGeneralTypeIssues] # reset per user message for m in user_message_types_block["content"]: if m.get("type", "") == "image_url": m = cast(ChatCompletionImageObject, m) @@ -2478,6 +2480,26 @@ def anthropic_messages_pt( _file_content_element, ) ) + if ( + len(user_content) > user_content_start + and user_message_types_block.get("cache_control") is not None + and not any( + "cache_control" in content and content["cache_control"] is not None + for content in user_content[user_content_start:] + ) + ): + user_content = [ # mutable-ok: Anthropic message content requires a list; build a fresh one + *user_content[:-1], + cast( # cast-ok: helper preserves the copied content block type + AnthropicMessagesUserMessageValues, + add_cache_control_to_content( + anthropic_content_element=cast( # cast-ok: copied TypedDict is a dict + dict[str, object], copy.copy(user_content[-1]) + ), + original_content_element=user_message_types_block, + ), + ), + ] elif isinstance(user_message_types_block["content"], str): _anthropic_content_text_element: AnthropicMessagesTextParam = { "type": "text", @@ -2538,6 +2560,28 @@ def anthropic_messages_pt( _has_server_tool_calls = True break + _content_is_list: bool = "content" in assistant_content_block and isinstance( + assistant_content_block["content"], list + ) # pyright: ignore[reportGeneralTypeIssues] # reset per assistant message + _content_list = assistant_content_block.get("content") if _content_is_list else None # pyright: ignore[reportGeneralTypeIssues] # reset per assistant message + _list_has_thinking: bool = ( + _content_is_list + and _content_list is not None + and any( + isinstance(item, dict) + and cast( # cast-ok: list items are inspected only for their type discriminator + dict[str, object], item + ).get("type") + in ("thinking", "redacted_thinking") + for item in _content_list + ) + ) # pyright: ignore[reportGeneralTypeIssues] # reset per assistant message + assistant_content_start: int = len(assistant_content) + ( + len(thinking_blocks) + if _content_is_list and thinking_blocks is not None and not _list_has_thinking + else 0 + ) # pyright: ignore[reportGeneralTypeIssues] # reset per assistant message + if ( thinking_blocks is not None and _has_server_tool_calls @@ -2661,20 +2705,6 @@ def anthropic_messages_pt( # duplication and preserve the original interleaved order. # Fixes the gap where list-content messages bypass INTERLEAVED # MODE and still get thinking blocks prepended out of order. - _content_is_list = "content" in assistant_content_block and isinstance( - assistant_content_block["content"], list - ) - _content_list = assistant_content_block.get("content") if _content_is_list else None - _list_has_thinking = False - if _content_is_list and _content_list is not None: - for _item in _content_list: - if isinstance(_item, dict) and _item.get("type") in ( - "thinking", - "redacted_thinking", - ): - _list_has_thinking = True - break - if ( thinking_blocks is not None and not _list_has_thinking ): # IMPORTANT: ADD THIS FIRST, ELSE ANTHROPIC WILL RAISE AN ERROR @@ -2764,6 +2794,28 @@ def anthropic_messages_pt( if assistant_function_call is not None: assistant_content.extend(convert_function_to_anthropic_tool_invoke(assistant_function_call)) + if ( + _content_is_list + and len(assistant_content) > assistant_content_start + and assistant_content_block.get("cache_control") is not None + and not any( + "cache_control" in content and content["cache_control"] is not None + for content in assistant_content[assistant_content_start:] + ) + ): + assistant_content = [ # mutable-ok: Anthropic message content requires a list; build a fresh one + *assistant_content[:-1], + cast( # cast-ok: helper preserves the copied content block type + AnthropicMessagesAssistantMessageValues, + add_cache_control_to_content( + anthropic_content_element=cast( # cast-ok: copied TypedDict is a dict + dict[str, object], copy.copy(assistant_content[-1]) + ), + original_content_element=assistant_content_block, + ), + ), + ] + msg_i += 1 if assistant_content: @@ -4031,24 +4083,32 @@ def _sort_bedrock_assistant_content_blocks( the request with a 400 error because the ``toolUse`` → ``toolResult`` pairing is broken by the intervening ``text`` block. - Sort order (stable): + Sort order (stable) inside each cache-point segment: 0 - reasoningContent 1 - text / image / document / video / other non-tool blocks 2 - toolUse + 3 - cachePoint (kept at its original boundary) """ - def _sort_key(block: BedrockContentBlock) -> int: - if "reasoningContent" in block: - return 0 - if "toolUse" in block: - return 2 - if "cachePoint" in block: - # cachePoint blocks are paired with their preceding toolUse block. - # Same key as toolUse so Python's stable sort keeps them together. - return 2 - return 1 + cache_point_indices: Final[tuple[int, ...]] = tuple( + index for index, block in enumerate(blocks) if "cachePoint" in block + ) - return sorted(blocks, key=_sort_key) + def _sort_key(indexed_block: tuple[int, BedrockContentBlock]) -> tuple[int, int, int]: + index, block = indexed_block + segment_index: Final[int] = bisect_left(cache_point_indices, index) + if "reasoningContent" in block: + return segment_index, 0, index + if "cachePoint" in block: + return segment_index, 3, index + if "toolUse" in block: + return segment_index, 2, index + return segment_index, 1, index + + sorted_indexed_blocks: Final = sorted(enumerate(blocks), key=_sort_key) + return [ # mutable-ok: Bedrock Converse requires a list of content blocks + block for _, block in sorted_indexed_blocks + ] def _insert_assistant_continue_message( @@ -4417,6 +4477,16 @@ class BedrockConverseMessagesProcessor: ) if _cache_point_block is not None: _parts.append(_cache_point_block) + if ( + _parts + and message_block.get("cache_control") is not None + and not any("cachePoint" in part for part in _parts) + ): + _cache_point_block = litellm.AmazonConverseConfig().get_cache_point_block( + message_block, block_type="content_block", model=model + ) + if _cache_point_block is not None: + _parts.append(_cache_point_block) user_content.extend(_parts) elif message_block["content"] and isinstance(message_block["content"], str): _part = BedrockContentBlock(text=messages[msg_i]["content"]) @@ -4502,6 +4572,7 @@ class BedrockConverseMessagesProcessor: message=messages[msg_i], assistant_continue_message=assistant_continue_message, ) + assistant_message_content_start: int = len(assistant_content) # pyright: ignore[reportGeneralTypeIssues] # reset per assistant message _assistant_content = assistant_message_block.get("content", None) thinking_blocks = cast( list[ChatCompletionThinkingBlock] | None, @@ -4560,16 +4631,30 @@ class BedrockConverseMessagesProcessor: assistant_content.append(BedrockContentBlock(text=_assistant_content)) # If content is empty/whitespace, skip it (don't add a placeholder) # Add cache point block for assistant string content - _cache_point_block = litellm.AmazonConverseConfig().get_cache_point_block( - assistant_message_block, block_type="content_block", model=model - ) - if _cache_point_block is not None: - assistant_content.append(_cache_point_block) + if ( + assistant_message_cache_point_block := litellm.AmazonConverseConfig().get_cache_point_block( + assistant_message_block, block_type="content_block", model=model + ) + ) is not None: + assistant_content.append(assistant_message_cache_point_block) _tool_calls = assistant_message_block.get("tool_calls", []) if _tool_calls: assistant_content.extend(_convert_to_bedrock_tool_call_invoke(_tool_calls, model=model)) + if ( + isinstance(_assistant_content, list) + and len(assistant_content) > assistant_message_content_start + and assistant_message_block.get("cache_control") is not None + and not any("cachePoint" in block for block in assistant_content[assistant_message_content_start:]) + ): + if ( + assistant_list_cache_point_block := litellm.AmazonConverseConfig().get_cache_point_block( + assistant_message_block, block_type="content_block", model=model + ) + ) is not None: + assistant_content.append(assistant_list_cache_point_block) + msg_i += 1 assistant_content = _deduplicate_bedrock_content_blocks(assistant_content, "toolUse") @@ -4790,6 +4875,16 @@ def _bedrock_converse_messages_pt( ) if _cache_point_block is not None: _parts.append(_cache_point_block) + if ( + _parts + and message_block.get("cache_control") is not None + and not any("cachePoint" in part for part in _parts) + ): + _cache_point_block = litellm.AmazonConverseConfig().get_cache_point_block( + message_block, block_type="content_block", model=model + ) + if _cache_point_block is not None: + _parts.append(_cache_point_block) user_content.extend(_parts) elif message_block["content"] and isinstance(message_block["content"], str): _part = BedrockContentBlock(text=messages[msg_i]["content"]) @@ -4877,6 +4972,7 @@ def _bedrock_converse_messages_pt( message=messages[msg_i], assistant_continue_message=assistant_continue_message, ) + assistant_message_content_start: int = len(assistant_content) # pyright: ignore[reportGeneralTypeIssues] # reset per assistant message _assistant_content = assistant_message_block.get("content", None) thinking_blocks = cast( list[ChatCompletionThinkingBlock] | None, @@ -4937,15 +5033,29 @@ def _bedrock_converse_messages_pt( if _assistant_content.strip(): assistant_content.append(BedrockContentBlock(text=_assistant_content)) # Add cache point block for assistant string content - _cache_point_block = litellm.AmazonConverseConfig().get_cache_point_block( - assistant_message_block, block_type="content_block", model=model - ) - if _cache_point_block is not None: - assistant_content.append(_cache_point_block) + if ( + assistant_message_cache_point_block := litellm.AmazonConverseConfig().get_cache_point_block( + assistant_message_block, block_type="content_block", model=model + ) + ) is not None: + assistant_content.append(assistant_message_cache_point_block) _tool_calls = assistant_message_block.get("tool_calls", []) if _tool_calls: assistant_content.extend(_convert_to_bedrock_tool_call_invoke(_tool_calls, model=model)) + if ( + isinstance(_assistant_content, list) + and len(assistant_content) > assistant_message_content_start + and assistant_message_block.get("cache_control") is not None + and not any("cachePoint" in block for block in assistant_content[assistant_message_content_start:]) + ): + if ( + assistant_list_cache_point_block := litellm.AmazonConverseConfig().get_cache_point_block( + assistant_message_block, block_type="content_block", model=model + ) + ) is not None: + assistant_content.append(assistant_list_cache_point_block) + msg_i += 1 assistant_content = _deduplicate_bedrock_content_blocks(assistant_content, "toolUse") diff --git a/litellm/llms/anthropic/chat/transformation.py b/litellm/llms/anthropic/chat/transformation.py index 3bffee48d6a..08f4833e6a5 100644 --- a/litellm/llms/anthropic/chat/transformation.py +++ b/litellm/llms/anthropic/chat/transformation.py @@ -1719,6 +1719,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig): anthropic_system_message_content["cache_control"] = system_message_block["cache_control"] anthropic_system_message_list.append(anthropic_system_message_content) elif isinstance(message["content"], list): + content_start: int = len(anthropic_system_message_list) # pyright: ignore[reportGeneralTypeIssues] # reset per system message for _content in message["content"]: # Skip empty text blocks - Anthropic API raises errors for empty text text_value = _content.get("text") @@ -1739,6 +1740,15 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig): anthropic_system_message_content["cache_control"] = _content["cache_control"] anthropic_system_message_list.append(anthropic_system_message_content) + if ( + len(anthropic_system_message_list) > content_start + and system_message_block.get("cache_control") is not None + and not any( + content.get("cache_control") is not None + for content in anthropic_system_message_list[content_start:] + ) + ): + anthropic_system_message_list[-1]["cache_control"] = system_message_block["cache_control"] if len(system_prompt_indices) > 0: for idx in reversed(system_prompt_indices): diff --git a/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py b/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py index 1e12a973cdb..af9b70a017e 100644 --- a/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py +++ b/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py @@ -15,10 +15,11 @@ from litellm.litellm_core_utils.prompt_templates.factory import ( BedrockImageProcessor, _bedrock_converse_messages_pt, _bedrock_tools_pt, - _rename_duplicate_bedrock_document_names, _convert_to_bedrock_tool_call_invoke, - _sanitize_anthropic_tool_use_id, _convert_to_bedrock_tool_call_result, + _rename_duplicate_bedrock_document_names, + _sanitize_anthropic_tool_use_id, + _sort_bedrock_assistant_content_blocks, anthropic_messages_pt, convert_to_anthropic_tool_result, convert_to_gemini_tool_call_result, @@ -30,6 +31,266 @@ from litellm.litellm_core_utils.prompt_templates.factory import ( from litellm.types.llms.openai import ChatCompletionToolMessage +@pytest.mark.parametrize("role", ["user", "assistant"]) +@pytest.mark.parametrize("block_cache_control", [False, True]) +def test_anthropic_list_content_preserves_message_cache_control(role, block_cache_control): + message_cache_control: Final = {"type": "ephemeral"} + block_cache: Final = {"type": "ephemeral", "ttl": "1h"} + content: Final = [ + {"type": "text", "text": "first", **({"cache_control": block_cache} if block_cache_control else {})}, + {"type": "text", "text": "second"}, + ] + message: Final = {"role": role, "content": content, "cache_control": message_cache_control} + messages: Final = [message] if role == "user" else [{"role": "user", "content": "hi"}, message] + + result: Final = anthropic_messages_pt(messages=messages, model="claude-sonnet-4-6", llm_provider="anthropic") + + expected: Final = [ + {"type": "text", "text": "first", **({"cache_control": block_cache} if block_cache_control else {})}, + { + "type": "text", + "text": "second", + **({} if block_cache_control else {"cache_control": message_cache_control}), + }, + ] + assert result[-1]["content"] == expected + + +def test_anthropic_list_message_cache_control_skips_separate_thinking_blocks(): + thinking_block: Final = { + "type": "thinking", + "thinking": "reasoning", + "signature": "valid-signature", + } + cache_control: Final = {"type": "ephemeral"} + messages: Final = [ + {"role": "user", "content": "hi"}, + { + "role": "assistant", + "content": [{"type": "text", "text": "answer"}], + "thinking_blocks": [thinking_block], + "cache_control": cache_control, + }, + ] + + result: Final = anthropic_messages_pt(messages=messages, model="claude-sonnet-4-6", llm_provider="anthropic") + + assert result[-1]["content"] == [ + thinking_block, + {"type": "text", "text": "answer", "cache_control": cache_control}, + ] + + +@pytest.mark.parametrize("call_field", ["tool_calls", "function_call"]) +def test_anthropic_list_message_cache_control_follows_tool_calls(call_field): + cache_control: Final = {"type": "ephemeral"} + call: Final = ( + [{"id": "call_1", "type": "function", "function": {"name": "lookup", "arguments": "{}"}}] + if call_field == "tool_calls" + else {"name": "lookup", "arguments": "{}"} + ) + messages: Final = [ + {"role": "user", "content": "hi"}, + { + "role": "assistant", + "content": [{"type": "text", "text": "checking"}], + "cache_control": cache_control, + call_field: call, + }, + ] + + result: Final = anthropic_messages_pt(messages=messages, model="claude-sonnet-4-6", llm_provider="anthropic") + blocks: Final = result[-1]["content"] + + assert blocks[0] == {"type": "text", "text": "checking"} + assert blocks[-1]["type"] == "tool_use" + assert blocks[-1]["cache_control"] == cache_control + + +@pytest.mark.parametrize("role", ["user", "assistant"]) +def test_anthropic_list_message_cache_control_preserves_caller_block(role): + block: Final = ( + {"type": "document", "source": {"type": "text", "media_type": "text/plain", "data": "document"}} + if role == "user" + else {"type": "server_tool_use", "id": "srvtoolu_1", "name": "web_search", "input": {}} + ) + messages: Final = [ + {"role": "user", "content": "hi"}, + {"role": role, "content": [block], "cache_control": {"type": "ephemeral"}}, + ] + + result: Final = anthropic_messages_pt(messages=messages, model="claude-sonnet-4-6", llm_provider="anthropic") + + assert "cache_control" not in block + assert result[-1]["content"][-1]["cache_control"] == {"type": "ephemeral"} + assert result[-1]["content"][-1] is not block + + +@pytest.mark.parametrize("role", ["user", "assistant"]) +def test_anthropic_skipped_list_block_does_not_suppress_message_cache_control(role): + skipped_block: Final = ( + {"type": "unknown", "cache_control": {"type": "ephemeral", "ttl": "1h"}} + if role == "user" + else {"type": "thinking", "thinking": "unsigned", "cache_control": {"type": "ephemeral", "ttl": "1h"}} + ) + messages: Final = [ + {"role": "user", "content": "hi"}, + { + "role": role, + "content": [skipped_block, {"type": "text", "text": "retained"}], + "cache_control": {"type": "ephemeral"}, + }, + ] + + result: Final = anthropic_messages_pt(messages=messages, model="claude-sonnet-4-6", llm_provider="anthropic") + + assert result[-1]["content"][-1] == {"type": "text", "text": "retained", "cache_control": {"type": "ephemeral"}} + assert all(block.get("type") != skipped_block["type"] for block in result[-1]["content"]) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("block_cache_control", [False, True]) +async def test_bedrock_list_content_preserves_message_cache_control(block_cache_control): + content: Final = [ + {"type": "text", "text": "first", **({"cache_control": {"type": "ephemeral"}} if block_cache_control else {})}, + {"type": "text", "text": "second"}, + ] + messages: Final = [{"role": "user", "content": content, "cache_control": {"type": "ephemeral"}}] + model: Final = "us.anthropic.claude-sonnet-4-5-20250929-v1:0" + + sync_result: Final = _bedrock_converse_messages_pt(messages=messages, model=model, llm_provider="bedrock") + async_result: Final = await BedrockConverseMessagesProcessor._bedrock_converse_messages_pt_async( + messages=messages, model=model, llm_provider="bedrock" + ) + + cache_point: Final = {"cachePoint": {"type": "default"}} + expected: Final = ( + [{"text": "first"}, cache_point, {"text": "second"}] + if block_cache_control + else [{"text": "first"}, {"text": "second"}, cache_point] + ) + assert sync_result[0]["content"] == expected + assert async_result[0]["content"] == expected + + +@pytest.mark.asyncio +@pytest.mark.parametrize("block_cache_control", [False, True]) +@pytest.mark.parametrize("with_tool_calls", [False, True]) +async def test_bedrock_assistant_list_preserves_message_cache_control(block_cache_control, with_tool_calls): + content: Final = [ + { + "type": "text", + "text": "checking", + **({"cache_control": {"type": "ephemeral"}} if block_cache_control else {}), + } + ] + assistant_message: Final = { + "role": "assistant", + "content": content, + "cache_control": {"type": "ephemeral"}, + **( + {"tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "lookup", "arguments": "{}"}}]} + if with_tool_calls + else {} + ), + } + messages: Final = [{"role": "user", "content": "hi"}, assistant_message] + model: Final = "us.anthropic.claude-sonnet-4-5-20250929-v1:0" + + sync_result: Final = _bedrock_converse_messages_pt(messages=messages, model=model, llm_provider="bedrock") + async_result: Final = await BedrockConverseMessagesProcessor._bedrock_converse_messages_pt_async( + messages=messages, model=model, llm_provider="bedrock" + ) + + for result in (sync_result, async_result): + blocks = result[-1]["content"] + assert sum("cachePoint" in block for block in blocks) == 1 + if not block_cache_control: + assert blocks[-1] == {"cachePoint": {"type": "default"}} + if with_tool_calls: + assert "toolUse" in blocks[-2] + + +@pytest.mark.asyncio +async def test_bedrock_assistant_string_preserves_message_cache_control(): + messages: Final = [ + {"role": "user", "content": "hi"}, + {"role": "assistant", "content": "checking", "cache_control": {"type": "ephemeral"}}, + ] + model: Final = "us.anthropic.claude-sonnet-4-5-20250929-v1:0" + expected: Final = [{"text": "checking"}, {"cachePoint": {"type": "default"}}] + + sync_result: Final = _bedrock_converse_messages_pt(messages=messages, model=model, llm_provider="bedrock") + async_result: Final = await BedrockConverseMessagesProcessor._bedrock_converse_messages_pt_async( + messages=messages, model=model, llm_provider="bedrock" + ) + + assert sync_result[-1]["content"] == expected + assert async_result[-1]["content"] == expected + + +@pytest.mark.asyncio +async def test_bedrock_assistant_message_cache_point_stays_before_following_assistant_text(): + messages: Final = [ + {"role": "user", "content": "hi"}, + { + "role": "assistant", + "content": [{"type": "text", "text": "cached prefix"}], + "cache_control": {"type": "ephemeral"}, + }, + {"role": "assistant", "content": [{"type": "text", "text": "uncached suffix"}]}, + ] + model: Final = "us.anthropic.claude-sonnet-4-5-20250929-v1:0" + expected: Final = [ + {"text": "cached prefix"}, + {"cachePoint": {"type": "default"}}, + {"text": "uncached suffix"}, + ] + + sync_result: Final = _bedrock_converse_messages_pt(messages=messages, model=model, llm_provider="bedrock") + async_result: Final = await BedrockConverseMessagesProcessor._bedrock_converse_messages_pt_async( + messages=messages, model=model, llm_provider="bedrock" + ) + + assert sync_result[-1]["content"] == expected + assert async_result[-1]["content"] == expected + + +def test_bedrock_assistant_content_sorting_keeps_cache_point_boundaries(): + first_cache_point: Final = {"cachePoint": {"type": "default"}} + second_cache_point: Final = {"cachePoint": {"type": "default", "ttl": "5m"}} + before_cache_point: Final = {"text": "before cache point"} + after_first_cache_point: Final = {"text": "after first cache point"} + after_second_cache_point: Final = {"text": "after second cache point"} + first_tool_use: Final = {"toolUse": {"toolUseId": "call_1", "name": "first"}} + second_tool_use: Final = {"toolUse": {"toolUseId": "call_2", "name": "second"}} + reasoning: Final = {"reasoningContent": {"reasoningText": {"text": "reasoning"}}} + + result: Final = _sort_bedrock_assistant_content_blocks( + [ + first_tool_use, + before_cache_point, + first_cache_point, + second_tool_use, + after_first_cache_point, + reasoning, + second_cache_point, + after_second_cache_point, + ] + ) + + assert result == [ + before_cache_point, + first_tool_use, + first_cache_point, + reasoning, + after_first_cache_point, + second_tool_use, + second_cache_point, + after_second_cache_point, + ] + + def _get_gemini_function_response_inline_data_parts(result): assert isinstance(result, list), "expected Gemini parts list" assert len(result) == 1, "multimodal function responses should stay in one part" diff --git a/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py b/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py index 332153b4c7d..7fdf8032dba 100644 --- a/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py +++ b/tests/unit/llms/anthropic/chat/test_anthropic_chat_transformation.py @@ -2383,6 +2383,50 @@ def test_translate_system_message_preserves_cache_control(): assert result[0]["cache_control"] == {"type": "ephemeral"} +@pytest.mark.parametrize("block_cache_control", [False, True]) +def test_translate_system_list_content_preserves_message_cache_control(block_cache_control): + message_cache_control: Final = {"type": "ephemeral"} + block_cache: Final = {"type": "ephemeral", "ttl": "1h"} + content: Final = [ + {"type": "text", "text": "first", **({"cache_control": block_cache} if block_cache_control else {})}, + {"type": "text", "text": "second"}, + ] + messages: Final = [ + {"role": "system", "content": content, "cache_control": message_cache_control}, + {"role": "user", "content": "hi"}, + ] + + result: Final = AnthropicConfig().translate_system_message(messages) + + expected: Final = [ + {"type": "text", "text": "first", **({"cache_control": block_cache} if block_cache_control else {})}, + { + "type": "text", + "text": "second", + **({} if block_cache_control else {"cache_control": message_cache_control}), + }, + ] + assert result == expected + + +def test_translate_system_skipped_list_block_does_not_suppress_message_cache_control(): + messages: Final = [ + { + "role": "system", + "content": [ + {"type": "text", "text": "", "cache_control": {"type": "ephemeral", "ttl": "1h"}}, + {"type": "text", "text": "retained"}, + ], + "cache_control": {"type": "ephemeral"}, + }, + {"role": "user", "content": "hi"}, + ] + + result: Final = AnthropicConfig().translate_system_message(messages) + + assert result == [{"type": "text", "text": "retained", "cache_control": {"type": "ephemeral"}}] + + # ============ Dynamic max_tokens Tests ============