diff --git a/litellm/litellm_core_utils/prompt_templates/common_utils.py b/litellm/litellm_core_utils/prompt_templates/common_utils.py index b704dfa2675..5df8bad852e 100644 --- a/litellm/litellm_core_utils/prompt_templates/common_utils.py +++ b/litellm/litellm_core_utils/prompt_templates/common_utils.py @@ -16,7 +16,7 @@ from pathlib import Path from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, Literal, TypeVar, cast -from pydantic import TypeAdapter +from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError import litellm from litellm import verbose_logger @@ -28,6 +28,7 @@ from litellm.types.llms.openai import ( ChatCompletionImageObject, ChatCompletionReasoningItem, ChatCompletionReasoningSummaryTextBlock, + ChatCompletionReasoningTextBlock, ChatCompletionRedactedThinkingBlock, ChatCompletionResponseMessage, ChatCompletionTextObject, @@ -2256,6 +2257,31 @@ def reasoning_content_from_thinking_blocks( ENCRYPTED_REASONING_SIGNATURE_PREFIX: Final = "litellm_encrypted_reasoning:" +REASONING_CONTENT_SIGNATURE_PREFIX: Final = "litellm_reasoning_content:" + + +class _ReasoningContentReplay(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + summary: tuple[str, ...] + content: tuple[str, ...] + encrypted_content: str | None = None + + +def reasoning_content_signature( + summary: tuple[str, ...], content: tuple[str, ...], encrypted_content: str | None +) -> str: + payload: Final = _ReasoningContentReplay(summary=summary, content=content, encrypted_content=encrypted_content) + return f"{REASONING_CONTENT_SIGNATURE_PREFIX}{payload.model_dump_json()}" + + +def _reasoning_content_from_signature(signature: object) -> _ReasoningContentReplay | None: + if not isinstance(signature, str) or not signature.startswith(REASONING_CONTENT_SIGNATURE_PREFIX): + return None + try: + return _ReasoningContentReplay.model_validate_json(signature.removeprefix(REASONING_CONTENT_SIGNATURE_PREFIX)) + except ValidationError: + return None def encrypted_reasoning_signature(encrypted_content: str) -> str: @@ -2269,11 +2295,16 @@ def encrypted_reasoning_signature(encrypted_content: str) -> str: def _carries_encrypted_reasoning(signature: object) -> bool: - return isinstance(signature, str) and signature.startswith(ENCRYPTED_REASONING_SIGNATURE_PREFIX) + return isinstance(signature, str) and signature.startswith( + (ENCRYPTED_REASONING_SIGNATURE_PREFIX, REASONING_CONTENT_SIGNATURE_PREFIX) + ) def encrypted_content_from_signature(signature: object) -> str | None: - if not isinstance(signature, str) or not _carries_encrypted_reasoning(signature): + replay: Final = _reasoning_content_from_signature(signature) + if replay is not None: + return replay.encrypted_content + if not isinstance(signature, str) or not signature.startswith(ENCRYPTED_REASONING_SIGNATURE_PREFIX): return None return signature.removeprefix(ENCRYPTED_REASONING_SIGNATURE_PREFIX) or None @@ -2404,6 +2435,16 @@ def _reasoning_replay_group_key(indexed_block: tuple[int, Mapping[str, object]]) def _reasoning_item_from_block_group(group: tuple[Mapping[str, object], ...]) -> ChatCompletionReasoningItem | None: + replay: Final = _reasoning_content_from_signature(_encrypted_reasoning_field(group[0])) + if replay is not None: + return ChatCompletionReasoningItem( + type="reasoning", + summary=[ + ChatCompletionReasoningSummaryTextBlock(type="summary_text", text=text) for text in replay.summary + ], + content=[ChatCompletionReasoningTextBlock(type="reasoning_text", text=text) for text in replay.content], + **({"encrypted_content": replay.encrypted_content} if replay.encrypted_content is not None else {}), + ) summary: Final[list[ChatCompletionReasoningSummaryTextBlock]] = [ # mutable-ok: API message payload ChatCompletionReasoningSummaryTextBlock(type="summary_text", text=text) for block in group diff --git a/litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py b/litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py index 85287f5433a..01f6475768e 100644 --- a/litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py +++ b/litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py @@ -11,9 +11,6 @@ from litellm import verbose_logger from litellm._logging import redact_internal_details_from_client_message from litellm._uuid import uuid from litellm.exceptions import MidStreamFallbackError -from litellm.litellm_core_utils.prompt_templates.common_utils import ( - encrypted_reasoning_signature, -) from litellm.llms.anthropic.pass_through.messages.utils import ( INCOMPLETE_STREAM_ERROR_MESSAGE, refusal_stop_details, @@ -140,6 +137,7 @@ class AnthropicResponsesStreamWrapper: self._stream_failed = False self._chunk_queue: deque[dict[str, object]] = deque() self._refusal_text: str = "" + self._reasoning_text_parts_seen: frozenset[tuple[str | None, int]] = frozenset() self._sync_responses_iterator: Iterator[object] | None = None def _make_message_start(self) -> dict[str, object]: @@ -183,13 +181,26 @@ class AnthropicResponsesStreamWrapper: def _field(source: object, name: str) -> object: return source.get(name) if isinstance(source, dict) else getattr(source, name, None) + def _emit_thinking_text(self, item_id: str | None, text: str) -> None: + existing_index: Final = self._item_id_to_block_index.get(item_id, -1) if item_id else self._current_block_index + block_index: Final = ( + existing_index + if existing_index >= 0 + else self._open_block(item_id, {"type": "thinking", "thinking": "", "signature": ""}) + ) + self._chunk_queue.append( + {"type": "content_block_delta", "index": block_index, "delta": {"type": "thinking_delta", "thinking": text}} + ) + def _close_reasoning_item(self, item: object, item_id: str | None) -> None: + reasoning_block: Final = LiteLLMAnthropicToResponsesAPIAdapter.thinking_block_from_reasoning_item(item) + existing_index: Final = self._item_id_to_block_index.get(item_id, -1) if item_id else self._current_block_index + completed_text: Final = reasoning_block.get("thinking") if reasoning_block is not None else None + if existing_index < 0 and isinstance(completed_text, str) and completed_text: + self._emit_thinking_text(item_id, completed_text) block_idx: Final = self._item_id_to_block_index.get(item_id, -1) if item_id else self._current_block_index - encrypted_content: Final = self._field(item, "encrypted_content") signature: Final = ( - encrypted_reasoning_signature(encrypted_content) - if isinstance(encrypted_content, str) and encrypted_content - else None + reasoning_block.get("signature", reasoning_block.get("data")) if reasoning_block is not None else None ) if block_idx < 0 and signature is None: return @@ -213,9 +224,7 @@ class AnthropicResponsesStreamWrapper: def _process_event(self, event: object) -> None: """Convert one Responses API event into zero or more Anthropic chunks queued for emission.""" - event_type = getattr(event, "type", None) - if event_type is None and isinstance(event, dict): - event_type = event.get("type") + event_type: Final = self._field(event, "type") if event_type is None: return @@ -312,25 +321,25 @@ class AnthropicResponsesStreamWrapper: ) return - # ---- reasoning summary text delta ---- - if event_type == "response.reasoning_summary_text.delta": - item_id = getattr(event, "item_id", None) or (event.get("item_id") if isinstance(event, dict) else None) - delta = getattr(event, "delta", "") or (event.get("delta", "") if isinstance(event, dict) else "") - block_idx = self._item_id_to_block_index.get(item_id, -1) if item_id else self._current_block_index - if block_idx < 0: - if not delta: - return - block_idx = self._open_block( - item_id, - {"type": "thinking", "thinking": "", "signature": ""}, - ) - self._chunk_queue.append( - { - "type": "content_block_delta", - "index": block_idx, - "delta": {"type": "thinking_delta", "thinking": delta}, - } + if event_type in ( + "response.reasoning_summary_text.delta", + "response.reasoning_text.delta", + "response.reasoning_text.done", + ): + reasoning_item_id: Final = self._field(event, "item_id") + reasoning_id: Final = reasoning_item_id if isinstance(reasoning_item_id, str) else None + content_index: Final = self._field(event, "content_index") + part_key: Final = (reasoning_id, content_index if isinstance(content_index, int) else 0) + if event_type == "response.reasoning_text.done" and part_key in self._reasoning_text_parts_seen: + return + reasoning_text: Final = self._field( + event, "text" if event_type == "response.reasoning_text.done" else "delta" ) + if not isinstance(reasoning_text, str) or not reasoning_text: + return + if event_type != "response.reasoning_summary_text.delta": + self._reasoning_text_parts_seen = self._reasoning_text_parts_seen | frozenset((part_key,)) + self._emit_thinking_text(reasoning_id, reasoning_text) return # ---- function call arguments delta ---- diff --git a/litellm/llms/anthropic/pass_through/responses_adapters/transformation.py b/litellm/llms/anthropic/pass_through/responses_adapters/transformation.py index 97f2db36a85..7860bf7451e 100644 --- a/litellm/llms/anthropic/pass_through/responses_adapters/transformation.py +++ b/litellm/llms/anthropic/pass_through/responses_adapters/transformation.py @@ -10,10 +10,13 @@ from collections.abc import Iterable, Mapping, Sequence from itertools import groupby from typing import Any, Final, cast +from pydantic import BaseModel, ConfigDict + from litellm.litellm_core_utils.prompt_templates.common_utils import ( TOOL_RESULT_IMAGE_BOUNDARY, TOOL_RESULT_IMAGE_PLACEHOLDER, encrypted_reasoning_signature, + reasoning_content_signature, responses_reasoning_items_from_thinking_blocks, with_prompt_cache_breakpoint, ) @@ -54,6 +57,14 @@ REASONING_SUMMARY_PART_SEPARATOR: Final = "\n\n" RESPONSES_INCLUDE_ENCRYPTED_REASONING: Final = "reasoning.encrypted_content" +class _ReasoningItem(BaseModel): + model_config = ConfigDict(frozen=True, from_attributes=True) + + summary: tuple[object, ...] | None = None + content: tuple[object, ...] | None = None + encrypted_content: str | None = None + + class LiteLLMAnthropicToResponsesAPIAdapter: """ Converts Anthropic /v1/messages requests to OpenAI Responses API format and @@ -246,10 +257,9 @@ class LiteLLMAnthropicToResponsesAPIAdapter: return str(getattr(part, "text", None) or "") @classmethod - def _thinking_block_from_reasoning_item( + def thinking_block_from_reasoning_item( cls, - summary: Iterable[object], - encrypted_content: object, + item: object, ) -> dict[str, object] | None: # mutable-ok: API message payload """The one Anthropic block for a Responses reasoning item. @@ -258,23 +268,37 @@ class LiteLLMAnthropicToResponsesAPIAdapter: turn replays the very item OpenAI produced; without it the signature stays empty, since only Anthropic can sign a thinking block. """ - text: Final = REASONING_SUMMARY_PART_SEPARATOR.join( - part_text for part in summary if (part_text := cls._summary_part_text(part)) + reasoning: Final = _ReasoningItem.model_validate(item) + summary_texts: Final = tuple( + part_text for part in reasoning.summary or () if (part_text := cls._summary_part_text(part)) ) - if not isinstance(encrypted_content, str) or not encrypted_content: + content_texts: Final = tuple( + part_text for part in reasoning.content or () if (part_text := cls._summary_part_text(part)) + ) + text: Final = REASONING_SUMMARY_PART_SEPARATOR.join(content_texts or summary_texts) + if content_texts: + content_signature: Final = reasoning_content_signature( + summary_texts, content_texts, reasoning.encrypted_content + ) + return AnthropicResponseContentBlockThinking( + type="thinking", thinking=text, signature=content_signature + ).model_dump() + if not reasoning.encrypted_content: if not text: return None return AnthropicResponseContentBlockThinking(type="thinking", thinking=text, signature=None).model_dump() - signature: Final = encrypted_reasoning_signature(encrypted_content) + signature: Final = encrypted_reasoning_signature(reasoning.encrypted_content) if not text: return AnthropicResponseContentBlockRedactedThinking(type="redacted_thinking", data=signature).model_dump() return AnthropicResponseContentBlockThinking(type="thinking", thinking=text, signature=signature).model_dump() @staticmethod def _assistant_block_group_key(indexed_block: tuple[int, Mapping[str, object]]) -> str: - """Group a run of consecutive thinking blocks together; keep every other block alone.""" index, block = indexed_block - return "thinking" if block.get("type") in ("thinking", "redacted_thinking") else f"block:{index}" + block_type: Final = block.get("type") + if block_type in ("thinking", "redacted_thinking"): + return "thinking" + return "text" if block_type == "text" else f"block:{index}" @classmethod def _assistant_group_to_input_items( @@ -285,6 +309,14 @@ class LiteLLMAnthropicToResponsesAPIAdapter: if btype in ("thinking", "redacted_thinking"): replayed: Final = responses_reasoning_items_from_thinking_blocks(group) return tuple(dict(item) for item in replayed) + if btype == "text": + return ( + { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": block.get("text", "")} for block in group], + }, + ) if btype == "tool_use": return ( { @@ -451,19 +483,6 @@ class LiteLLMAnthropicToResponsesAPIAdapter: for _, group in groupby(enumerate(blocks), key=self._assistant_block_group_key) for item in self._assistant_group_to_input_items(tuple(block for _, block in group)) ) - asst_parts: list[dict[str, object]] = [ # mutable-ok: API message payload - {"type": "output_text", "text": block.get("text", "")} - for block in blocks - if block.get("type") == "text" - ] - if asst_parts: - input_items.append( - { - "type": "message", - "role": "assistant", - "content": asst_parts, - } - ) return input_items @@ -735,7 +754,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter: for item in response.output: if isinstance(item, ResponseReasoningItem): - reasoning_block = self._thinking_block_from_reasoning_item(item.summary, item.encrypted_content) + reasoning_block = self.thinking_block_from_reasoning_item(item) if reasoning_block is not None: content.append(reasoning_block) @@ -787,10 +806,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter: ).model_dump() ) elif item_type == "reasoning": - reasoning_block = self._thinking_block_from_reasoning_item( - cast(Iterable[object], item.get("summary") or ()), # cast-ok: untyped provider json - item.get("encrypted_content"), - ) + reasoning_block = self.thinking_block_from_reasoning_item(item) if reasoning_block is not None: content.append(reasoning_block) elif item_type == "function_call": diff --git a/litellm/types/llms/openai.py b/litellm/types/llms/openai.py index 8cc9e912e80..f71642f1082 100644 --- a/litellm/types/llms/openai.py +++ b/litellm/types/llms/openai.py @@ -703,13 +703,19 @@ class ChatCompletionReasoningSummaryTextBlock(TypedDict, total=False): text: str +class ChatCompletionReasoningTextBlock(TypedDict): + type: ReadOnly[Literal["reasoning_text"]] + text: ReadOnly[str] + + class ChatCompletionReasoningItem(TypedDict, total=False): """Represents an OpenAI Responses API reasoning item for round-tripping in conversation history.""" - type: Required[Literal["reasoning"]] - id: str - encrypted_content: str | None + type: ReadOnly[Required[Literal["reasoning"]]] + id: ReadOnly[str] + encrypted_content: ReadOnly[str | None] summary: ReadOnly[list[ChatCompletionReasoningSummaryTextBlock]] + content: ReadOnly[list[ChatCompletionReasoningTextBlock]] class WebSearchOptionsUserLocationApproximate(TypedDict, total=False): diff --git a/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_common_utils.py b/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_common_utils.py index c08eeea4bb9..565b70a7452 100644 --- a/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_common_utils.py +++ b/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_common_utils.py @@ -28,6 +28,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import ( merge_consecutive_system_messages, native_redacted_thinking_data, parse_tool_call_arguments, + reasoning_content_signature, responses_reasoning_items_from_thinking_blocks, salvage_concatenated_tool_arguments, schema_grammar_branch_count, @@ -1970,6 +1971,35 @@ class TestEncryptedReasoningReplay: def test_signature_round_trips_the_encrypted_content(self): assert encrypted_content_from_signature(encrypted_reasoning_signature("gAAAA_bytes")) == "gAAAA_bytes" + def test_reasoning_content_signatures_keep_consecutive_items_separate(self) -> None: + first_signature: Final = reasoning_content_signature(("Brief summary",), ("First thought",), "opaque") + second_signature: Final = reasoning_content_signature((), ("Second thought",), None) + + assert encrypted_content_from_signature(first_signature) == "opaque" + assert responses_reasoning_items_from_thinking_blocks( + ( + {"type": "thinking", "thinking": "First thought", "signature": first_signature}, + {"type": "thinking", "thinking": "Second thought", "signature": second_signature}, + ) + ) == ( + { + "type": "reasoning", + "summary": [{"type": "summary_text", "text": "Brief summary"}], + "content": [{"type": "reasoning_text", "text": "First thought"}], + "encrypted_content": "opaque", + }, + {"type": "reasoning", "summary": [], "content": [{"type": "reasoning_text", "text": "Second thought"}]}, + ) + + @pytest.mark.parametrize("payload", ["not-json", '{"summary":[],"content":[1]}']) + def test_malformed_reasoning_content_signature_keeps_readable_thinking(self, payload: str) -> None: + signature: Final = "litellm_reasoning_content:" + payload + + assert encrypted_content_from_signature(signature) is None + assert responses_reasoning_items_from_thinking_blocks( + ({"type": "thinking", "thinking": "Keep this thought", "signature": signature},) + ) == ({"type": "reasoning", "summary": [{"type": "summary_text", "text": "Keep this thought"}]},) + @pytest.mark.parametrize( "signature", [None, "", "ErcBCkgIValidAnthropicSignature", "litellm_encrypted_reasoning:", 7] ) diff --git a/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py b/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py index e1dded214bf..e5c19a2d919 100644 --- a/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py +++ b/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py @@ -8,8 +8,11 @@ import json import os import sys from types import SimpleNamespace +from typing import Final, Literal import pytest +from anthropic.types import RawContentBlockDeltaEvent, RawMessageStreamEvent, SignatureDelta, ThinkingDelta +from pydantic import TypeAdapter sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../.."))) @@ -25,6 +28,69 @@ from litellm.llms.anthropic.pass_through.responses_adapters.streaming_iterator i from litellm.types.llms.openai import ResponseFailedEvent, ResponsesAPIResponse +@pytest.mark.asyncio +@pytest.mark.parametrize("delivery", ["delta", "done", "item"]) +@pytest.mark.parametrize("encrypted_content", [None, "opaque-reasoning"]) +async def test_reasoning_text_reaches_client_and_survives_replay( + delivery: Literal["delta", "done", "item"], encrypted_content: str | None +) -> None: + from litellm.llms.anthropic.pass_through.responses_adapters.transformation import ( + LiteLLMAnthropicToResponsesAPIAdapter, + ) + from litellm.types.llms.anthropic import AnthropicMessagesRequest + + reasoning: Final = { + "type": "reasoning", + "id": "rs_text", + "summary": [], + "content": [{"type": "reasoning_text", "text": "Use a tool"}], + **({"encrypted_content": encrypted_content} if encrypted_content is not None else {}), + } + events: Final = ( + {"type": "response.created"}, + {"type": "response.output_item.added", "item": {"type": "reasoning", "id": "rs_text"}}, + *( + ( + {"type": "response.reasoning_text.delta", "item_id": "rs_text", "content_index": 0, "delta": "Use "}, + {"type": "response.reasoning_text.delta", "item_id": "rs_text", "content_index": 0, "delta": "a tool"}, + ) + if delivery == "delta" + else () + ), + *( + ({"type": "response.reasoning_text.done", "item_id": "rs_text", "content_index": 0, "text": "Use a tool"},) + if delivery != "item" + else () + ), + {"type": "response.output_item.done", "item": reasoning}, + {"type": "response.completed"}, + ) + wrapper: Final = AnthropicResponsesStreamWrapper(responses_stream=iter(events), model="test-model") + chunks: Final = TypeAdapter(tuple[RawMessageStreamEvent, ...]).validate_python(tuple([c async for c in wrapper])) + thinking: Final = "".join( + event.delta.thinking + for event in chunks + if isinstance(event, RawContentBlockDeltaEvent) and isinstance(event.delta, ThinkingDelta) + ) + signature: Final = "".join( + event.delta.signature + for event in chunks + if isinstance(event, RawContentBlockDeltaEvent) and isinstance(event.delta, SignatureDelta) + ) + request: Final = AnthropicMessagesRequest( + model="test-model", + max_tokens=1024, + messages=[ + {"role": "assistant", "content": [{"type": "thinking", "thinking": thinking, "signature": signature}]} + ], + ) + + assert thinking == "Use a tool" + assert LiteLLMAnthropicToResponsesAPIAdapter().translate_messages_to_responses_input(request["messages"]) == [ + {key: value for key, value in reasoning.items() if key != "id"} + ] + + def _process_all(events: list) -> list: wrapper = AnthropicResponsesStreamWrapper(responses_stream=None, model="m") for event in events: diff --git a/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_transformation.py b/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_transformation.py index 829000c0c5f..ef24945c3f1 100644 --- a/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_transformation.py +++ b/tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_transformation.py @@ -33,6 +33,85 @@ from litellm.types.llms.anthropic import ( from litellm.types.llms.openai import ResponseAPIUsage +def test_parallel_tool_round_keeps_assistant_block_order() -> None: + request: Final = AnthropicMessagesRequest( + model="test-model", + max_tokens=1024, + messages=[ + {"role": "user", "content": "Run both tools"}, + { + "role": "assistant", + "content": [ + {"type": "thinking", "thinking": "Use both tools"}, + {"type": "text", "text": "Working"}, + {"type": "tool_use", "id": "call_a", "name": "noop", "input": {}}, + {"type": "tool_use", "id": "call_b", "name": "noop", "input": {}}, + ], + }, + { + "role": "user", + "content": [ + {"type": "tool_result", "tool_use_id": "call_a", "content": "done a"}, + {"type": "tool_result", "tool_use_id": "call_b", "content": "done b"}, + ], + }, + ], + ) + + assert _ADAPTER.translate_messages_to_responses_input(request["messages"]) == [ + {"type": "message", "role": "user", "content": [{"type": "input_text", "text": "Run both tools"}]}, + {"type": "reasoning", "summary": [{"type": "summary_text", "text": "Use both tools"}]}, + {"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "Working"}]}, + {"type": "function_call", "call_id": "call_a", "name": "noop", "arguments": "{}"}, + {"type": "function_call", "call_id": "call_b", "name": "noop", "arguments": "{}"}, + {"type": "function_call_output", "call_id": "call_a", "output": "done a"}, + {"type": "function_call_output", "call_id": "call_b", "output": "done b"}, + ] + + +@pytest.mark.parametrize("encrypted_content", [None, "opaque-reasoning"]) +@pytest.mark.parametrize("with_summary", [False, True]) +@pytest.mark.parametrize("as_dict", [False, True]) +def test_reasoning_content_survives_response_and_replay( + encrypted_content: str | None, with_summary: bool, as_dict: bool +) -> None: + from openai.types.responses import ResponseReasoningItem + + from litellm.types.llms.openai import ResponsesAPIResponse + + reasoning: Final = ResponseReasoningItem( + id="rs_content", + type="reasoning", + summary=[{"type": "summary_text", "text": "Brief summary"}] if with_summary else [], + content=[ + {"type": "reasoning_text", "text": "First thought"}, + {"type": "reasoning_text", "text": "Second thought"}, + ], + encrypted_content=encrypted_content, + ) + response: Final = ResponsesAPIResponse( + id="resp_content", + created_at=1, + model="test-model", + object="response", + status="completed", + output=[reasoning.model_dump(exclude_none=True) if as_dict else reasoning], + parallel_tool_calls=True, + tool_choice="auto", + tools=[], + ) + translated: Final = _ADAPTER.translate_response(response) + request: Final = AnthropicMessagesRequest( + model="test-model", max_tokens=1024, messages=[{"role": "assistant", "content": translated["content"]}] + ) + + assert translated["content"][0]["type"] == "thinking" + assert translated["content"][0]["thinking"] == "First thought\n\nSecond thought" + assert _ADAPTER.translate_messages_to_responses_input(request["messages"]) == [ + {key: value for key, value in reasoning.model_dump(exclude_none=True).items() if key != "id"} + ] + + def _make_request(**overrides) -> AnthropicMessagesRequest: base: dict = { "model": "openai.gpt-5.1-codex",